EDA con NATS: una sola fuente de verdad, múltiples modelos de lectura — Proyecciones CQRS

CQRS orientado a eventos no significa “muchas bases de datos en desacuerdo”. Se trata de un modelo de escritura autoritativo que se proyecta en múltiples modelos de lectura diseñados para propósitos específicos. NATS es la infraestructura de conexión —subjects, buckets KV y señales de indexación con fan-in—, no el sistema de registro.

Este artículo presenta un ejemplo genérico de procesamiento de pedidos de la serie Arquitectura orientada a eventos con NATS: una base de datos relacional como fuente de verdad, un espejo en NATS KV, un indexador que recibe señales mediante fan-in y las proyecta en OpenSearch, y un warehouse de analítica para reportes. Los patrones se aplican a catálogos, usuarios, inventarios o cualquier entidad cuyas necesidades de escritura y lectura sean distintas.

Investigación basada en los escritos de Martin Fowler sobre CQRS, la replicación lógica de PostgreSQL, la documentación de NATS y las API de indexación de OpenSearch el July 6, 2026.

TL;DR

• CQRS separa los modelos de escritura de los modelos de lectura cuando la estructura de las consultas, la escala o los patrones de acceso difieren de los de escritura.
• La base de datos primaria (en este caso, PostgreSQL) es la fuente de verdad. NATS KV, OpenSearch y el warehouse de analítica son proyecciones: copias optimizadas para responder preguntas distintas.
• CDC (captura de cambios de la base de datos de escritura → espejo KV) alimenta una ruta de lectura por clave de baja latencia sin convertir a KV en la fuente autoritativa.
• El fan-in del indexador (index.<type> → un indexador → OpenSearch) centraliza la lógica de la proyección de búsqueda.
• La consistencia eventual es una decisión arquitectónica explícita: la búsqueda puede tener un retraso de segundos; las vistas de detalle deben leer de la fuente de verdad o del espejo KV, según la política definida.
• Esto es CQRS basado en CDC, no event sourcing completo. El registro de cambios proviene del stream de replicación de la base de datos, junto con subjects de indexación opcionales, no de un almacén de eventos de dominio de solo adición.

Qué aprenderás aquí

• Modelo mental: fuente de verdad, infraestructura de conexión y proyecciones
• Ruta de escritura de pedidos (API → PostgreSQL)
• Espejo CDC: base de datos → NATS KV (app-objects)
• Fan-in del indexador: index.<type> → OpenSearch
• Warehouse de analítica como proyección para reportes
• Expectativas de consistencia y retraso para cada ruta de lectura
• Implementación de un indexador idempotente
• Cómo se relaciona esto con las partes 1–3

Para conocer la progresión arquitectónica que motiva los patrones CQRS y outbox, consulta Las piezas del diseño de software moderno y eficaz. Ese artículo usa un outbox transaccional; este usa CDC de una base de datos de escritura existente. Ambos son mecanismos válidos para activar proyecciones.

Modelo mental: fuente de verdad, infraestructura de conexión y proyecciones

Ten presentes dos reglas:

1. Un bucket KV es un stream + una vista del valor más reciente. Los tipos de entidad son prefijos de clave, no buckets. (Consulta la Parte 3.)
2. NATS es infraestructura de conexión. La base de datos de escritura determina qué ocurrió. OpenSearch y el warehouse responden preguntas distintas.

  TRUTH                    PLUMBING                  PROJECTIONS
  ─────                    ────────                  ───────────
  PostgreSQL        ──►    NATS subjects      ──►     OpenSearch (search)
  (orders table)            NATS KV mirror     ──►     In-memory caches
       │                   index.*            ──►     Warehouse (analytics)
       │
       └── source of record — everything else is a copy

Las escrituras pasan por la API hacia PostgreSQL. Todos los demás almacenes son proyecciones optimizadas para una estructura de consulta específica.

Glosario

Ruta de escritura de pedidos

  [Client / API]
        |
        |  POST /orders, PATCH /orders/{id}
        v
  +------------------+
  |  Order API       |  <<<< WRITE MODEL (system of record)
  |  (write service) |
  +--------+---------+
           |
           |  INSERT / UPDATE / DELETE
           v
  +------------------+
  |  PostgreSQL      |  authoritative order state
  |  orders table    |
  +------------------+

