SSE Backend Setup

1// The shape every event you stream must match — this is the one contract
2// your server needs to satisfy. It's the same shape regardless of which
3// realtime transport a Flux app uses (SSE, WebSocket).
4{
5  entity:    string   // e.g. 'item', 'notification', 'job'
6  id:        string   // the row's stable identifier
7  op:        'CREATE' | 'UPDATE' | 'DELETE'
8  data:      any      // the full row (or the deleted row's id-only shape)
9  timestamp: number   // Date.now() at the moment YOUR server sent this
10  source:    string   // any string identifying your server, for logging
11}

1// The SSE wire format itself — plain text over an open HTTP response:
2//
3//   event: UPDATE\n
4//   data: {"entity":"items","id":"662f...","op":"UPDATE", ...}\n
5//   \n
6//
7// The "event:" name must match the `event` field on the corresponding
8// flux.register() call on the client (commonly 'UPDATE'). The blank line
9// after "data:" is required — it's what tells the browser's EventSource
10// parser that one event frame is complete.

1// server/sse-registry.js — per-channel response registry
2const sseClients = new Map() // channel → Set<Response>
3 
4function registerSseClient(channel, res) {
5  if (!sseClients.has(channel)) {
6    sseClients.set(channel, new Set())
7  }
8  sseClients.get(channel).add(res)
9}
10 
11function removeSseClient(channel, res) {
12  const clients = sseClients.get(channel)
13  if (clients) {
14    clients.delete(res)
15    if (clients.size === 0) sseClients.delete(channel)
16  }
17}
18 
19// Pushes ONE event to every currently-open connection for a channel.
20// Call this the moment your data actually changes — a DB trigger, an ORM
21// hook, a change-stream callback, a message-queue consumer.
22function broadcastToSse(channel, eventName, event) {
23  const clients = sseClients.get(channel)
24  if (!clients) return
25 
26  const data = JSON.stringify(event)
27  clients.forEach((res) => {
28    try {
29      res.write(`event: ${eventName}\ndata: ${data}\n\n`)
30    } catch {
31      // A dead socket write throws synchronously in Node — swallow it here,
32      // the client's own 'close' handler removes it from the registry.
33    }
34  })
35}
36 
37module.exports = { registerSseClient, removeSseClient, broadcastToSse }

1# server/sse_registry.py
2from typing import Dict, Set
3import asyncio
4import json
5 
6sse_clients: Dict[str, Set[asyncio.Queue]] = {}
7 
8def register_sse_client(channel: str, queue: asyncio.Queue) -> None:
9    sse_clients.setdefault(channel, set()).add(queue)
10 
11def remove_sse_client(channel: str, queue: asyncio.Queue) -> None:
12    clients = sse_clients.get(channel)
13    if clients:
14        clients.discard(queue)
15        if not clients:
16            sse_clients.pop(channel, None)
17 
18async def broadcast_to_sse(channel: str, event_name: str, event: dict) -> None:
19    clients = sse_clients.get(channel)
20    if not clients:
21        return
22    data = json.dumps(event)
23    frame = f"event: {event_name}\ndata: {data}\n\n"
24    for queue in clients:
25        await queue.put(frame)

