SSE Backend Setup

1// The shape every row you stream must match, live or backfill.
2{
3  channel:   string   // REQUIRED on SSE. One stream carries every channel
4  entity:    string   // e.g. 'item', 'notification', 'job'
5  id:        string   // the row's stable identifier, never empty
6  op:        'CREATE' | 'UPDATE' | 'DELETE'   // 'INSERT' is accepted as CREATE
7  data:      object   // the FULL row. For DELETE send { id } and nothing else is needed
8  timestamp: number   // epoch ms on YOUR server clock, always a finite number
9  source:    string   // any string identifying your server, for logging
10}
11 
12// A message with no channel, no id, no finite timestamp, or no object data on
13// a non DELETE is dropped and counted as lost. The adapter never invents an id.

1// ONE stream for every channel. The client opens:
2//   GET /api/stream?channels={"items":1718000000000,"tasks":1718000000000}
3// "channels" is URL encoded JSON: channel name to sync anchor (epoch ms).
4//
5// Row events are plain text over the open HTTP response:
6//
7//   event: UPDATE\n
8//   data: {"channel":"items","entity":"item","id":"662f...","op":"UPDATE", ...}\n
9//   \n
10//
11// The "event:" name must match the `event` on the matching flux.register()
12// call on the client (commonly 'UPDATE'). Routing to a channel is done by the
13// "channel" field inside the data, never by the event name.
14//
15// Each channel's backfill ENDS with a control event:
16//
17//   event: flux:backfill\n
18//   data: {"channel":"items","asOf":1718000099000}\n
19//   \n
20//
21// or, when you cannot answer from that anchor:
22//
23//   event: flux:backfill\n
24//   data: {"channel":"items","tooOld":true}\n
25//   \n
26//
27// The blank line after "data:" is required. It tells the browser's
28// EventSource parser that one event is complete.

1// What the adapter concludes from the rows and the marker for one channel:
2//
3//   marker arrives, no rows before it      confirmed current
4//   marker arrives, rows before it         applied
5//   marker has tooOld: true                too old, the client refetches
6//   a row before the marker was lost       failed, never claimed as covered
7//   no marker within 30 seconds            failed (timeout), the gap stays open
8//
9// Rows that arrive BEFORE a channel's marker are its backfill. Rows that
10// arrive AFTER it are live. The marker's asOf becomes the next anchor, so
11// send it slightly in the past (the helpers below use 5 seconds).
12//
13// If your backfill query FAILS, do not send a marker. A marker with no rows
14// reads as "nothing changed" and the client would believe it is up to date.
15// With no marker the adapter times out and reports a failure, which is safe.

1// server/auth.js
2const jwt = require('jsonwebtoken')
3 
4// EventSource cannot set headers. Identify the user with a token in the query
5// string, or with an HttpOnly cookie when the API is same origin (pass
6// { withCredentials: true } in the adapter's options).
7function authenticate(req) {
8  const token = req.query.token || (req.cookies && req.cookies.session)
9  if (!token) return null
10 
11  try {
12    return String(jwt.verify(token, process.env.JWT_SECRET).sub)
13  } catch {
14    return null
15  }
16}
17 
18module.exports = { authenticate }

1# server/auth.py
2import os
3import jwt
4from fastapi import Request
5 
6# EventSource cannot set headers. Identify the user with a token in the query
7# string, or with an HttpOnly cookie when the API is same origin.
8def authenticate(request: Request):
9    token = request.query_params.get("token") or request.cookies.get("session")
10    if not token:
11        return None
12    try:
13        return str(jwt.decode(token, os.environ["JWT_SECRET"], algorithms=["HS256"])["sub"])
14    except Exception:
15        return None

