Server-Sent Events

app.get("/events", async (req, res) => {
  res.set({ "Content-Type": "text/event-stream", "Cache-Control": "no-cache", Connection: "keep-alive", "X-Accel-Buffering": "no" });
  res.flushHeaders();
  const since = req.get("Last-Event-ID");
  for (const e of await events.since(req.user.id, since)) res.write(`id: ${e.id}\nevent: ${e.type}\ndata: ${JSON.stringify(e.data)}\n\n`);
  const unsubscribe = bus.subscribe(req.user.id, (e) => res.write(`id: ${e.id}\nevent: ${e.type}\ndata: ${JSON.stringify(e.data)}\n\n`));
  const ping = setInterval(() => res.write(": ping\n\n"), 15_000);
  req.on("close", () => { clearInterval(ping); unsubscribe(); });
});

@app.get("/events")
async def events(request: Request, user: CurrentUser, last_event_id: str | None = Header(None)):
    async def stream():
        for e in await store.since(user.id, last_event_id):          # resume after reconnect
            yield f"id: {e.id}\nevent: {e.type}\ndata: {json.dumps(e.data)}\n\n"
        async for e in bus.subscribe(user.id):
            if await request.is_disconnected():
                break
            yield f"id: {e.id}\nevent: {e.type}\ndata: {json.dumps(e.data)}\n\n"
    return StreamingResponse(stream(), media_type="text/event-stream",
                             headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})

const es = new EventSource("/events", { withCredentials: true });
es.addEventListener("order.updated", (e) => render(JSON.parse(e.data)));
es.onerror = () => console.warn("reconnecting…");     // EventSource retries automatically