La ruta de escritura es responsable de la validación, la autorización y las invariantes. Las lecturas de confirmación posteriores a una mutación (crear, actualizar, cancelar) deben consultar la Order API o PostgreSQL, no OpenSearch.

Espejo CDC: base de datos → NATS KV

La captura de cambios de datos (CDC) lee el registro de cambios de la base de datos y lo proyecta en otro destino. Con PostgreSQL, la replicación lógica o herramientas como Debezium consumen inserciones, actualizaciones y eliminaciones a nivel de fila desde el registro de escritura anticipada.

Un worker de espejo CDC consume esos cambios y los inserta o actualiza en NATS KV:

  PostgreSQL orders table
       |
       |  CDC stream (INSERT, UPDATE, DELETE)
       v
  +------------------+
  |  CDC Mirror      |
  |  Worker          |
  +--------+---------+
           |
           |  kv.Put / kv.Delete
           v
  +-------------------+
  | NATS JetStream KV |
  | bucket: app-objects|
  |                   |
  | key:              |
  | orders-v1.{orderId}|
  +-------------------+

import { connect, StringCodec } from "nats";

const nc = await connect({ servers: process.env.NATS_URL });
const js = nc.jetstream();
const kv = await js.views.kv("app-objects");
const sc = StringCodec();

type ChangeEvent = {
  op: "create" | "update" | "delete";
  table: string;
  key: { id: string };
  after?: Record<string, unknown>;
};

async function handleChange(event: ChangeEvent) {
  if (event.table !== "orders") return;

  const orderId = event.key.id;
  const key = `orders-v1.${orderId}`;

  if (event.op === "delete") {
    await kv.delete(key);
    return;
  }

  await kv.put(key, sc.encode(JSON.stringify(event.after)));
}

¿Por qué crear un espejo en KV?

• Inferencia editorial: Los servicios que necesitan blobs de pedidos por ID pueden leerlos de KV sin consultar PostgreSQL en cada solicitud, lo que reduce la carga de lectura en la ruta crítica.
• KV no es la fuente de verdad. Si KV y PostgreSQL discrepan, prevalece la base de datos. Implementa mecanismos de detección de discrepancias y de reconstrucción.

Relación con la Parte 3: La clave orders-v1.{id} vive en el bucket app-objects (stream KV_app-objects). Los servicios posteriores pueden usar watch("orders-v1.>") para recibir actualizaciones reactivas.

CDC vs. outbox

Ambos llegan al mismo resultado: una señal de cambio confiable que consumen los proyectores.

Advertencia sobre CDC: Es necesario monitorear los slots de replicación y el retraso de los consumidores. Si el worker de CDC se retrasa más allá de los límites de retención, necesitas una vía de reconstrucción a partir de una instantánea o exportación de la base de datos. La retención exacta depende de tus herramientas de CDC y de la configuración de la base de datos.

Fan-in del indexador: index.<type> → OpenSearch

La búsqueda necesita documentos desnormalizados, analizadores y agregaciones: una estructura diferente de las filas normalizadas de una tabla. En lugar de N proyectores independientes, muchos servicios publican señales de indexación:

  Order API / CDC   ──►  index.order
  Customer service  ──►  index.customer
  Catalog service   ──►  index.product
                              │
                              │  all subjects match index.>
                              v
                    +------------------+
                    |  Indexer         |
                    |  (single worker  |
                    |   or queue grp)  |
                    +--------+---------+
                             |
                             |  idempotent upsert
                             v
                    +------------------+
                    |  OpenSearch      |
                    |  orders index    |
                    +------------------+

¿Por qué usar fan-in?

• Un solo lugar donde gestionar los mappings de OpenSearch, la configuración de operaciones masivas y el manejo de errores
• Cada servicio de dominio publica una señal pequeña, no documentos completos de búsqueda
• La suscripción con comodín index.> (subjects de la Parte 1) captura todos los tipos de entidad

Relación con la Parte 2: El indexador debe usar un consumidor durable de JetStream con ack explícito, MaxDeliver finito y handlers idempotentes, no pub/sub de Core NATS.

Proyección en un warehouse de analítica

OpenSearch responde «buscar los pedidos del cliente X realizados esta semana». Un warehouse responde «¿cuál fue el ingreso total por región durante el último trimestre?».