1// server/sse_registry.js: per channel registry with an audience filter
2//
3// One browser opens ONE stream that carries every channel, so a client is a
4// single object registered under each channel it asked for.
5//   client = { res, userId, ready, buffer }
6//
7// Live events that arrive while a client's backfill is still running are
8// BUFFERED and flushed after the markers. Otherwise a live row could be sent
9// first and then overwritten by an older backfill row for the same entity.
10const sseClients = new Map() // channel to Set of client
11 
12function writeEvent(res, eventName, payload) {
13  res.write('event: ' + eventName + '\n' + 'data: ' + JSON.stringify(payload) + '\n\n')
14}
15 
16function registerSseClient(channel, client) {
17  if (!sseClients.has(channel)) sseClients.set(channel, new Set())
18  sseClients.get(channel).add(client)
19}
20 
21function removeSseClient(channel, client) {
22  const clients = sseClients.get(channel)
23  if (clients) {
24    clients.delete(client)
25    if (clients.size === 0) sseClients.delete(channel)
26  }
27}
28 
29// Marks a client live and flushes anything buffered during its backfill.
30function markSseClientReady(client) {
31  client.ready = true
32  client.buffer.forEach(({ eventName, payload }) => {
33    try { writeEvent(client.res, eventName, payload) } catch {}
34  })
35  client.buffer = []
36}
37 
38// Pushes ONE event to the clients subscribed to a channel.
39//   ownerId null     a GLOBAL row: every client receives it
40//   ownerId 'u_123'  a PER USER row: only that user's streams
41function broadcastToSse(channel, eventName, event, ownerId = null) {
42  const clients = sseClients.get(channel)
43  if (!clients) return
44 
45  const payload = { channel, ...event }
46 
47  clients.forEach((client) => {
48    if (ownerId && client.userId !== String(ownerId)) return
49    if (!client.ready) {
50      client.buffer.push({ eventName, payload })
51      return
52    }
53    try {
54      writeEvent(client.res, eventName, payload)
55    } catch {
56      // A dead socket write throws synchronously. The request's close
57      // handler removes the client from the registry.
58    }
59  })
60}
61 
62module.exports = {
63  writeEvent,
64  registerSseClient,
65  removeSseClient,
66  markSseClientReady,
67  broadcastToSse,
68}

1# server/sse_registry.py
2import asyncio
3import json
4from typing import Dict, Optional, Set
5 
6# One browser opens ONE stream that carries every channel, so a client is a
7# single object registered under each channel it asked for. Live events that
8# arrive during a client's backfill are buffered and flushed after the markers,
9# so a live row is never overwritten by an older backfill row.
10class SseClient:
11    def __init__(self, user_id: Optional[str]):
12        self.user_id = user_id
13        self.queue: asyncio.Queue = asyncio.Queue()
14        self.ready = False
15        self.buffer: list = []
16 
17sse_clients: Dict[str, Set[SseClient]] = {}
18 
19def frame(event_name: str, payload: dict) -> str:
20    return f"event: {event_name}\ndata: {json.dumps(payload)}\n\n"
21 
22def register_sse_client(channel: str, client: SseClient) -> None:
23    sse_clients.setdefault(channel, set()).add(client)
24 
25def remove_sse_client(channel: str, client: SseClient) -> None:
26    clients = sse_clients.get(channel)
27    if clients:
28        clients.discard(client)
29        if not clients:
30            sse_clients.pop(channel, None)
31 
32async def mark_sse_client_ready(client: SseClient) -> None:
33    client.ready = True
34    for item in client.buffer:
35        await client.queue.put(item)
36    client.buffer = []
37 
38# owner_id None: a GLOBAL row for every client. owner_id set: a PER USER row
39# for that user's streams only.
40async def broadcast_to_sse(channel: str, event_name: str, event: dict, owner_id: Optional[str] = None) -> None:
41    clients = sse_clients.get(channel)
42    if not clients:
43        return
44    text = frame(event_name, {"channel": channel, **event})
45    for client in list(clients):
46        if owner_id and client.user_id != str(owner_id):
47            continue
48        if not client.ready:
49            client.buffer.append(text)
50        else:
51            await client.queue.put(text)

