Streams SSE y WebSocket

Un stream es una conexión larga por la que el motor te manda los eventos del ledger a medida que llegan. Sirve cuando tu proceso no puede recibir conexiones (un worker sin puerto, un proceso detrás de un NAT): en lugar de una URL de entrada, guardas un cursor y reanudas desde ahí.

El stream lee del ledger, no de una cola en memoria: un corte, un reinicio del motor o un cliente lento no pierden eventos. Reanudar con tu cursor no salta ni repite ninguno. Cada evento llega con el mismo sobre que un webhook, con la misma redacción por retención.

Las rutas

Ruta	Transporte	Qué eventos
GET /v1/events/stream	SSE	Los de todas las instancias que alcanza tu key.
GET /v1/events/ws	WebSocket	Ídem.
GET /v1/instances/{id}/events/stream	SSE	Los de una instancia.
GET /v1/instances/{id}/events/ws	WebSocket	Ídem.
GET /v1/events/streams	JSON	Los streams de tu app abiertos en este proceso y sus últimos cierres.

Todas piden el scope events:read.

Parámetro	Qué es
desde (query)	El cursor desde el que reanudar. Gana sobre Last-Event-ID. En el stream de una instancia también puede ser un seq.
Last-Event-ID (encabezado)	El cursor del último evento recibido. Un cliente SSE lo manda solo al reconectar.
types	Tipos separados por coma, hasta 20. Vacío: todos.
instances	Solo en el stream de la app: IDs de instancia separados por coma. Vacío: todas las que alcanza la key.
contrato	La versión del contrato de eventos que esperas (hoy, 1). Una que el motor no sirve es 422.

Sin cursor, el stream empieza desde ahora: lo que ya estaba confirmado no se manda.

La autenticación

La key va en el encabezado Authorization: Bearer TU_API_KEY, igual que en el resto de /v1; en WebSocket, en el handshake. No hay otra forma: el motor no acepta la key en la query.

EventSource y el WebSocket del navegador no sirven

EventSource y WebSocket de un navegador no pueden mandar el encabezado Authorization, así que no pueden abrir un stream del motor. Tampoco tendría sentido: el motor vive en tu red privada y una API key no va en un navegador. Abre el stream desde tu servidor (Node 18+ con fetch, el SDK de Go o cualquier cliente que mande el encabezado) y reparte desde ahí lo que necesite tu interfaz.

Las tramas

SSE

El cuerpo es text/event-stream, con tres tipos de trama y comentarios de keepalive:

retry: 3000
id: mw1.AQ…
event: listo
data: {"contrato":"1","stream_id":"…","cursor":"mw1.AQ…","keepalive_segundos":15,"instancias":[{"instance_id":"…","seq":1186,"ultimo_seq":1186,"atraso":0}]}

id: mw1.AQ…
event: evento
data: {"schema_version":1,"event_id":"…","instance_id":"…","seq":1187,"type":"message.received",…}

: keepalive

event: cierre
data: {"motivo":"apagado","reanudable":true,"cursor":"mw1.AQ…"}

Trama	Qué trae
listo	La primera. Su id: es el cursor de partida; data dice el contrato, el stream_id, cada cuánto llega el keepalive y la posición de cada instancia (seq entregado, ultimo_seq del ledger y atraso).
evento	Un evento. Su id: es el cursor después de este evento; data es el sobre.
: keepalive	Un comentario, cuando no hubo nada durante el intervalo del keepalive.
cierre	La última, cuando el motor corta: motivo, si es reanudable y el cursor.

WebSocket

Los mismos datos, como mensajes JSON de texto. El cliente no manda nada: el motor solo atiende los pings y el cierre.

{"tipo": "listo", "id": "mw1.AQ…", "listo": {"contrato": "1", "cursor": "mw1.AQ…", "keepalive_segundos": 15, "…": "…"}}
{"tipo": "evento", "id": "mw1.AQ…", "evento": {"schema_version": 1, "event_id": "…", "seq": 1187, "…": "…"}}
{"tipo": "keepalive"}
{"tipo": "cierre", "cierre": {"motivo": "apagado", "reanudable": true, "cursor": "mw1.AQ…"}}

El frame de cierre lleva el motivo: credencial_revocada y sin_scope cierran con el código 1008 (policy violation); apagado, cortado_por_operador y error_interno, con 1001 (going away); un cliente lento se corta sin frame de cierre.

