Capítulo 4 - Fuentes de Datos y Persistencia (CQRS)

El sistema utiliza dos fuentes de datos principales con estrategia CQRS (Command Query Responsibility Segregation), separando las operaciones de lectura y escritura para optimizar el rendimiento.

Principio CQRS: Separar los modelos de lectura (queries) y escritura (commands) permite optimizar cada uno para su propósito específico, mejorando significativamente el rendimiento del sistema.

4.1.1 Base de Datos Core (Escritura)

Propósito: Operaciones transaccionales (CREATE, UPDATE, DELETE)
Tecnología: PostgreSQL 15+
Contenido: Movimientos financieros, productos, transacciones
Características: ACID completo, integridad referencial, transacciones complejas

4.1.2 Base de Datos de Lectura (Read Model)

Propósito: Consultas optimizadas (SELECT)
Tecnología: MongoDB 7.0+ (NoSQL denormalizada)
Contenido: Vistas materializadas, históricos denormalizados
Características: Sin joins, queries ultra-rápidas, agregaciones eficientes

4.1.3 Base de Datos de Detalles de Cliente

Propósito: Información complementaria del cliente
Tecnología: PostgreSQL 15+
Contenido: Datos personales, preferencias, configuraciones

4.1.4 Base de Datos de Auditoría

Propósito: Compliance, logs, trazabilidad
Tecnología: Elasticsearch 8.11+
Contenido: Todos los eventos del sistema, búsqueda full-text

4.2 Diagrama de Arquitectura CQRS

Separación de Lecturas y Escrituras