1// server/routes/sse.js
2const express = require('express')
3const { authenticate } = require('../auth')
4const {
5  writeEvent,
6  registerSseClient,
7  removeSseClient,
8  markSseClientReady,
9} = require('../sse_registry')
10const { queryChangesSince } = require('../lib/changelog')
11 
12const router = express.Router()
13 
14// GET /api/stream?channels={"items":<anchorMs>,"tasks":<anchorMs>}
15router.get('/', async (req, res) => {
16  let anchors = {}
17  try {
18    anchors = JSON.parse(req.query.channels || '{}')
19  } catch {
20    return res.status(400).json({ error: 'channels must be JSON' })
21  }
22  const channels = Object.keys(anchors)
23 
24  // null means anonymous: global rows only
25  const userId = authenticate(req)
26 
27  res.setHeader('Content-Type', 'text/event-stream')
28  res.setHeader('Cache-Control', 'no-cache')
29  res.setHeader('Connection', 'keep-alive')
30  res.setHeader('X-Accel-Buffering', 'no')
31  res.flushHeaders()
32 
33  const client = { res, userId, ready: false, buffer: [] }
34 
35  // 1. REGISTER FIRST, so a change that commits during the backfill query is
36  //    captured. It is buffered and flushed after the markers.
37  channels.forEach((channel) => registerSseClient(channel, client))
38  console.log('[sse] Client connected, user: ' + (userId || 'anonymous') + ', channels: ' + channels.join(','))
39 
40  res.write(': connected\n\n')
41 
42  // 2. BACKFILL each channel, then its marker.
43  for (const channel of channels) {
44    try {
45      const { rows, asOf, tooOld } = await queryChangesSince(channel, anchors[channel], userId)
46 
47      if (tooOld) {
48        writeEvent(res, 'flux:backfill', { channel, tooOld: true })
49        continue
50      }
51 
52      rows.forEach((row) => {
53        writeEvent(res, 'UPDATE', {
54          channel,
55          entity:    channel,
56          id:        row.id,
57          op:        row.op,
58          data:      row.data,
59          timestamp: row.timestamp,
60          source:    'my_sse_backfill',
61        })
62      })
63 
64      writeEvent(res, 'flux:backfill', { channel, asOf })
65    } catch (err) {
66      // No marker on failure. The adapter times out and reports a failure,
67      // which is correct. A marker here would claim a coverage we do not have.
68      console.error('[sse] Backfill failed for "' + channel + '":', err)
69    }
70  }
71 
72  // 3. Live from here on. Flush anything buffered during the backfill.
73  markSseClientReady(client)
74 
75  const heartbeat = setInterval(() => {
76    try {
77      res.write(': heartbeat\n\n')
78    } catch {
79      clearInterval(heartbeat)
80    }
81  }, 30000)
82 
83  const cleanup = () => {
84    clearInterval(heartbeat)
85    channels.forEach((channel) => removeSseClient(channel, client))
86  }
87 
88  req.on('close', () => {
89    cleanup()
90    console.log('[sse] Client disconnected')
91  })
92  res.on('error', cleanup)
93})
94 
95module.exports = router