El cursor

El cursor es un texto opaco y firmado (empieza con mw1.): dice, por cada instancia, el último seq entregado. Es de tu app: uno de otra app es 403 y uno alterado o que adelanta al ledger es 422. No lo armes ni lo modifiques.

La regla para no perder ni repetir:

1. Procesa el evento (idempotente por event_id).
2. Después, guarda su cursor (el id de la trama).
3. Al reconectar, manda el último guardado en Last-Event-ID o en ?desde=.

Si tu proceso se cae entre el paso 1 y el 2, al reanudar recibes ese evento otra vez: por eso el paso 1 es idempotente.

En el stream de una instancia, desde acepta también un seq simple (?desde=1186 empieza en el 1187), útil para enganchar el stream después de leer el ledger con GET /v1/instances/{id}/events.

El cursor vencido: 410

El motor sirve un stream desde lo que queda dentro de la retención de reanudación (30 días por omisión; streams.dias_reanudacion). Si tu cursor tiene eventos sin leer más viejos que eso (o ya purgados), el motor no salta en silencio: responde 410 con el primer cursor disponible.

410 Gone

{
  "title": "Gone",
  "status": 410,
  "detail": "El cursor tiene eventos sin leer más viejos que la retención del stream. Para seguir, reanuda con primer_cursor: los eventos anteriores ya no se pueden leer por el stream.",
  "primer_cursor": "mw1.AQ…",
  "instancias_vencidas": ["8a1d2c3e-0000-4000-8000-0000000000a1"]
}

Reanudar con primer_cursor es aceptar el salto en las instancias de instancias_vencidas. Antes, decide si necesitas lo que falta y cómo recuperarlo.

El cierre y la reconexión

motivo	¿Reanudable?	Qué pasó	Qué hacer
apagado	Sí	El motor se reinicia.	Reconecta con tu cursor.
cortado_por_operador	Sí	Un operador cortó el stream desde el panel.	Reconecta con tu cursor.
error_interno	Sí	Un error del motor.	Reconecta con backoff.
cliente_lento	Sí	No leíste a tiempo: tu buffer se llenó o una escritura se trabó. Se corta sin trama de cierre.	Reconecta y lee más rápido (procesa fuera del bucle de lectura).
credencial_revocada	No	La key se revocó, venció o la app se suspendió.	Usa otra key.
sin_scope	No	La key perdió events:read.	Pide el scope.

Un corte de red, sin trama de cierre, también es reanudable. Reconecta con un backoff que crezca (el SDK de Go va de 1 s a 30 s). La trama listo de SSE trae retry: 3000, la espera que usa un cliente SSE estándar.

La revocación y la expiración de la key

Un stream abierto vuelve a validar su key cada 5 segundos, y al instante si la revocación pasa en el mismo proceso. Si la key se revocó, venció, perdió el scope o la app se suspendió, el stream se corta con credencial_revocada o sin_scope, sin reanudación: reconectar con la misma key responde 401 o 403.

El keepalive

Si no hubo eventos durante el intervalo (15 s por omisión; streams.segundos_keepalive, de 5 a 120), el motor manda un keepalive. listo dice el intervalo en keepalive_segundos. Si pasan dos intervalos sin recibir nada, da la conexión por muerta y reconecta con tu cursor.

Los límites

Límite	De fábrica	Clave en Ajustes	Al pasarlo
Streams abiertos por app	5, en cada proceso del motor	streams.conexiones_por_app	429 al abrir otro.
Buffer por conexión	1024 KB	streams.kb_buffer	Se corta con cliente_lento.
Espera a un cliente lento	10 s	streams.segundos_cliente_lento	Se corta con cliente_lento.
Retención de reanudación	30 días	streams.dias_reanudacion	410 con primer_cursor.

Al abrir, un stream puede responder además 401 (la key), 403 (sin events:read o un cursor de otra app), 404 (la instancia no existe para tu app), 422 (un cursor inválido, un contrato que no existe, un ID de instances inválido o más de 20 tipos) y 503 (el motor se está reiniciando: vuelve a conectar con tu cursor).

GET /v1/events/streams muestra tus streams abiertos en ese proceso, con su posición y su atraso por instancia, y los últimos cierres con su motivo. Úsalo para saber por qué se cortó uno.

El orden

