Implementing Server-Sent Events

res.set({
  "Content-Type":  "text/event-stream; charset=utf-8",
  "Cache-Control": "no-cache, no-transform",
  "Connection":    "keep-alive",
  "X-Accel-Buffering": "no"  // tell Nginx not to buffer
});
res.flushHeaders();

res.write(`data: ${JSON.stringify({ msg: "hello" })}\n\n`);

function sse(res, { event, data, id, retry }) {
  if (id)    res.write(`id: ${id}\n`);
  if (event) res.write(`event: ${event}\n`);
  if (retry) res.write(`retry: ${retry}\n`);
  res.write(`data: ${typeof data === "string" ? data : JSON.stringify(data)}\n\n`);
}

sse(res, { event: "user.created",  data: user });
sse(res, { event: "user.updated",  data: user });
sse(res, { event: "user.deleted",  data: { id } });
// Client: src.addEventListener("user.created", ...)

app.get("/events", async (req, res) => {
  // standard SSE headers...
  const since = req.get("Last-Event-ID");
  for await (const evt of bus.replay(since)) {
    sse(res, { id: evt.id, event: evt.type, data: evt.payload });
  }
  // continue streaming new events...
});

res.write("retry: 10000\n\n");

const ping = setInterval(() => res.write(": ping\n\n"), 15_000);
req.on("close", () => clearInterval(ping));

const clients = new Set();
app.get("/feed", (req, res) => {
  // standard SSE headers...
  clients.add(res);
  req.on("close", () => clients.delete(res));
});

export function broadcast(event, data) {
  for (const res of clients) sse(res, { event, data });
}

sse(res, { event: "shutdown", data: { reason: "deploy" } });
res.end();  // browser will reconnect automatically per retry: directive