1# server/routes/sse.py
2import asyncio
3import json
4from fastapi import APIRouter, Request
5from fastapi.responses import StreamingResponse
6from auth import authenticate
7from sse_registry import (
8    SseClient, frame, register_sse_client, remove_sse_client,
9    mark_sse_client_ready,
10)
11from lib.changelog import query_changes_since
12 
13router = APIRouter()
14 
15# GET /api/stream?channels={"items":<anchorMs>,"tasks":<anchorMs>}
16@router.get("/stream")
17async def stream(request: Request, channels: str = "{}"):
18    anchors = json.loads(channels)
19    names = list(anchors.keys())
20    user_id = authenticate(request)  # None means anonymous: global rows only
21 
22    client = SseClient(user_id)
23 
24    # 1. REGISTER FIRST, so a change that commits during the backfill query is
25    #    captured. It is buffered and flushed after the markers.
26    for name in names:
27        register_sse_client(name, client)
28    print("[sse] Client connected, user:", user_id or "anonymous", "channels:", names)
29 
30    async def event_generator():
31        yield ": connected\n\n"
32 
33        # 2. BACKFILL each channel, then its marker.
34        for name in names:
35            try:
36                result = await query_changes_since(name, anchors[name], user_id)
37 
38                if result["tooOld"]:
39                    yield frame("flux:backfill", {"channel": name, "tooOld": True})
40                    continue
41 
42                for row in result["rows"]:
43                    yield frame("UPDATE", {
44                        "channel": name,
45                        "entity": name,
46                        "id": row["id"],
47                        "op": row["op"],
48                        "data": row["data"],
49                        "timestamp": row["timestamp"],
50                        "source": "my_sse_backfill",
51                    })
52 
53                yield frame("flux:backfill", {"channel": name, "asOf": result["asOf"]})
54            except Exception as err:
55                # No marker on failure. The adapter times out and reports a
56                # failure, which is correct.
57                print("[sse] Backfill failed for", name, err)
58 
59        # 3. Live from here on. Flush anything buffered during the backfill.
60        await mark_sse_client_ready(client)
61 
62        try:
63            while True:
64                try:
65                    text = await asyncio.wait_for(client.queue.get(), timeout=30)
66                    yield text
67                except asyncio.TimeoutError:
68                    yield ": heartbeat\n\n"
69 
70                if await request.is_disconnected():
71                    break
72        finally:
73            for name in names:
74                remove_sse_client(name, client)
75            print("[sse] Client disconnected")
76 
77    return StreamingResponse(
78        event_generator(),
79        media_type="text/event-stream",
80        headers={
81            "Cache-Control": "no-cache",
82            "Connection": "keep-alive",
83            "X-Accel-Buffering": "no",
84        },
85    )

1// server/models/FluxChangelog.js
2const mongoose = require('mongoose')
3 
4// Single state changelog: at most ONE entry per row per table. A new mutation
5// REPLACES the entry, which is what lets a deletion be sent during backfill as
6// an op 'DELETE' row. A plain query on the live collection cannot do that.
7//
8// userId decides who may read the entry:
9//   null      a GLOBAL row, readable by every client
10//   a string  a PER USER row, readable only by that user
11const FluxChangelogSchema = new mongoose.Schema({
12  tableName: { type: String, required: true },
13  rowId:     { type: String, required: true },
14  userId:    { type: String, default: null },
15  op:        { type: String, required: true, enum: ['CREATE', 'UPDATE', 'DELETE'] },
16  rowData:   { type: mongoose.Schema.Types.Mixed, default: null },
17  changedAt: { type: Date, required: true, default: Date.now },
18})
19 
20FluxChangelogSchema.index({ tableName: 1, rowId: 1 }, { unique: true })
21FluxChangelogSchema.index({ tableName: 1, userId: 1, changedAt: 1 })
22// Cleanup after 14 days. queryChangesSince() refuses anchors older than 13
23// days, so a backfill is never answered from a half purged window.
24FluxChangelogSchema.index({ changedAt: 1 }, { expireAfterSeconds: 60 * 60 * 24 * 14 })
25 
26module.exports = mongoose.model('FluxChangelog', FluxChangelogSchema)