El warehouse es un modelo de lectura para analítica, alimentado por el mismo linaje de cambios (exportación CDC, carga desde un stream o ETL periódico). No forma parte de la ruta de lectura en línea. Un retraso de minutos a horas suele ser aceptable.

Snowflake, BigQuery o cualquier warehouse columnar cumple esta función. Las herramientas de ingesta dependen del stack; el patrón no.

Expectativas de consistencia y retraso

Usa estos presupuestos de retraso como insumos para los SLO: conecta las vistas de detalle y las confirmaciones de mutaciones a la ruta de escritura; trata la búsqueda y la analítica como proyecciones eventualmente consistentes.

Implementación concreta: indexador idempotente

import { connect, StringCodec } from "nats";

const nc = await connect({ servers: process.env.NATS_URL });
const js = nc.jetstream();
const sc = StringCodec();

type IndexSignal = {
  id: string;          // dedupe key
  entityType: string;  // "order"
  entityId: string;
  op: "upsert" | "delete";
  version: number;
};

const sub = await js.pullSubscribe("index.>", {
  config: { durable_name: "search-indexer", ack_policy: "explicit" },
  stream: "INDEX_SIGNALS",
});

for (;;) {
  const msgs = await sub.fetch(20, { expires: 5000 });
  for (const msg of msgs) {
    const signal: IndexSignal = JSON.parse(sc.decode(msg.data));

    if (await alreadyProcessed(signal.id)) {
      msg.ack();
      continue;
    }

    try {
      if (signal.op === "delete") {
        await openSearch.delete({
          index: "orders",
          id: signal.entityId,
        });
      } else {
        const doc = await buildSearchDocument(signal.entityType, signal.entityId);
        await openSearch.index({
          index: "orders",
          id: signal.entityId,
          body: doc,
        });
      }
      await markProcessed(signal.id);
      msg.ack();
    } catch (err) {
      msg.nak(); // JetStream redelivers per AckWait / MaxDeliver
    }
  }
}

buildSearchDocument obtiene el estado actual de KV o de la Order API. El indexador construye el documento de búsqueda a partir de la fuente de verdad o del espejo, no solo del payload de la señal. Esto mantiene pequeñas las señales y actualizados los documentos.

Modos de falla y replay

Métricas operativas

• projection_lag_seconds — tiempo desde la escritura en la base de datos hasta que el cambio es visible en OpenSearch
• kv_mirror_age_seconds — retraso del worker de CDC
• cdc_slot_lag_bytes — estado del consumidor de replicación (PostgreSQL)
• indexer_dlq_depth — mensajes que alcanzaron MaxDeliver
• opensearch_bulk_failures — estado del indexador

Cómo se relaciona esto con las capas 1–3

                         ONE TRUTH → MANY READ MODELS
                         =================================

  Order API ──► PostgreSQL (truth)
                    │
                    ├── CDC ──► KV app-objects / orders-v1.*
                    │              │
                    │              └── watch ──► cache warmers
                    │
                    ├── index.order ──┐
                    │   index.*       ├──► Indexer ──► OpenSearch
                    │                 │
                    └── ETL ─────────────────────────────► Warehouse

Cuándo NO agregar otra proyección

Aplica la recomendación de Fowler sobre CQRS: usa modelos de lectura cuando el lado de lectura realmente haya superado las capacidades del lado de escritura. No agregues KV + OpenSearch + un warehouse desde el primer día solo porque el diagrama se vea sofisticado.

A continuación en la serie: Observabilidad y SLO (Parte 5) · NATS vs. Kafka vs. SQS (Parte 6)

Fuentes

• CQRS — Martin Fowler — cuándo separar los modelos de lectura y escritura
• ¿Qué quieres decir con “orientado a eventos”? — Martin Fowler — notificación de eventos vs. transferencia de estado vs. event sourcing
• Replicación lógica de PostgreSQL — fundamento de CDC para modelos de escritura relacionales
• Subjects de NATS — fan-in con el comodín index.>
• Almacén clave/valor de NATS — modelo de consistencia del espejo KV
• NATS JetStream — consumidores durables para indexadores
• API de indexación de documentos de OpenSearch — inserción o actualización idempotente por ID de documento
• Las piezas del diseño de software moderno y eficaz — progresión hacia CQRS y patrón outbox
• Core NATS (Parte 1)
• JetStream (Parte 2)
• JetStream KV (Parte 3)