1// server/routes/sse.js
2const express = require('express')
3const { registerSseClient, removeSseClient } = require('../sse-registry')
4const { queryChangesSince } = require('../lib/changelog')
5 
6const router = express.Router()
7 
8// GET /api/stream/:channel?syncAnchor=<unixMs>
9router.get('/:channel', async (req, res) => {
10  const { channel } = req.params
11  const syncAnchorParam = req.query.syncAnchor
12  const syncAnchor = syncAnchorParam ? Number(syncAnchorParam) : 0
13 
14  // ── Required SSE headers ────────────────────────────────────────────────
15  res.setHeader('Content-Type', 'text/event-stream')
16  res.setHeader('Cache-Control', 'no-cache')
17  res.setHeader('Connection', 'keep-alive')
18  res.setHeader('X-Accel-Buffering', 'no')
19  res.flushHeaders()
20 
21  // 1. REGISTER FIRST to ensure live events during backfill query are captured
22  registerSseClient(channel, res)
23  console.log(`[sse] Client connected to channel: ${channel}`)
24 
25  // Leading comment line confirms stream connection
26  res.write(': connected\n\n')
27 
28  // 2. IMMEDIATE CATCH-UP (Backfill)
29  // Run queryChangesSince unconditionally (even for syncAnchor = 0 / cold start)
30  try {
31    const changedRows = await queryChangesSince(channel, syncAnchor)
32    changedRows.forEach((row) => {
33      const event = {
34        entity: channel,
35        id: row.data?.id ?? '',
36        op: row.op,
37        data: row.data,
38        timestamp: row.timestamp,
39        source: 'flux-test-server-sse-backfill',
40      }
41      res.write(`event: UPDATE\ndata: ${JSON.stringify(event)}\n\n`)
42    })
43    if (changedRows.length > 0) {
44      console.log(`[sse] Sent ${changedRows.length} backfill event(s) for "${channel}"`)
45    }
46  } catch (err) {
47    console.error(`[sse] Backfill query failed for "${channel}":`, err)
48  }
49 
50  // ── Heartbeat ────────────────────────────────────────────────────────────
51  const heartbeat = setInterval(() => {
52    try {
53      res.write(': heartbeat\n\n')
54    } catch {
55      clearInterval(heartbeat)
56    }
57  }, 30_000)
58 
59  // ── Cleanup ──────────────────────────────────────────────────────────────
60  req.on('close', () => {
61    clearInterval(heartbeat)
62    removeSseClient(channel, res)
63    console.log(`[sse] Client disconnected from channel: ${channel}`)
64  })
65 
66  res.on('error', (err) => {
67    clearInterval(heartbeat)
68    removeSseClient(channel, res)
69    console.error(`[sse] Response stream error on "${channel}":`, err)
70  })
71})
72 
73module.exports = router

1# server/routes/sse.py
2import asyncio
3import json
4from fastapi import APIRouter, Request
5from fastapi.responses import StreamingResponse
6from sse_registry import register_sse_client, remove_sse_client
7from lib.changelog import query_changes_since
8 
9router = APIRouter()
10 
11@router.get("/stream/{channel}")
12async def stream(channel: str, request: Request, syncAnchor: int = 0):
13    queue: asyncio.Queue = asyncio.Queue()
14    register_sse_client(channel, queue)
15    print(f"[sse] Client connected to channel: {channel}")
16 
17    async def event_generator():
18        yield ": connected\n\n"
19 
20        # Backfill query executes unconditionally
21        try:
22            changed_rows = await query_changes_since(channel, syncAnchor)
23            for row in changed_rows:
24                event = {
25                    "entity": channel,
26                    "id": row.get("data", {}).get("id", ""),
27                    "op": row["op"],
28                    "data": row["data"],
29                    "timestamp": row["timestamp"],
30                    "source": "flux-test-server-sse-backfill",
31                }
32                yield f"event: UPDATE\ndata: {json.dumps(event)}\n\n"
33            if changed_rows:
34                print(f'[sse] Sent {len(changed_rows)} backfill event(s) for "{channel}"')
35        except Exception as err:
36            print(f'[sse] Backfill query failed for "{channel}":', err)
37 
38        try:
39            while True:
40                try:
41                    frame = await asyncio.wait_for(queue.get(), timeout=30)
42                    yield frame
43                except asyncio.TimeoutError:
44                    yield ": heartbeat\n\n"
45 
46                if await request.is_disconnected():
47                    break
48        finally:
49            remove_sse_client(channel, queue)
50            print(f"[sse] Client disconnected from channel: {channel}")
51 
52    return StreamingResponse(
53        event_generator(),
54        media_type="text/event-stream",
55        headers={
56            "Cache-Control": "no-cache",
57            "Connection": "keep-alive",
58            "X-Accel-Buffering": "no",
59        },
60    )

1// server/models/FluxChangelog.js
2const mongoose = require('mongoose')
3 
4// Single-state changelog — at most ONE row per entity per table.
5// A new mutation on a row REPLACES its changelog entry (upsert),
6// rather than appending beside it. This makes deletions catchable
7// for both WS and SSE backfill streams.
8const FluxChangelogSchema = new mongoose.Schema({
9  tableName: { type: String, required: true },
10  rowId:     { type: String, required: true },
11  op:        { type: String, required: true, enum: ['CREATE', 'UPDATE', 'DELETE'] },
12  rowData:   { type: mongoose.Schema.Types.Mixed, default: null },
13  changedAt: { type: Date, required: true, default: Date.now },
14})
15 
16FluxChangelogSchema.index({ tableName: 1, rowId: 1 }, { unique: true })
17FluxChangelogSchema.index({ tableName: 1, changedAt: 1 })
18FluxChangelogSchema.index(
19  { changedAt: 1 },
20  { expireAfterSeconds: 60 * 60 * 24 * 14 }
21)
22 
23module.exports = mongoose.model('FluxChangelog', FluxChangelogSchema)