1// server/lib/changelog.js
2const mongoose = require('mongoose')
3const FluxChangelog = require('../models/FluxChangelog')
4 
5const RETENTION_MS = 13 * 24 * 60 * 60 * 1000 // must stay below the 14 day cleanup
6const OVERLAP_MS   = 5000                     // asOf is reported this far behind now
7const ROW_LIMIT    = 2000
8 
9// ownerField names the column that holds the owner. null means the table is
10// GLOBAL: the same rows for everyone.
11const tables = {
12  items:         { model: require('../models/Item'),         ownerField: 'userId' },
13  tasks:         { model: require('../models/Task'),         ownerField: 'userId' },
14  notifications: { model: require('../models/Notification'), ownerField: 'userId' },
15  announcements: { model: require('../models/Announcement'), ownerField: null },
16}
17 
18function normalizeRowData(rowId, rawData) {
19  if (!rawData || typeof rawData !== 'object') return { id: String(rowId) }
20  const data = typeof rawData.toObject === 'function' ? rawData.toObject() : { ...rawData }
21  data.id = String(data.id ?? data._id ?? rowId)
22  return data
23}
24 
25// Records one change. Returns { ownerId, timestamp } so the caller broadcasts
26// with the SAME owner and timestamp that backfill will later report.
27async function logChange(tableName, rowId, op, rowData, ownerId = null) {
28  const cleanId = String(rowId ?? rowData?.id ?? rowData?._id ?? '')
29  if (!cleanId) return null
30 
31  const existing = await FluxChangelog.findOne({ tableName, rowId: cleanId })
32    .select('op userId')
33    .lean()
34 
35  // A row the client has not seen yet stays a CREATE even if edited again.
36  const finalOp = op === 'UPDATE' && existing?.op === 'CREATE' ? 'CREATE' : op
37 
38  // A delete carries no document, so the owner comes from the earlier entry.
39  // If the row was never logged it becomes a global tombstone with only an id.
40  const owner = ownerId ? String(ownerId) : (existing?.userId ?? null)
41 
42  const payload   = finalOp === 'DELETE' ? { id: cleanId } : normalizeRowData(cleanId, rowData)
43  const changedAt = new Date()
44 
45  await FluxChangelog.findOneAndUpdate(
46    { tableName, rowId: cleanId },
47    { tableName, rowId: cleanId, userId: owner, op: finalOp, rowData: payload, changedAt },
48    { upsert: true, setDefaultsOnInsert: true }
49  )
50 
51  return { ownerId: owner, timestamp: changedAt.getTime() }
52}
53 
54// Answers a backfill for ONE user. Returns { rows, asOf, tooOld }.
55//   - the user receives their own rows plus global rows, never anyone else's
56//   - an anchor older than the retention window, or no anchor at all (0),
57//     is answered tooOld so the client refetches
58//   - more rows than ROW_LIMIT is answered tooOld, never a partial delta
59async function queryChangesSince(tableName, sinceMs, userId = null) {
60  const now  = Date.now()
61  const asOf = now - OVERLAP_MS
62 
63  const parsed = Number(sinceMs)
64  const since  = Number.isFinite(parsed) && parsed > 0 ? Math.floor(parsed) : 0
65 
66  if (since < now - RETENTION_MS) return { rows: [], asOf, tooOld: true }
67 
68  const audience = userId ? [null, String(userId)] : [null]
69 
70  const found = await FluxChangelog.find({
71    tableName,
72    changedAt: { $gt: new Date(since) },
73    userId:    { $in: audience },
74  })
75    .sort({ changedAt: 1, _id: 1 })
76    .limit(ROW_LIMIT + 1)
77    .lean()
78 
79  if (found.length > ROW_LIMIT) return { rows: [], asOf, tooOld: true }
80 
81  const rows = found.map((r) => ({
82    op:        r.op,
83    id:        r.rowId,
84    data:      r.op === 'DELETE' ? { id: r.rowId } : normalizeRowData(r.rowId, r.rowData),
85    timestamp: r.changedAt.getTime(),
86  }))
87 
88  return { rows, asOf, tooOld: false }
89}
90 
91// Which of these ids still exist for THIS user? Returns null when the server
92// cannot answer (unknown table, or a per user table with no user), and the
93// caller must then report NO deletions.
94async function queryExistingIds(tableName, ids, userId = null) {
95  const entry = tables[tableName]
96  if (!entry || !Array.isArray(ids) || ids.length === 0) return null
97  if (entry.ownerField && !userId) return null
98 
99  const validIds = ids.filter((id) => mongoose.isValidObjectId(id))
100  const filter   = { _id: { $in: validIds } }
101  if (entry.ownerField) filter[entry.ownerField] = userId
102 
103  const found = await entry.model.find(filter).select('_id').lean()
104  return found.map((doc) => doc._id.toString())
105}
106 
107module.exports = { logChange, queryChangesSince, queryExistingIds }

