Skip to content

Add SSE id: and retry: field support to RiverEmitter #3

Description

@Bewinxed

Problem

RiverEmitter.emitSingleEvent() currently emits SSE messages in this format:

event: <type>
data: <json>

The SSE spec defines two additional fields that are critical for production use:

  • id: — sets lastEventId on the client. On reconnection, the browser's EventSource automatically sends a Last-Event-ID header, allowing the server to replay missed events. Without id:, reconnection = data loss.
  • retry: — tells the client how many milliseconds to wait before reconnecting. Without it, the browser uses its own default (varies by engine).

Neither field is currently supported by RiverEmitter.

Desired Behavior

1. id: field on emit()

Allow passing an optional id in the emit payload:

await emit('event:created', { data: { ... }, id: '019abc12-3456-7890' })

Wire format becomes:

id: 019abc12-3456-7890
event: event:created
data: {"childId":"..."}

The id field should be:

  • Optional (backwards compatible — omit id: line if not provided)
  • Available on both emitSingleEvent and emitStreamEvent (each chunk gets the same id, or sequential ids)
  • Passed through broadcast() and sendToClient() unchanged

2. retry: field

Add a way to emit a retry: directive, either:

  • As part of the emit payload: await emit('connected', { data: {...}, retry: 3000 })
  • Or as a standalone method: emitter.setRetry(writer, 3000)

Wire format:

retry: 3000

3. Last-Event-ID access on server

The stream() callback should receive the Last-Event-ID request header value (if present) so the server can replay missed events:

emitter.stream({
  callback: async (emit, clientId, lastEventId) => {
    if (lastEventId) {
      // Replay events since lastEventId
      const missed = await db.query('SELECT * FROM events WHERE id > $1', [lastEventId])
      for (const event of missed) {
        await emit('event:created', { data: event, id: event.id })
      }
    }
    // Then enter live mode...
  },
  signal: request.signal,
  lastEventId: request.headers.get('Last-Event-ID'), // new option
})

Type Changes

The EmitPayload type should be extended to accept optional id: string and retry: number fields without breaking existing usage. These should be stripped from the data before serialization (they're SSE framing, not payload).

Context

This is needed for Faseela Phase 4 (Realtime SSE). Parents need reliable reconnection with catch-up when their phone goes to sleep or loses network briefly. UUIDv7 primary keys are used as SSE ids (chronologically sortable → WHERE id > lastEventId for catch-up).

Implementation Notes

The change is small — mostly in emitSingleEvent() and emitStreamEvent():

// Current format string (server/index.mjs line 115-120):
`event: ${String(event_type)}\ndata: ${JSON.stringify(payload)}\n\n`

// New format:
`${id ? `id: ${id}\n` : ''}${retry ? `retry: ${retry}\n` : ''}event: ${String(event_type)}\ndata: ${JSON.stringify(dataOnly)}\n\n`

Plus threading lastEventId through the stream() options to the callback.

Metadata

Metadata

Labels

No labels
No labels

Projects

No projects

Milestone

No milestone

Relationships

None yet

Development

No branches or pull requests

Issue actions