1// server/lib/changelog.js
2const FluxChangelog = require('../models/FluxChangelog')
3 
4const modelRegistry = {
5  items:         require('../models/Item'),
6  notifications: require('../models/Notification'),
7  tasks:         require('../models/Task'),
8}
9 
10function normalizeRowData(rowId, rawData) {
11  if (!rawData || typeof rawData !== 'object') return rawData ?? null
12  const data = typeof rawData.toObject === 'function' ? rawData.toObject() : { ...rawData }
13  const resolvedId = String(data.id ?? data._id ?? rowId ?? '')
14  if (resolvedId) data.id = resolvedId
15  return data
16}
17 
18async function logChange(tableName, rowId, op, rawData, updatedFields = null) {
19  const cleanId = String(rowId ?? rawData?.id ?? rawData?._id ?? '')
20  if (!cleanId) return
21 
22  let payloadData = null
23  if (op === 'UPDATE') {
24    const sparseFields = updatedFields ? { ...updatedFields } : (rawData ? { ...rawData } : {})
25    delete sparseFields._id
26    payloadData = { id: cleanId, ...sparseFields }
27  } else if (op === 'DELETE') {
28    payloadData = { id: cleanId }
29  } else {
30    payloadData = normalizeRowData(cleanId, rawData)
31  }
32 
33  await FluxChangelog.findOneAndUpdate(
34    { tableName, rowId: cleanId },
35    { tableName, rowId: cleanId, op, rowData: payloadData, changedAt: new Date() },
36    { upsert: true, new: true, setDefaultsOnInsert: true }
37  )
38}
39 
40async function queryChangesSince(tableName, sinceMs) {
41  let validMs = 0
42  if (typeof sinceMs === 'number' && !isNaN(sinceMs)) {
43    validMs = sinceMs > 0 ? sinceMs : 0
44  } else if (typeof sinceMs === 'string') {
45    const parsedNum = Number(sinceMs)
46    if (!isNaN(parsedNum) && parsedNum > 0) validMs = parsedNum
47    else {
48      const parsedDate = new Date(sinceMs).getTime()
49      if (!isNaN(parsedDate) && parsedDate > 0) validMs = parsedDate
50    }
51  }
52 
53  const sinceDate = new Date(validMs)
54  const rawRows = await FluxChangelog.find({
55    tableName,
56    changedAt: { $gt: sinceDate },
57  }).sort({ changedAt: -1 }).lean()
58 
59  const latestByRowId = new Map()
60  for (const row of rawRows) {
61    if (!latestByRowId.has(row.rowId)) latestByRowId.set(row.rowId, row)
62  }
63 
64  const deduplicatedRows = Array.from(latestByRowId.values())
65    .sort((a, b) => new Date(a.changedAt).getTime() - new Date(b.changedAt).getTime())
66    .slice(0, 2000)
67 
68  return deduplicatedRows.map((r) => ({
69    op:        r.op,
70    data:      normalizeRowData(r.rowId, r.rowData),
71    timestamp: new Date(r.changedAt).getTime(),
72  }))
73}
74 
75async function queryExistingIds(tableName, ids) {
76  const Model = modelRegistry[tableName]
77  if (!Model || !Array.isArray(ids) || ids.length === 0) return []
78  const found = await Model.find({ _id: { $in: ids } }).select('_id').lean()
79  return found.map((doc) => doc._id.toString())
80}
81 
82module.exports = { logChange, queryChangesSince, queryExistingIds }