1# server/lib/changelog.py
2import time
3from datetime import datetime, timezone
4from typing import Any, Dict, List, Optional
5from bson import ObjectId
6from db_client import db
7 
8changelog = db.flux_changelogs
9 
10RETENTION_MS = 13 * 24 * 60 * 60 * 1000  # must stay below the 14 day cleanup
11OVERLAP_MS = 5000                        # asOf is reported this far behind now
12ROW_LIMIT = 2000
13 
14# owner_field names the column that holds the owner. None means the table is
15# GLOBAL: the same rows for everyone.
16tables = {
17    "items":         {"collection": db.items,         "owner_field": "userId"},
18    "tasks":         {"collection": db.tasks,         "owner_field": "userId"},
19    "notifications": {"collection": db.notifications, "owner_field": "userId"},
20    "announcements": {"collection": db.announcements, "owner_field": None},
21}
22 
23def to_ms(dt: datetime) -> int:
24    if dt.tzinfo is None:
25        dt = dt.replace(tzinfo=timezone.utc)
26    return int(dt.timestamp() * 1000)
27 
28def to_plain(value: Any) -> Any:
29    if isinstance(value, ObjectId):
30        return str(value)
31    if isinstance(value, datetime):
32        return value.isoformat()
33    if isinstance(value, dict):
34        return {k: to_plain(v) for k, v in value.items() if k != "_id"}
35    if isinstance(value, list):
36        return [to_plain(v) for v in value]
37    return value
38 
39def normalize_row_data(row_id: str, raw: Any) -> dict:
40    data = to_plain(raw) if isinstance(raw, dict) else {}
41    data["id"] = str(data.get("id") or row_id)
42    return data
43 
44async def log_change(table_name: str, row_id: str, op: str, row_data: Any, owner_id: Optional[str] = None):
45    clean_id = str(row_id or "")
46    if not clean_id:
47        return None
48 
49    existing = await changelog.find_one({"tableName": table_name, "rowId": clean_id}, {"op": 1, "userId": 1})
50 
51    # A row the client has not seen yet stays a CREATE even if edited again.
52    final_op = "CREATE" if (op == "UPDATE" and existing and existing["op"] == "CREATE") else op
53 
54    # A delete carries no document, so the owner comes from the earlier entry.
55    owner = str(owner_id) if owner_id else (existing.get("userId") if existing else None)
56 
57    payload = {"id": clean_id} if final_op == "DELETE" else normalize_row_data(clean_id, row_data)
58    changed_at = datetime.now(timezone.utc)
59 
60    await changelog.update_one(
61        {"tableName": table_name, "rowId": clean_id},
62        {"$set": {"tableName": table_name, "rowId": clean_id, "userId": owner,
63                  "op": final_op, "rowData": payload, "changedAt": changed_at}},
64        upsert=True,
65    )
66    return {"ownerId": owner, "timestamp": to_ms(changed_at)}
67 
68async def query_changes_since(table_name: str, since_ms: Any, user_id: Optional[str] = None) -> dict:
69    now = int(time.time() * 1000)
70    as_of = now - OVERLAP_MS
71    since = int(since_ms) if isinstance(since_ms, (int, float)) and since_ms > 0 else 0
72 
73    if since < now - RETENTION_MS:
74        return {"rows": [], "asOf": as_of, "tooOld": True}
75 
76    audience = [None, str(user_id)] if user_id else [None]
77    cursor = (
78        changelog.find({
79            "tableName": table_name,
80            "changedAt": {"$gt": datetime.fromtimestamp(since / 1000, tz=timezone.utc)},
81            "userId": {"$in": audience},
82        })
83        .sort([("changedAt", 1), ("_id", 1)])
84        .limit(ROW_LIMIT + 1)
85    )
86    found = await cursor.to_list(length=ROW_LIMIT + 1)
87 
88    if len(found) > ROW_LIMIT:
89        return {"rows": [], "asOf": as_of, "tooOld": True}
90 
91    rows = [{
92        "op": r["op"],
93        "id": r["rowId"],
94        "data": {"id": r["rowId"]} if r["op"] == "DELETE" else normalize_row_data(r["rowId"], r.get("rowData")),
95        "timestamp": to_ms(r["changedAt"]),
96    } for r in found]
97 
98    return {"rows": rows, "asOf": as_of, "tooOld": False}
99 
100async def query_existing_ids(table_name: str, ids: List[str], user_id: Optional[str] = None) -> Optional[List[str]]:
101    entry = tables.get(table_name)
102    if not entry or not ids:
103        return None
104    if entry["owner_field"] and not user_id:
105        return None
106 
107    valid = [ObjectId(i) for i in ids if ObjectId.is_valid(i)]
108    query: Dict[str, Any] = {"_id": {"$in": valid}}
109    if entry["owner_field"]:
110        query[entry["owner_field"]] = user_id
111 
112    found = entry["collection"].find(query, {"_id": 1})
113    return [str(doc["_id"]) async for doc in found]