Dentro de una instancia, los eventos llegan en orden de seq. Entre instancias, en el orden en que el motor los recibió (received_at). Si filtras con types, verás saltos en seq: son los eventos de otros tipos.

Ejemplos

curl

# -N: sin buffer, para ver las tramas a medida que llegan.
curl -N "https://motor.ejemplo.com/v1/events/stream?types=message.received,message.updated" \
  -H "Authorization: Bearer TU_API_KEY" \
  -H "Accept: text/event-stream"

# Reanudar desde el último cursor procesado:
curl -N https://motor.ejemplo.com/v1/events/stream \
  -H "Authorization: Bearer TU_API_KEY" \
  -H "Last-Event-ID: CURSOR_GUARDADO"

Go

import "github.com/projectx27/motor-whatsapp/sdk/go/stream"

// El SDK lee por SSE y reconecta solo con el último cursor entregado.
s := stream.Abrir(stream.Opciones{
  Base:   "https://motor.ejemplo.com",
  Token:  apiKey,
  Tipos:  []string{"message.received", "message.updated"},
  Cursor: cursorGuardado, // vacío: desde ahora
})
defer s.Cerrar()

for {
  e, err := s.Siguiente(ctx)
  var vencido *stream.CursorVencido
  switch {
  case errors.As(err, &vencido):
    // Decide antes de saltar: vencido.Primer acepta perder lo vencido.
    return err
  case err != nil:
    return err // *stream.Cerrado, stream.ErrRechazado o tu ctx
  }
  procesar(e.Evento)      // idempotente por event_id
  guardarCursor(e.Cursor) // después de procesar
}

JavaScript

// Node 18+: fetch lee el cuerpo como stream. EventSource no sirve (no manda la key).
let cursor = cursorGuardado; // '' para empezar desde ahora

async function escuchar() {
  const headers = { Authorization: `Bearer ${apiKey}`, Accept: 'text/event-stream' };
  if (cursor) headers['Last-Event-ID'] = cursor;
  const res = await fetch('https://motor.ejemplo.com/v1/events/stream?types=message.received', { headers });
  if (res.status === 410) {
    const { primer_cursor, instancias_vencidas } = await res.json();
    throw new Error('cursor vencido: decide antes de usar primer_cursor');
  }
  if (!res.ok) throw new Error(`stream rechazado: ${res.status}`);

  const decoder = new TextDecoder();
  let pendiente = '';
  for await (const trozo of res.body) {
    pendiente += decoder.decode(trozo, { stream: true });
    let fin;
    while ((fin = pendiente.indexOf('\n\n')) >= 0) {
      const trama = pendiente.slice(0, fin);
      pendiente = pendiente.slice(fin + 2);
      let id = '', tipo = '', datos = '';
      for (const linea of trama.split('\n')) {
        if (linea.startsWith('id:')) id = linea.slice(3).trim();
        else if (linea.startsWith('event:')) tipo = linea.slice(6).trim();
        else if (linea.startsWith('data:')) datos += linea.slice(5).trimStart();
      }
      if (tipo === 'listo' && !cursor) cursor = id;
      if (tipo === 'evento') {
        await procesar(JSON.parse(datos)); // idempotente por event_id
        cursor = id;
        await guardarCursor(cursor); // después de procesar
      }
      if (tipo === 'cierre') {
        const { motivo, reanudable } = JSON.parse(datos);
        if (!reanudable) throw new Error(`stream cerrado: ${motivo}`);
      }
    }
  }
  // La conexión terminó: quien llama reconecta con backoff y el cursor guardado.
}

Para WebSocket no hay ejemplo con dependencias cero: la biblioteca estándar de Go no trae un cliente WebSocket y el WebSocket estándar no manda encabezados. Usa un cliente que permita Authorization en el handshake y trata cada mensaje según su tipo, con la misma regla del cursor.

Probado con tráfico real

En las pruebas manuales de la v1, un cliente que reanudaba con Last-Event-ID recibió por el stream los mismos 25 eventos que el webhook de la misma app, con el mismo event_id, una vez cada uno (caso A85, 21/10/2026). Con docker compose restart del motor y el cliente conectado, llegó cierre con motivo: apagado y reanudable: true, el SDK reconectó solo en 1 a 2 s, y el stream y el ledger coincidieron en los 25 eventos, sin repetidos ni faltantes (caso A86, 21/10/2026).