1# server/lib/changelog.py
2from datetime import datetime
3from typing import List, Dict, Any
4 
5changelog_collection = db.flux_changelogs
6model_registry = {
7    "items": db.items,
8    "notifications": db.notifications,
9    "tasks": db.tasks,
10}
11 
12def normalize_row_data(row_id: str, raw_data: Any) -> Any:
13    if not raw_data or not isinstance(raw_data, dict):
14        return raw_data
15    data = dict(raw_data)
16    resolved = str(data.get("id") or data.get("_id") or row_id or "")
17    if resolved:
18        data["id"] = resolved
19    return data
20 
21async def log_change(table_name: str, row_id: str, op: str, raw_data: Any, updated_fields: dict = None):
22    clean_id = str(row_id or raw_data.get("id") or raw_data.get("_id") or "")
23    if not clean_id:
24        return
25    if op == "UPDATE":
26        sparse = dict(updated_fields) if updated_fields else (dict(raw_data) if raw_data else {})
27        sparse.pop("_id", None)
28        payload = {"id": clean_id, **sparse}
29    elif op == "DELETE":
30        payload = {"id": clean_id}
31    else:
32        payload = normalize_row_data(clean_id, raw_data)
33    
34    await changelog_collection.update_one(
35        {"tableName": table_name, "rowId": clean_id},
36        {"$set": {"tableName": table_name, "rowId": clean_id, "op": op, "rowData": payload, "changedAt": datetime.utcnow()}},
37        upsert=True,
38    )
39 
40async def query_changes_since(table_name: str, since_ms: int) -> List[dict]:
41    valid_ms = max(0, int(since_ms)) if isinstance(since_ms, (int, float)) else 0
42    since_date = datetime.utcfromtimestamp(valid_ms / 1000)
43    raw_rows = await changelog_collection.find({
44        "tableName": table_name,
45        "changedAt": {"$gt": since_date},
46    }).sort("changedAt", -1).to_list(length=None)
47    
48    latest_by_row_id: Dict[str, dict] = {}
49    for row in raw_rows:
50        rid = row["rowId"]
51        if rid not in latest_by_row_id:
52            latest_by_row_id[rid] = row
53    
54    deduplicated = sorted(latest_by_row_id.values(), key=lambda r: r["changedAt"])
55    deduplicated = deduplicated[:2000]
56    
57    return [{
58        "op": r["op"],
59        "data": normalize_row_data(r["rowId"], r.get("rowData")),
60        "timestamp": int(r["changedAt"].timestamp() * 1000),
61    } for r in deduplicated]
62 
63async def query_existing_ids(table_name: str, ids: List[str]) -> List[str]:
64    model = model_registry.get(table_name)
65    if not model or not ids:
66        return []
67    from bson import ObjectId
68    found = model.find({"_id": {"$in": [ObjectId(i) for i in ids]}}).projection({"_id": 1})
69    return [str(doc["_id"]) async for doc in found]

1// server/routes/reconcile.js
2//
3// POST /api/reconcile/:channel
4//
5// Because SSE is a strictly server-to-client unidirectional transport,
6// Flux's SSE adapter executes reconcile() via standard HTTP POST if
7// `reconcileUrl` is supplied to `createSseAdapter({ reconcileUrl: ... })`.
8const express = require('express')
9const { queryExistingIds } = require('../lib/changelog')
10 
11const router = express.Router()
12 
13router.post('/:channel', async (req, res) => {
14  const { channel } = req.params
15  const localIds = Array.isArray(req.body?.localIds) ? req.body.localIds : []
16 
17  try {
18    const existingIds = await queryExistingIds(channel, localIds)
19    const deletedIds = localIds.filter((id) => !existingIds.includes(id))
20 
21    return res.json({ channel, deletedIds })
22  } catch (err) {
23    console.error(`[reconcile] Failed for channel "${channel}":`, err)
24    return res.json({ channel, deletedIds: [] })
25  }
26})
27 
28module.exports = router