1// server/routes/reconcile.js
2//
3// POST /api/reconcile/:channel
4//
5// SSE only flows server to client, so the adapter runs reconcile() as a plain
6// HTTP POST when you pass reconcileUrl to createSseAdapter. It sends
7// { channel, localIds } in chunks of up to 5000 and expects { deletedIds }.
8const express = require('express')
9const { authenticate } = require('../auth')
10const { queryExistingIds } = require('../lib/changelog')
11 
12const router = express.Router()
13 
14router.post('/:channel', async (req, res) => {
15  const { channel } = req.params
16  const localIds = Array.isArray(req.body?.localIds) ? req.body.localIds : []
17  const userId = authenticate(req)
18 
19  try {
20    const existingIds = await queryExistingIds(channel, localIds, userId)
21    // null means "cannot answer", so report no deletions rather than all.
22    const deletedIds = existingIds === null ? [] : localIds.filter((id) => !existingIds.includes(id))
23 
24    return res.json({ channel, deletedIds })
25  } catch (err) {
26    console.error('[reconcile] Failed for channel "' + channel + '":', err)
27    // A non OK response makes the adapter stop and keep what earlier chunks confirmed.
28    return res.status(500).json({ error: 'reconcile failed' })
29  }
30})
31 
32module.exports = router

1# server/routes/reconcile.py
2from fastapi import APIRouter, HTTPException, Request
3from pydantic import BaseModel
4from typing import List
5from auth import authenticate
6from lib.changelog import query_existing_ids
7 
8router = APIRouter()
9 
10class ReconcileRequest(BaseModel):
11    channel: str = ""
12    localIds: List[str] = []
13 
14@router.post("/reconcile/{channel}")
15async def reconcile(channel: str, body: ReconcileRequest, request: Request):
16    user_id = authenticate(request)
17    try:
18        existing = await query_existing_ids(channel, body.localIds, user_id)
19        # None means "cannot answer", so report no deletions rather than all.
20        deleted = [] if existing is None else [i for i in body.localIds if i not in existing]
21        return {"channel": channel, "deletedIds": deleted}
22    except Exception as err:
23        print("[reconcile] Failed for channel", channel, err)
24        # A non OK response makes the adapter stop and keep what earlier chunks confirmed.
25        raise HTTPException(status_code=500, detail="reconcile failed")

1// Client side: the one place the adapter learns about these routes.
2import { createSseAdapter } from '@tsworldtech/flux'
3 
4const adapter = createSseAdapter({
5  url: 'https://api.example.com/api/stream',
6  options: { withCredentials: true },          // only if you use the cookie
7  reconcileUrl: (channel) => 'https://api.example.com/api/reconcile/' + channel,
8  deltaRetentionMs: 13 * 24 * 60 * 60 * 1000,  // match RETENTION_MS on the server
9})

