--- id: "ADR-003" title: "Comunicación orientada a eventos con NATS JetStream" date: "2026-04-12" status: "active" superseded_by: "" tags: ["events", "nats", "jetstream", "cloudevents", "messaging", "producers", "consumers", "async"] summary: "En el contexto de los microservicios White Label que necesitan coordinar acciones entre dominios y servicios sin crear acoplamiento en tiempo de ejecución, frente al problema de que las llamadas síncronas entre microservicios crean un monolito distribuido donde un fallo en cascada afecta a todos los participantes, decidimos usar NATS JetStream con el estándar CloudEvents para comunicación asíncrona entre microservicios en operaciones de escritura, reservando las llamadas síncronas gRPC para operaciones de lectura que requieren agregación en tiempo real, para lograr que cada microservicio pueda desplegarse y operar de forma independiente, aceptando que la trazabilidad distribuida requiere instrumentación explícita de OpenTelemetry en el shared package." --- ## Contexto Cuando el servicio A llama al B, y el B llama al C, los tres quedan acoplados en tiempo de ejecución: si el C falla, el B falla, y el A falla. Si se necesita desplegar el C, hay que asegurarse de que el B sigue siendo compatible. Lo que empezó como tres microservicios independientes se comporta como un monolito distribuido. White Label está en transición activa desde un modelo de llamadas síncronas entre microservicios hacia comunicación orientada a eventos. El objetivo es que cada microservicio pueda desplegarse, escalar y fallar de forma independiente. El sistema no es completamente event driven hoy. Hay flujos con deuda técnica documentada donde un consumer llama a otro microservicio por gRPC como solución temporal. Al diseñar features nuevas, el punto de partida es siempre el modelo event driven. ## Alternativas consideradas ### Opción A — Llamadas síncronas gRPC para toda comunicación Mantener el modelo actual donde los microservicios se llaman entre sí por gRPC para coordinar flujos de negocio. **Por qué se descartó:** Crea acoplamiento en tiempo de ejecución. Un fallo en cualquier punto de la cadena puede propagar fallos a todos los participantes. Los deploys de un servicio requieren coordinar compatibilidad con todos los que lo llaman. El sistema se comporta como un monolito distribuido. ### Opción B — Message queue genérico (RabbitMQ, SQS) Usar un broker de mensajería estándar sin garantías de persistencia configurables por dominio. **Por qué se descartó:** NATS JetStream ya está en la infraestructura de la plataforma. Agregar otro sistema de mensajería sin una razón concreta agrega complejidad operacional. JetStream proporciona persistencia, durabilidad y garantías de entrega suficientes para los casos de uso de White Label. ### Opción C — Streaming reactivo (Kafka) Usar un sistema de streaming con log inmutable para toda comunicación. **Por qué se descartó:** La complejidad operacional de Kafka no está justificada para los volúmenes actuales de White Label. NATS JetStream proporciona durabilidad y replay sin ese overhead. Kafka es candidato a evaluar si el volumen escala significativamente. ### Opción elegida — NATS JetStream + CloudEvents Comunicación asíncrona con NATS JetStream para operaciones de escritura, llamadas síncronas gRPC para operaciones de lectura con agregación en tiempo real. Todos los eventos siguen el estándar CloudEvents. ## Decisión We will use NATS JetStream with the CloudEvents standard for asynchronous communication between microservices and domains. ### Modelo de despliegue - Un broker NATS JetStream compartido por toda la plataforma. - Cada instancia del microservicio se despliega para un solo tenant (flag `--tenant=co-jumbo`). - El tenant actúa como namespace: todos los subjects llevan el prefijo `.`, garantizando aislamiento lógico entre tenants en el broker compartido. ### Cuándo usar eventos vs. llamadas síncronas **Operaciones de escritura → eventos** Cuando el usuario dispara una acción que tiene consecuencias en otros servicios, el primer servicio procesa la acción, publica un evento y se desentiende de lo que ocurre después. Los demás reaccionan de forma independiente. ``` Usuario actualiza dirección → customers procesa y publica customer.address.seller_resolved → cart consume el evento y actualiza el carrito → auth consume el evento y actualiza los tokens de segmento (customers no sabe de cart ni de auth) ``` **Operaciones de lectura → llamadas síncronas** Cuando se necesita construir una respuesta con datos agregados de múltiples fuentes en tiempo real, la llamada síncrona es el modelo correcto. ``` search devuelve productos con promociones → llama síncronamente a promotions para calcular descuentos (respuesta inmediata requerida) ``` **La pregunta clave:** ¿esta operación es un comando o una query? - Comando que dispara consecuencias en otros servicios → eventos - Query que necesita agregar datos en tiempo real → llamada síncrona ### Estructura de eventos: CloudEvents Todos los eventos siguen el estándar [CloudEvents](https://cloudevents.io/): ```go const CUSTOMER_SOURCE = "dc-wl-groceries-core-customers" const CUSTOMER_ADDRESS_CHANNEL = "customer.address" const AddressSellerResolvedEvent = "customer.address.seller_resolved" type AddressSellerResolved struct { ID string `json:"id"` AddressID string `json:"addressId"` Username string `json:"username"` // ... } ``` Convenciones: - El tipo de evento (`type`) va en formato `..` en pasado. - El canal (`channel`) agrupa todos los eventos de un dominio: `customer.address`. - El subject de NATS incluye el prefijo del tenant para aislamiento: `co-jumbo.customer.address.seller_resolved`. - **Los eventos describen hechos, no órdenes.** `address.seller_resolved`, no `update_cart`. ### Producers Los producers viven en la **capa Data** del dominio que emite el evento. Su responsabilidad es mínima: recibir el evento del dominio, construir el CloudEvent y delegarle la publicación al NatsProducer del shared package. ```go // data/producer/address_seller_resolved.go func (p *AddressSellerResolvedProducer) Publish(ctx context.Context, ev *event.AddressSellerResolved) error { cEvent := ce.NewEvent() cEvent.SetID(ev.ID) cEvent.SetType(event.AddressSellerResolvedEvent) cEvent.SetSource(event.CUSTOMER_SOURCE) if err := cEvent.SetData(ce.ApplicationJSON, ev); err != nil { return err } return p.producer.Publish(ctx, &cEvent) } ``` **El Core decide cuándo y qué publicar. El producer solo sabe cómo enviarlo.** ### El producer es el propietario del stream Cada dominio inicializa un `NatsProducer` por canal, no por evento. Todos los producers de eventos de ese dominio comparten el mismo `NatsProducer`. **El producer es responsable de crear y configurar el stream.** Al inicializarse llama `CreateOrUpdateStream` con la configuración del canal: nombre, subjects y retention period. Esta operación es idempotente — si el stream ya existe, lo actualiza; si no existe, lo crea. - **Nombre del stream**: `_` (mayúsculas, puntos → guiones bajos). Ejemplo: `co-jumbo.customer.address` → `CO_JUMBO_CUSTOMER_ADDRESS`. - **Subjects del stream**: `..>` (wildcard — captura todos los eventos del canal). Ejemplo: `co-jumbo.customer.address.>`. - **Subject de cada mensaje**: `.`, donde `event.Type()` sigue el formato `..`. Ejemplo: `co-jumbo.customer.address.seller_resolved`. - **Retention period**: definido por el propietario del stream según el caso de uso. Puede ser un valor fijo en código o configurable. No hay un valor estándar de plataforma. ```go // data/producer/module.go func NewCustomerAddressProducer(tenant config.TenantValue, js jetstream.JetStream) (*nats.NatsProducer, error) { return nats.NewNatsProducer(nats.NatsProducerParams{ Channel: event.CUSTOMER_ADDRESS_CHANNEL, RetentionPeriod: 72 * time.Hour, // definido por el propietario según el caso de uso // ... }) } ``` ### Consumers Los consumers viven en la carpeta `consumer/` del dominio y son el equivalente del Service para eventos entrantes. Cada consumer maneja **un solo tipo de evento**. **El consumer no crea el stream — asume que ya fue creado por el producer propietario.** Los consumers son durables: su nombre sigue el formato `-`, lo que permite que JetStream recuerde la posición de lectura entre reinicios del servicio. El `NatsConsumer` del shared-pkg gestiona internamente un worker pool configurable (`NumWorkers`, `WorkerBufferSize`). La política de reentrega es automática: si el handler devuelve error, el mensaje recibe un NAK y JetStream lo reentrega; si el handler tiene éxito, recibe un ACK. El consumer de dominio solo implementa el handler. ```go // consumer/address_seller_resolved.go func (c *AddressSellerResolvedConsumer) Subscribe(ctx context.Context) error { consumer, err := nats.NewNatsConsumer(nats.NatsConsumerParams{ Channel: event.CUSTOMER_ADDRESS_CHANNEL, Event: event.AddressSellerResolvedEvent, NumWorkers: 5, // concurrencia configurable según el caso de uso Handler: func(ctx context.Context, ev ce.Event) error { var e event.AddressSellerResolved if err := ev.DataAs(&e); err != nil { return err } return c.Handler(ctx, &e) }, }) // ... } ``` **El consumer deserializa y delega al Core. Sin lógica de negocio propia.** ### Documentar deuda técnica Cuando un consumer llama a otro microservicio por gRPC como solución temporal, se deja un comentario explícito en el código con el diseño correcto al que migrar: ```go // DEUDA TÉCNICA: este consumer llama a cart por gRPC como solución temporal. // El diseño correcto es que cart consuma el evento customer.address.seller_resolved // directamente y reaccione de forma independiente. // Ver: https://... ``` La deuda que no se documenta no se paga. ## Consecuencias **Positivas:** - Cada microservicio puede desplegarse y operar de forma independiente. - Un fallo en un consumer no afecta al producer ni a los demás consumers. - JetStream garantiza que ningún mensaje se pierde: si el consumer falla, reintenta. - El tracing distribuido es transparente: el shared package propaga el contexto OTel del producer al consumer automáticamente. **Negativas:** - La depuración de flujos asíncronos es más compleja que las llamadas síncronas. - La consistencia eventual requiere que los consumidores sean idempotentes. - El sistema no es completamente event driven hoy: hay deuda técnica documentada. **Riesgos y mitigaciones:** - **Microservicio que llama a otro por gRPC en un flujo de escritura** → Deuda técnica a documentar explícitamente en el código. Agregar un issue/ticket con el diseño correcto. - **Eventos que actúan como comandos** (nombre en imperativo, acoplamiento implícito) → Renombrar al patrón `..`. Ejemplo: `cart.updated` en lugar de `update_cart`. - **Consumer que tiene lógica de negocio** → Refactorizar al Core. El consumer solo deserializa y delega. ## ADRs relacionados - ADR-000: Arquitectura en cuatro capas (producers en Data, consumers en Service layer) - ADR-002: Modelo multi-tenant (el canal NATS incluye el prefijo del tenant para aislamiento)