1# server/routes/reconcile.py
2from fastapi import APIRouter
3from pydantic import BaseModel
4from typing import List
5from lib.changelog import query_existing_ids
6 
7router = APIRouter()
8 
9class ReconcileRequest(BaseModel):
10    localIds: List[str] = []
11 
12@router.post("/reconcile/{channel}")
13async def reconcile(channel: str, body: ReconcileRequest):
14    try:
15        existing_ids = await query_existing_ids(channel, body.localIds)
16        deleted_ids = [i for i in body.localIds if i not in existing_ids]
17        return {"channel": channel, "deletedIds": deleted_ids}
18    except Exception as err:
19        print(f'[reconcile] Failed for channel "{channel}":', err)
20        return {"channel": channel, "deletedIds": []}

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, updateDescription } = 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    let updatedFields = null
18    if (operationType === 'update' && updateDescription?.updatedFields) {
19      updatedFields = { ...updateDescription.updatedFields }
20      delete updatedFields.updatedAt
21      delete updatedFields.__v
22    }
23 
24    const rowData = fullDocument
25      ? { ...fullDocument.toObject?.() ?? fullDocument, id: rowId }
26      : { id: rowId }
27 
28    // 1. Log change for backfill and reconciliation
29    await logChange('items', rowId, op, rowData, updatedFields)
30 
31    const wireData = op === 'UPDATE' && updatedFields && Object.keys(updatedFields).length > 0
32      ? { id: rowId, ...updatedFields }
33      : rowData
34 
35    const event = {
36      entity:    'item',
37      id:        rowId,
38      op,
39      data:      wireData,
40      timestamp: Date.now(),
41      source:    'flux-test-server',
42    }
43 
44    // 2. Broadcast live event over open SSE channels
45    broadcastToSse('items', 'UPDATE', event)
46  })
47 
48  changeStream.on('error', (err) => console.error('[changestream] items error:', err))
49}
50 
51module.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
5import time
6 
7OP_MAP = {"insert": "CREATE", "update": "UPDATE", "replace": "UPDATE", "delete": "DELETE"}
8 
9async def watch_item_changes():
10    async with db.items.watch(full_document="updateLookup") as stream:
11        async for change in stream:
12            op = OP_MAP.get(change["operationType"])
13            if not op:
14                continue
15 
16            full_doc = change.get("fullDocument")
17            update_desc = change.get("updateDescription")
18            row_id = str(change["documentKey"]["_id"])
19 
20            updated_fields = None
21            if change["operationType"] == "update" and update_desc:
22                updated_fields = dict(update_desc.get("updatedFields", {}))
23                updated_fields.pop("updatedAt", None)
24                updated_fields.pop("__v", None)
25 
26            row_data = {**full_doc, "id": str(full_doc["_id"])} if full_doc else {"id": row_id}
27            await log_change("items", row_id, op, row_data, updated_fields)
28 
29            wire_data = updated_fields if (op == "UPDATE" and updated_fields) else row_data
30            event = {
31                "entity": "item",
32                "id": row_id,
33                "op": op,
34                "data": wire_data,
35                "timestamp": int(time.time() * 1000),
36                "source": "flux-test-server",
37            }
38 
39            await broadcast_to_sse("items", "UPDATE", event)

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

1# server/main.py
2import asyncio
3from fastapi import FastAPI
4from fastapi.middleware.cors import CORSMiddleware
5from routes.sse import router as sse_router
6from routes.reconcile import router as reconcile_router
7from watchers.items import watch_item_changes
8 
9app = FastAPI()
10 
11app.add_middleware(
12    CORSMiddleware,
13    allow_origins=["*"],
14    allow_methods=["*"],
15    allow_headers=["*"],
16)
17 
18app.include_router(sse_router, prefix="/api")
19app.include_router(reconcile_router, prefix="/api")
20 
21@app.on_event("startup")
22async def startup():
23    asyncio.create_task(watch_item_changes())
24 
25# uvicorn main:app --host 0.0.0.0 --port 3001

1// Client-side store adapter pattern (Zustand / Redux / Jotai)
2//
3// Flux's adapter forwards either full arrays (bootstrap) or single delta
4// objects (live SSE events & backfill frames). Your store's setState must
5// handle both payload shapes and operations.
6 
7function createStoreAdapter(set, get) {
8  return {
9    setState: (data) => {
10      if (!data) return
11 
12      // ── 1. Array payload — initial bootstrap or IDB hydration ──────
13      if (Array.isArray(data)) {
14        set({ items: [...data], isHydrated: true, isLive: true })
15        return
16      }
17 
18      // ── 2. Object payload — single delta frame from SSE stream ─────
19      if (typeof data === 'object') {
20        const current = get().items
21 
22        // DELETE
23        if (data.op === 'DELETE' || data.eventType === 'DELETE') {
24          const id = String(data.id ?? data._id ?? data.data?.id ?? '')
25          if (id) set({ items: current.filter((i) => i.id !== id) })
26          return
27        }
28 
29        // Extract record (handles both nested data payloads and direct shapes)
30        const rawRecord = data.data ?? data.new ?? data
31        const recordId  = String(rawRecord?.id ?? rawRecord?._id ?? data.id ?? '')
32 
33        // Sparse UPDATE — shallow merge onto existing record
34        const idx = current.findIndex((i) => i.id === recordId)
35        if (idx !== -1) {
36          const next = [...current]
37          next[idx] = { ...next[idx], ...rawRecord, id: recordId }
38          set({ items: next, isHydrated: true, isLive: true })
39        } else {
40          // CREATE — prepend new record
41          set({
42            items: [{ id: recordId, ...rawRecord }, ...current],
43            isHydrated: true,
44            isLive: true,
45          })
46        }
47      }
48    }
49  }
50}

1curl -N "http://localhost:3001/api/stream/items?syncAnchor=0"