flowchart TD
    %% Estilos globales
    classDef capa fill:#cfe8fc,stroke:#4682b4,stroke-width:2px,color:#000,border-radius:10px;
    classDef dbwrite fill:#d5f5e3,stroke:#2e8b57,stroke-width:2px,color:#000,border-radius:10px;
    classDef dbread fill:#fce5cd,stroke:#d2691e,stroke-width:2px,color:#000,border-radius:10px;
    classDef kafka fill:#f6d8f8,stroke:#8b008b,stroke-width:2px,color:#000,border-radius:10px;
    classDef detalle fill:#ffffff,stroke:#888,stroke-width:1.5px,color:#000,border-radius:8px;

    %% Nodos principales
    A["CAPA DE APLICACION

        Microservicios GraphQL"]
    A -->|WRITE - Mutations| B["PostgreSQL WRITE DB

                                Transaccional"]
    A -->|READ - Queries| C["MongoDB READ DB

                             Optimizada"]

    %% Detalles PostgreSQL
    B --> B1["Movimientos"]
    B --> B2["Productos"]
    B --> B3["Transacciones"]
    B --> B4["ACID completo"]

    %% Detalles MongoDB
    C --> C1["Vista Histórico"]
    C --> C2["Agregaciones"]
    C --> C3["Búsquedas"]
    C --> C4["Denormalizado / Sin Joins"]

    %% Kafka Event Bus
    B -->|Event Emitted| D["KAFKA EVENT BUS

                            Topic: data.sync

                            Latencia: 100–500 ms"]
    D -->|Sync - Consistencia Eventual| C

    %% Asignar estilos
    class A capa;
    class B dbwrite;
    class C dbread;
    class D kafka;
    class B1,B2,B3,B4,C1,C2,C3,C4 detalle;

4.3 Consistencia Eventual

Justificación de Consistencia Eventual

La consistencia eventual es apropiada para este sistema porque:

Mayor disponibilidad del sistema (Teorema CAP: elegimos AP sobre C)
Escalabilidad horizontal sin bloqueos entre servicios
Mejor performance en lecturas (sin joins complejos)
Resiliencia ante fallos de red entre servicios
Reducción de contención en la base de datos de escritura

4.3.1 Implementación de la Sincronización

Escritura en PostgreSQL: Las escrituras se realizan primero en PostgreSQL con garantías ACID completas
Emisión de evento: Se emite un evento a Kafka tras cada escritura exitosa
Consumidor escucha: Un consumidor dedicado escucha el topic data.sync
Actualización de lectura: El consumidor actualiza la base de lectura (MongoDB)
Latencia típica: 100-500ms de sincronización

4.3.2 Casos de Uso Apropiados

Tipo de Consulta	Base de Datos	Justificación
Histórico de movimientos	MongoDB (Read)	Consistencia eventual aceptable (delay de segundos OK)
Dashboards y reportes	MongoDB (Read)	Datos agregados, no requieren tiempo real
Saldo actual de cuenta	PostgreSQL (Write)	Requiere consistencia fuerte, lectura directa
Validación de transferencia	PostgreSQL (Write)	Operación crítica, requiere ACID
Búsqueda de transacciones	MongoDB (Read)	Optimizada para búsquedas complejas

Resultado: 70% de las consultas van a MongoDB (rápidas, sin carga en PostgreSQL), 30% a PostgreSQL (operaciones críticas con consistencia fuerte).

4.4 Flujo de Sincronización Detallado

Flujo Completo de Sincronización CQRS

flowchart TD
    %% Estilos
    classDef paso fill:#d6eaf8,stroke:#2471a3,stroke-width:2px,color:#000,border-radius:10px;
    classDef dbwrite fill:#d5f5e3,stroke:#239b56,stroke-width:2px,color:#000,border-radius:10px;
    classDef kafka fill:#f6d8f8,stroke:#8b008b,stroke-width:2px,color:#000,border-radius:10px;
    classDef worker fill:#fdebd0,stroke:#d68910,stroke-width:2px,color:#000,border-radius:10px;
    classDef dbread fill:#fcf3cf,stroke:#b7950b,stroke-width:2px,color:#000,border-radius:10px;
    classDef resultado fill:#e8daef,stroke:#6c3483,stroke-width:2px,color:#000,border-radius:10px;

    %% PASO 1
    A["🟢 PASO 1:
Usuario realiza transferencia"]
    A --> B["GraphQL Mutation → Microservicio de Pagos"]

    %% PASO 2
    B --> C["PASO 2:
Escritura en PostgreSQL (Write DB)"]
    C --> C1["BEGIN TRANSACTION"]
    C --> C2["INSERT INTO transactions(...)"]
    C --> C3["UPDATE accounts SET balance = balance - amount"]
    C --> C4["COMMIT TRANSACTION ✓"]
    C --> C5["Transacción exitosa
(ACID garantizado)"]

    %% PASO 3
    C --> D["PASO 3:
Emisión de Evento a Kafka"]
    D --> D1["Event:
{ type: 'TRANSACTION_CREATED',
transactionId: 'txn-12345',
accountId: 'acc-67890',
amount: 1000,
timestamp: '2025-10-06T10:30:00Z' }"]
    D --> D2["Publicado en topic: data.sync"]

    %% PASO 4
    D --> E["PASO 4:
Consumidor procesa evento"]
    E --> E1["Worker Thread procesa en paralelo"]
    E --> E2["Enriquece datos:
+ info cliente, + categoría"]
    E --> E3["Latencia: 100–300 ms"]

    %% PASO 5
    E --> F["PASO 5:
Actualización en MongoDB (Read DB)"]
    F --> F1["db.transactions.insertOne({
transactionId:'txn-12345',
accountId:'acc-67890',
amount:1000,
customerName:'Juan Pérez',
category:'Transferencia',
timestamp:'2025-10-06T10:30:00Z'
})"]
    F --> F2["Actualización denormalizada
(sin joins)"]

    %% PASO 6
    F --> G["PASO 6:
Dato disponible para consultas rápidas"]
    G --> G1["Consultas desde MongoDB: < 20ms
(vs 150–200ms si fuera join en PostgreSQL)"]

    %% Asignación de clases
    class A,B paso;
    class C,C1,C2,C3,C4,C5 dbwrite;
    class D,D1,D2 kafka;
    class E,E1,E2,E3 worker;
    class F,F1,F2 dbread;
    class G,G1 resultado;

4.5 Persistencia para Clientes Frecuentes - Caché con Redis

Para optimizar aún más el rendimiento, se implementa una capa de caché con Redis para datos de clientes frecuentes.

Estrategia de Caché Multi-Nivel

flowchart TD
    %% ==== Estilos ====
    classDef ms fill:#d6eaf8,stroke:#2471a3,stroke-width:2px,color:#000,border-radius:10px;
    classDef cache fill:#fdebd0,stroke:#d68910,stroke-width:2px,color:#000,border-radius:10px;
    classDef db fill:#d5f5e3,stroke:#1e8449,stroke-width:2px,color:#000,border-radius:10px;

    %% ==== Microservicio ====
    A["MS-Datos-Cliente"]
    class A ms

    %% ==== Conexiones ====
    A --> B["Redis
(Cache)

TTL: 1 hora
LRU eviction
Hit rate: 70%"]
    A --> C["PostgreSQL
(Fuente)

Datos Cliente
Authoritative"]

    %% ==== Clases ====
    class B cache
    class C db

Flujo de Lectura:

1. Request → Redis (caché)
2. Cache HIT (70%) → Respuesta inmediata (2ms)
3. Cache MISS (30%) → PostgreSQL (50ms) + Guardar en Redis

4.5.1 Estrategia de Caché

Datos cacheados: Información de clientes frecuentes (regla 80/20: Top 20% genera 80% del tráfico)
TTL (Time To Live): 1 hora para balance entre frescura y performance
Política de evicción: LRU (Least Recently Used)
Invalidación proactiva: Cuando se actualiza el cliente, se invalida el caché
Hit rate objetivo: >70%

4.5.2 Beneficios Medibles

Métrica	Sin Caché	Con Redis	Mejora
Latencia promedio	50ms	15ms (70% hits a 2ms)	70%
Queries a PostgreSQL	10,000/min	3,000/min	70%
Carga CPU PostgreSQL	60%	20%	67%
Throughput (req/seg)	500	1,500	200%

4.6 Modelo de Datos

4.6.1 Esquema PostgreSQL (Write DB)

-- Tabla de Cuentas
CREATE TABLE accounts (
    id UUID PRIMARY KEY,
    customer_id UUID NOT NULL,
    account_number VARCHAR(20) UNIQUE NOT NULL,
    balance DECIMAL(15,2) NOT NULL DEFAULT 0,
    account_type VARCHAR(20) NOT NULL,
    status VARCHAR(20) NOT NULL,
    created_at TIMESTAMP NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMP NOT NULL DEFAULT NOW()
);

-- Tabla de Transacciones (Write Model)
CREATE TABLE transactions (
    id UUID PRIMARY KEY,
    account_id UUID NOT NULL REFERENCES accounts(id),
    transaction_type VARCHAR(50) NOT NULL,
    amount DECIMAL(15,2) NOT NULL,
    balance_after DECIMAL(15,2) NOT NULL,
    description TEXT,
    status VARCHAR(20) NOT NULL,
    created_at TIMESTAMP NOT NULL DEFAULT NOW(),
    idempotency_key VARCHAR(100) UNIQUE
);

-- Índices optimizados para escritura
CREATE INDEX idx_transactions_account ON transactions(account_id);
CREATE INDEX idx_transactions_created ON transactions(created_at DESC);

4.6.2 Esquema MongoDB (Read DB)

// Colección de Transacciones (Read Model - Denormalizado)
{
  "_id": "txn-12345",
  "transactionId": "txn-12345",
  "accountId": "acc-67890",
  "accountNumber": "1234567890",
  
  // Datos denormalizados del cliente
  "customer": {
    "id": "cust-111",
    "name": "Juan Pérez",
    "email": "juan@example.com"
  },
  
  // Datos de la transacción
  "type": "TRANSFER_OUT",
  "amount": 1000.00,
  "balanceAfter": 48000.00,
  "description": "Transferencia a cuenta externa",
  "category": "Transferencia",
  "status": "COMPLETED",
  
  // Metadata enriquecida
  "timestamp": ISODate("2025-10-06T10:30:00Z"),
  "tags": ["transfer", "external", "high-amount"],
  
  // Índices para búsquedas rápidas
  "searchableText": "transferencia juan perez 1000"
}

// Índices optimizados para lectura
db.transactions.createIndex({ "accountId": 1, "timestamp": -1 })
db.transactions.createIndex({ "customer.id": 1 })
db.transactions.createIndex({ "searchableText": "text" })

4.7 Gestión de Réplicas y Alta Disponibilidad

4.7.1 PostgreSQL - Configuración Master-Replica

Arquitectura de Replicación PostgreSQL

flowchart TD
    A["Primary
(Read+Write)"] -- Streaming Replication
(Asíncrona) --> R1["Replica 1
(Read-only)
Zona AZ-A"]
    A --> R2["Replica 2
(Read-only)
Zona AZ-B"] & R3["Replica 3
(Read-only)
Zona AZ-C"]

    A@{ shape: rounded}
    R1@{ shape: rounded}
    R2@{ shape: rounded}
    R3@{ shape: rounded}
     A:::pastel
     A:::Peach
     A:::Pine
     R1:::pastel
     R1:::Ash
     R1:::Peach
     R2:::pastel
     R2:::Peach
     R3:::pastel
     R3:::Peach
    classDef pastel fill:#ffe7ba,stroke:#a88039,stroke-width:2px,color:#463918
    classDef Pine stroke-width:1px, stroke-dasharray:none, stroke:#254336, fill:#27654A, color:#FFFFFF
    classDef Ash stroke-width:1px, stroke-dasharray:none, stroke:#999999, fill:#EEEEEE, color:#000000
    classDef Peach stroke-width:1px, stroke-dasharray:none, stroke:#FBB35A, fill:#FFEFDB, color:#8F632D

Replicación asíncrona: Balance entre performance y durabilidad
Failover automático: Patroni o Stolon para promoción automática
Backup automático: Diario + PITR (Point-in-Time Recovery)
Multi-AZ: Réplicas en diferentes zonas de disponibilidad

4.7.2 MongoDB - Replica Set

Replica Set de 3 nodos: 1 Primary + 2 Secondary
Read Preference: Configurado para leer de secundarios (reduce carga en primary)
Write Concern: majority (espera confirmación de mayoría)
Elección automática: Si primary cae, se elige nuevo primary automáticamente

4.8 Estrategia de Backup y Recuperación

Tipo de Backup	Frecuencia	Retención	RTO	RPO
Full Backup PostgreSQL	Semanal (Domingo 2am)	3 meses	30 min	7 días
Incremental PostgreSQL	Diario (2am)	30 días	20 min	1 día
WAL Logs PostgreSQL	Continuo	7 días	15 min	5 min
MongoDB Snapshot	Cada 6 horas	48 horas	10 min	6 horas
Redis Persistence (RDB)	Cada hora	24 horas	5 min	1 hora

RTO (Recovery Time Objective): Tiempo máximo aceptable para recuperar el servicio.
RPO (Recovery Point Objective): Cantidad máxima aceptable de datos que se pueden perder.

4.9 Ventajas del Patrón CQRS Implementado

✓ Beneficios Comprobados

Performance de Lectura: Queries 5-10x más rápidas (sin joins, datos denormalizados)
Escalabilidad Independiente: Escalar lectura y escritura por separado según necesidad
Optimización Específica: PostgreSQL para transacciones ACID, MongoDB para consultas complejas
Menor Contención: Lecturas no bloquean escrituras
Flexibilidad de Schema: Read model puede tener múltiples vistas sin afectar write model
Reducción de Carga: 70% de queries van a MongoDB, liberando PostgreSQL para transacciones

⚠️ Consideraciones Importantes

Complejidad: Mayor complejidad operacional (2 bases de datos a mantener)
Consistencia Eventual: Los clientes deben entender delays de 100-500ms
Sincronización: Kafka debe estar siempre disponible
Debugging: Más difícil debuggear problemas de sincronización