1// server/watchers/items.js
2const Item = require('../models/Item')
3const { broadcastToSse } = require('../sse_registry')
4const { logChange } = require('../lib/changelog')
5 
6function watchItemChanges() {
7  const changeStream = Item.watch([], { fullDocument: 'updateLookup' })
8 
9  changeStream.on('change', async (change) => {
10    const { operationType, documentKey, fullDocument } = change
11    const opMap = { insert: 'CREATE', update: 'UPDATE', replace: 'UPDATE', delete: 'DELETE' }
12    const op = opMap[operationType]
13    if (!op) return
14 
15    const rowId = documentKey._id.toString()
16 
17    // Always the FULL row. updateLookup returns the whole document even for
18    // a partial update, so the wire never carries a partial record.
19    const fullRow = fullDocument ? { ...fullDocument, id: rowId } : { id: rowId }
20 
21    // The owner of a per user row. null for a global collection. A delete has
22    // no document, so logChange() recovers the owner from the earlier entry.
23    const ownerId = fullDocument?.userId ? String(fullDocument.userId) : null
24 
25    // 1. Log the change for backfill and reconciliation
26    const logged = await logChange('items', rowId, op, fullRow, ownerId)
27    if (!logged) return
28 
29    // 2. Broadcast to open streams. Same owner and timestamp the changelog holds.
30    broadcastToSse(
31      'items',
32      'UPDATE',
33      {
34        entity:    'item',
35        id:        rowId,
36        op,
37        data:      op === 'DELETE' ? { id: rowId } : fullRow,
38        timestamp: logged.timestamp,
39        source:    'my_sse_server',
40      },
41      logged.ownerId
42    )
43  })
44 
45  changeStream.on('error', (err) => console.error('[changestream] items error:', err))
46}
47 
48module.exports = { watchItemChanges }

1# server/watchers/items.py
2from db_client import db
3from sse_registry import broadcast_to_sse
4from lib.changelog import log_change, normalize_row_data
5 
6OP_MAP = {"insert": "CREATE", "update": "UPDATE", "replace": "UPDATE", "delete": "DELETE"}
7 
8async def watch_item_changes():
9    async with db.items.watch(full_document="updateLookup") as stream:
10        async for change in stream:
11            op = OP_MAP.get(change["operationType"])
12            if not op:
13                continue
14 
15            row_id = str(change["documentKey"]["_id"])
16            full_doc = change.get("fullDocument")
17 
18            # Always the FULL row. update_lookup returns the whole document.
19            full_row = normalize_row_data(row_id, full_doc) if full_doc else {"id": row_id}
20 
21            # The owner of a per user row. None for a global collection.
22            owner_id = str(full_doc["userId"]) if full_doc and full_doc.get("userId") else None
23 
24            logged = await log_change("items", row_id, op, full_row, owner_id)
25            if not logged:
26                continue
27 
28            await broadcast_to_sse(
29                "items",
30                "UPDATE",
31                {
32                    "entity": "item",
33                    "id": row_id,
34                    "op": op,
35                    "data": {"id": row_id} if op == "DELETE" else full_row,
36                    "timestamp": logged["timestamp"],
37                    "source": "my_sse_server",
38                },
39                logged["ownerId"],
40            )

1// server/server.js
2require('dotenv').config()
3 
4const express      = require('express')
5const cors         = require('cors')
6const cookieParser = require('cookie-parser')
7const { connectDb } = require('./db')
8 
9const sseRoute       = require('./routes/sse')
10const reconcileRoute = require('./routes/reconcile')
11const { watchItemChanges } = require('./watchers/items')
12 
13const app = express()
14 
15app.use(cors({ origin: process.env.CLIENT_ORIGIN?.split(','), credentials: true }))
16app.use(cookieParser())
17app.use(express.json())
18 
19// ONE multiplexed stream: GET /api/stream?channels=...
20app.use('/api/stream', sseRoute)
21 
22// Deletion reconciliation: POST /api/reconcile/:channel
23app.use('/api/reconcile', reconcileRoute)
24 
25const PORT = process.env.PORT || 3001
26 
27connectDb().then(() => {
28  watchItemChanges()
29 
30  app.listen(PORT, '0.0.0.0', () => {
31    console.log('[server] SSE ready on http://0.0.0.0:' + PORT + '/api/stream')
32  })
33})