Recipe

Server-Sent Events as a typed contract

TypeFirst · AgileLabs Engineering · 10 min read

When the server needs to push — live notifications, job progress, a feed that updates itself — most teams jump straight to WebSockets and inherit a second protocol, a reconnect strategy, and a connection manager. For one-directional push, HTTP already has the right tool: Server-Sent Events. It's plain HTTP (works through proxies and the existing auth header), and the browser side is a streaming response body.

The problem is typing it. A normal T3 contract describes one request → one response. An SSE endpoint is one request → many payloads over time. So we extend the contract pattern: instead of a responseDecoder, the contract declares an eventDecoder — and both sides answer to it.

Step 1 — The contract

// Core/Data/SseApi.ts
export type SseApi<
  Route extends string,
  UrlParams extends UrlRecord<Route>,
  Event,
> = {
  method: "GET"
  route: Route
  urlDecoder: JD.Decoder<UrlParams>
  eventDecoder: JD.Decoder<Event>
}

A concrete contract makes the event a sum type — so every kind of message the server may push is enumerated, and the client's handling will be exhaustive:

// Core/Api/Auth/NotificationStream.ts
export type Event =
  | { _t: "JobProgress"; jobID: UUID; percent: Nat }
  | { _t: "JobDone"; jobID: UUID }
  | { _t: "NewMessage"; from: Name; preview: Text256 }

export const contract: SseApi<"/notifications", NoUrlParams, Event> = {
  method: "GET",
  route: "/notifications",
  urlDecoder: noUrlParamsDecoder,
  eventDecoder: JD.taggedUnion("_t", { … }),
}

Step 2 — The server binding

One generic function binds any SSE contract to Express, the same way AuthApi binds normal handlers. The handler receives a typed send function — it cannot push anything that isn't an Event — and returns a cleanup function for when the client disconnects:

// Api/src/Api/SseApi.ts
export type SseHandler<P, Event> = (
  authUser: UserRow,
  params: P,
  send: (event: Event) => void,
) => Promise<() => void>  // returns cleanup

export function sseApi<R extends string, P extends UrlRecord<R>, Event>(
  app: Express,
  contract: SseApi<R, P, Event>,
  handler: SseHandler<P, Event>,
): void {
  app.get(contract.route, async (req, res) => {
    // …auth + urlDecoder checks as in AuthApi, then:
    res.writeHead(200, {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache",
      "Connection": "keep-alive",
    })
    res.flushHeaders()

    const send = (event: Event): void => {
      res.write(`data: ${JSON.stringify(event)}\n\n`)
    }

    // keep proxies from killing an idle stream
    const heartbeat = setInterval(() => res.write(":hb\n\n"), 15_000)

    const cleanup = await handler(authUser, params, send)
    req.on("close", () => {
      clearInterval(heartbeat)
      cleanup()
    })
  })
}

Notice what stayed true to the architecture: the handler is still decoupled from Express. It knows nothing about res — it gets a send function and returns a cleanup. Testing it means passing an array-pushing send and asserting on the collected events. The JSON.stringify works on opaque types for free, because every opaque value carries toJSON().

Step 3 — The client: decode, then emit

The browser's EventSource can't set an Authorization header, so for authenticated streams use fetch and read the body stream — ts-bedrock already ships a streaming reader (Core/Data/fetch.ts) whose pattern this follows. Each parsed line goes through the contract's eventDecoder: a malformed event is a decode failure at the boundary, never a mystery crash in a view.

// Web/src/Api/Auth/NotificationStream.ts
export async function listen(
  onEvent: (event: Event) => void,
  signal: AbortSignal,
): Promise<void> {
  const res = await fetch(toUrl(contract.route), {
    headers: authHeader(),
    signal,
  })

  for await (const line of sseLines(res.body)) {  // splits on "\n\n", strips "data: "
    const decoded = contract.eventDecoder.decode(parseJSON(line))
    if (decoded.ok) onEvent(decoded.value)
    // else: log decode error — bad event rejected at the door
  }
}

And here's where it clicks into the TEA runtime: a pushed event is just another reason for state to change, so onEvent wraps each decoded event in an action and emits it. A server push and a button click become indistinguishable to the rest of the app — one state, one action log, same debugging story:

// Web/src/Action/Notification.ts
export function onServerEvent(event: Event): Action {
  return _AuthState((state) => {
    switch (event._t) {
      case "JobProgress": return [_updateJob(state, event), []]
      case "JobDone":     return [_completeJob(state, event), []]
      case "NewMessage":  return [_pushToast(state, event), []]
    }
  })
}

Exhaustive, of course: add a fourth event variant to the contract and this switch — plus the server's send call sites — fail to compile until both ends agree. That's the whole point of extending the contract rather than bolting SSE on ad hoc: the stream evolves with the same compiler-checked discipline as every other endpoint.

Production notes

• Reconnects: wrap listen() with retryPromiseWithDelay (already in Core/Data/Promise.ts) and treat the connection state as a RemoteData in your app state.
• Missed events: include a sequence number in each event and pass ?since=:since as a typed URL param on reconnect — it's in the contract, so it's decoded like everything else.
• Fan-out: the typed send function is the seam: back it with an in-process emitter first, swap in Redis pub/sub later — the contract and the client never change.

← All posts