Deliver new items live and catch up after a gap

How to do it

Use Server-Sent Events for one-way updates, or a WebSocket when the page also sends. Send each event with the item's ID. On reconnect, the browser sends Last-Event-ID for Server-Sent Events, so the server can replay newer rows from the table. Subscribe to live items before the replay query, hold them until the replay ends, and skip any ID already sent. Otherwise an item created during the replay is lost. Send a comment line every 15 to 30 seconds, so proxies keep the connection open. Check the session and, for WebSockets, the origin, as in check the origin of WebSocket connections. Fall back to polling every 60 seconds when the live connection fails.

Example

app.get('/api/notifications/stream', requireUser, async (req, res) => {
  res.set({ 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-store' }).flushHeaders();
  let lastSentId = Number(req.get('Last-Event-ID') ?? 0);
  const sendItem = item => {
    if (item.id <= lastSentId) return;
    lastSentId = item.id;
    res.write(`id: ${item.id}\ndata: ${JSON.stringify(item)}\n\n`);
  };
  const heldDuringReplay = [];
  let replaying = true;
  const unsubscribe = live.subscribe(req.user.id, item => (replaying ? heldDuringReplay.push(item) : sendItem(item)));
  (await db.notifications.after(req.user.id, lastSentId)).forEach(sendItem);
  replaying = false;
  heldDuringReplay.forEach(sendItem);
  const keepAlive = setInterval(() => res.write(': ping\n\n'), 20_000);
  req.on('close', () => { clearInterval(keepAlive); unsubscribe(); });
});

func notificationStream(w http.ResponseWriter, r *http.Request) {
	user := currentUser(r)
	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-store")
	flusher := w.(http.Flusher)
	lastSentID, _ := strconv.ParseInt(r.Header.Get("Last-Event-ID"), 10, 64)
	send := func(n Notification) {
		if n.ID <= lastSentID {
			return
		}
		lastSentID = n.ID
		fmt.Fprintf(w, "id: %d\ndata: %s\n\n", n.ID, n.JSON())
		flusher.Flush()
	}
	items, stop := live.Subscribe(user.ID) // buffered channel holds items that arrive during the replay
	defer stop()
	for _, n := range store.After(user.ID, lastSentID) {
		send(n)
	}
	keepAlive := time.NewTicker(20 * time.Second)
	defer keepAlive.Stop()
	for {
		select {
		case n := <-items:
			send(n)
		case <-keepAlive.C:
			fmt.Fprint(w, ": ping\n\n")
			flusher.Flush()
		case <-r.Context().Done():
			return
		}
	}
}

How to check

Open the inbox, cut the network for 30 seconds while two notifications are created, then restore it. Both appear once, in order, without a reload. A test that creates an item during the replay query sees it exactly once.