WebSocket Backend Setup

1// The shape every event you send 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// For LIVE pushes, the adapter routes by the outer "channel" key:
2//   { channel, ...normalizedEvent }
3//
4// e.g.
5{
6  channel:   "items",
7  entity:    "item",
8  id:        "662f...",
9  op:        "UPDATE",
10  data:      { ...row },
11  timestamp: 1718000001234,
12  source:    "my-ws-server",
13}
14//
15// The client keeps exactly ONE callback per channel — the "event" field
16// inside the payload is NOT used for routing. Only the outer "channel" key is.
17//
18// PROTOCOL envelopes (SYNC_COMPLETE, RECONCILE_COMPLETE) are different:
19// they carry an "action" field and are handled by the adapter internally
20// before being reshaped into the normalized event above and forwarded
21// to your store.

1// server/broadcast.js — per-channel WebSocket registry
2const wsClients = new Map() // channel → Set<WebSocket>
3 
4function registerWsClient(channel, ws) {
5  if (!wsClients.has(channel)) {
6    wsClients.set(channel, new Set())
7  }
8  wsClients.get(channel).add(ws)
9}
10 
11function removeWsClient(channel, ws) {
12  const clients = wsClients.get(channel)
13  if (clients) {
14    clients.delete(ws)
15    if (clients.size === 0) wsClients.delete(channel)
16  }
17}
18 
19// Pushes ONE event to every socket currently subscribed to 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 broadcastToWs(channel, event) {
23  const clients = wsClients.get(channel)
24  if (!clients) return
25 
26  const frame = JSON.stringify({ channel, ...event })
27 
28  clients.forEach((ws) => {
29    try {
30      if (ws.readyState === 1) { // WebSocket.OPEN
31        ws.send(frame)
32      }
33    } catch {
34      // A dead socket send throws synchronously — swallow it here,
35      // the client's own 'close' handler removes it from the registry.
36    }
37  })
38}
39 
40function sendToWsClient(ws, channel, event) {
41  try {
42    if (ws.readyState === 1) {
43      ws.send(JSON.stringify({ channel, ...event }))
44    }
45  } catch {}
46}
47 
48module.exports = { registerWsClient, removeWsClient, broadcastToWs, sendToWsClient }

1# server/broadcast.py
2from typing import Dict, Set
3from starlette.websockets import WebSocket
4 
5ws_clients: Dict[str, Set[WebSocket]] = {}
6 
7def register_ws_client(channel: str, ws: WebSocket) -> None:
8    ws_clients.setdefault(channel, set()).add(ws)
9 
10def remove_ws_client(channel: str, ws: WebSocket) -> None:
11    clients = ws_clients.get(channel)
12    if clients:
13        clients.discard(ws)
14        if not clients:
15            ws_clients.pop(channel, None)
16 
17async def broadcast_to_ws(channel: str, event: dict) -> None:
18    clients = ws_clients.get(channel)
19    if not clients:
20        return
21    frame = {"channel": channel, **event}
22    dead = []
23    for ws in clients:
24        try:
25            await ws.send_json(frame)
26        except Exception:
27            dead.append(ws)
28    for ws in dead:
29        remove_ws_client(channel, ws)
30 
31async def send_to_ws_client(ws: WebSocket, channel: str, event: dict) -> None:
32    try:
33        await ws.send_json({"channel": channel, **event})
34    except Exception:
35        pass

1// server/ws-connection.js
2const { registerWsClient, removeWsClient, sendToWsClient } = require('./broadcast')
3const { queryChangesSince, queryExistingIds } = require('./lib/changelog')
4 
5function handleWsConnection(ws, req) {
6  console.log('[ws] Client connected')
7  const subscribedChannels = new Set()
8 
9  ws.on('message', async (raw) => {
10    let frame
11    try {
12      frame = JSON.parse(raw.toString())
13    } catch {
14      return // non-JSON frame — ignore
15    }
16 
17    // ── SUBSCRIBE ──────────────────────────────────────────────────────
18    if (frame.action === 'SUBSCRIBE' && frame.channel) {
19      registerWsClient(frame.channel, ws)
20      subscribedChannels.add(frame.channel)
21      console.log(`[ws] Subscribed to channel: ${frame.channel}`)
22 
23      // REQUIRED — without this ack, _onChannelSubscribedCb never fires
24      // on the client and delta catch-up (Step 6) is never triggered.
25      sendToWsClient(ws, frame.channel, {
26        action:    'SUBSCRIBED',
27        entity:    frame.channel,
28        id:        '',
29        op:        'UPDATE',
30        data:      null,
31        timestamp: Date.now(),
32        source:    'flux-test-server',
33      })
34    }
35 
36    // ── UNSUBSCRIBE ────────────────────────────────────────────────────
37    if (frame.action === 'UNSUBSCRIBE' && frame.channel) {
38      removeWsClient(frame.channel, ws)
39      subscribedChannels.delete(frame.channel)
40      console.log(`[ws] Unsubscribed from channel: ${frame.channel}`)
41    }
42 
43    // ── SYNC_REQUEST (authoritative catch-up) ─────────────────────────
44    if (frame.action === 'SYNC_REQUEST' && frame.channel) {
45      try {
46        const rows = await queryChangesSince(frame.channel, frame.last_sync_anchor)
47        ws.send(JSON.stringify({
48          action:  'SYNC_COMPLETE',
49          channel: frame.channel,
50          rows,
51        }))
52        console.log(
53          `[ws] SYNC_REQUEST "${frame.channel}" since ${frame.last_sync_anchor} — sent ${rows.length} row(s)`
54        )
55      } catch (err) {
56        console.error(`[ws] SYNC_REQUEST failed for "${frame.channel}":`, err)
57        ws.send(JSON.stringify({ action: 'SYNC_COMPLETE', channel: frame.channel, rows: [] }))
58      }
59    }
60 
61    // ── RECONCILE_REQUEST (deletion safety net) ───────────────────────
62    if (frame.action === 'RECONCILE_REQUEST' && frame.channel) {
63      try {
64        const localIds    = Array.isArray(frame.localIds) ? frame.localIds : []
65        const existingIds = await queryExistingIds(frame.channel, localIds)
66        const deletedIds  = localIds.filter((id) => !existingIds.includes(id))
67 
68        ws.send(JSON.stringify({
69          action:     'RECONCILE_COMPLETE',
70          channel:    frame.channel,
71          deletedIds,
72        }))
73      } catch (err) {
74        console.error(`[ws] RECONCILE_REQUEST failed for "${frame.channel}":`, err)
75        ws.send(JSON.stringify({ action: 'RECONCILE_COMPLETE', channel: frame.channel, deletedIds: [] }))
76      }
77    }
78  })
79 
80  ws.on('close', () => {
81    subscribedChannels.forEach((channel) => removeWsClient(channel, ws))
82    subscribedChannels.clear()
83    console.log('[ws] Client disconnected')
84  })
85 
86  ws.on('error', () => {
87    subscribedChannels.forEach((channel) => removeWsClient(channel, ws))
88    subscribedChannels.clear()
89  })
90}
91 
92module.exports = { handleWsConnection }

1# server/ws_connection.py
2from starlette.websockets import WebSocket, WebSocketDisconnect
3from broadcast import register_ws_client, remove_ws_client, send_to_ws_client
4from lib.changelog import query_changes_since, query_existing_ids
5import time
6 
7async def handle_ws_connection(ws: WebSocket):
8    await ws.accept()
9    print("[ws] Client connected")
10    subscribed_channels = set()
11 
12    try:
13        while True:
14            frame = await ws.receive_json()
15 
16            if frame.get("action") == "SUBSCRIBE" and frame.get("channel"):
17                channel = frame["channel"]
18                register_ws_client(channel, ws)
19                subscribed_channels.add(channel)
20                print(f"[ws] Subscribed to channel: {channel}")
21 
22                # REQUIRED ack
23                await send_to_ws_client(ws, channel, {
24                    "action": "SUBSCRIBED",
25                    "entity": channel, "id": "", "op": "UPDATE", "data": None,
26                    "timestamp": int(time.time() * 1000), "source": "flux-test-server",
27                })
28 
29            if frame.get("action") == "UNSUBSCRIBE" and frame.get("channel"):
30                channel = frame["channel"]
31                remove_ws_client(channel, ws)
32                subscribed_channels.discard(channel)
33                print(f"[ws] Unsubscribed from channel: {channel}")
34 
35            if frame.get("action") == "SYNC_REQUEST" and frame.get("channel"):
36                channel = frame["channel"]
37                rows = await query_changes_since(channel, frame.get("last_sync_anchor", 0))
38                await ws.send_json({
39                    "action": "SYNC_COMPLETE",
40                    "channel": channel,
41                    "rows": [{"op": r["op"], "data": r["data"], "timestamp": r["timestamp"]} for r in rows],
42                })
43 
44            if frame.get("action") == "RECONCILE_REQUEST" and frame.get("channel"):
45                channel = frame["channel"]
46                local_ids = frame.get("localIds", [])
47                existing = await query_existing_ids(channel, local_ids)
48                deleted_ids = [i for i in local_ids if i not in existing]
49                await ws.send_json({
50                    "action": "RECONCILE_COMPLETE",
51                    "channel": channel,
52                    "deletedIds": deleted_ids,
53                })
54 
55    except WebSocketDisconnect:
56        for channel in subscribed_channels:
57            remove_ws_client(channel, ws)
58        print("[ws] Client disconnected")

1// server/sync-anchors.js — the EAGER catch-up path
2//
3// createWebSocketAdapter encodes the SyncAnchorMap into the connection URL
4// as a single URL-encoded JSON query parameter, evaluated fresh at connect()
5// time — before any channel subscription is confirmed:
6//
7//   ws://localhost:3001?syncAnchors=%7B%22items%22%3A1718000000000%7D
8//
9// This is a best-effort head start, not the authoritative catch-up
10// mechanism — see Step 6 for the request/response handshake that actually
11// gates engine.ts's cache-clock refresh.
12 
13const { URL } = require('url')
14const { queryChangesSince } = require('./lib/changelog')
15 
16async function handleSyncAnchors(ws, req) {
17  const url = new URL(req.url, `http://${req.headers.host}`)
18  const raw = url.searchParams.get('syncAnchors')
19  if (!raw) return // cold start — no anchors, nothing to catch up
20 
21  let anchors
22  try {
23    anchors = JSON.parse(decodeURIComponent(raw))
24  } catch {
25    console.warn('[ws] Malformed syncAnchors param — skipping eager catch-up')
26    return
27  }
28 
29  for (const [channel, anchorMs] of Object.entries(anchors)) {
30    if (!anchorMs || anchorMs <= 0) continue
31 
32    try {
33      const rows = await queryChangesSince(channel, anchorMs)
34      rows.forEach((row) => {
35        ws.send(JSON.stringify({
36          channel,
37          entity:    channel,
38          id:        row.data?.id ?? '',
39          op:        row.op,
40          data:      row.data,
41          timestamp: row.timestamp,
42          source:    'flux-test-server-eager',
43        }))
44      })
45      if (rows.length > 0) {
46        console.log(`[ws] Eager catch-up sent ${rows.length} row(s) for "${channel}"`)
47      }
48    } catch (err) {
49      console.error(`[ws] Eager catch-up failed for "${channel}":`, err)
50    }
51  }
52}
53 
54module.exports = { handleSyncAnchors }

1// Inside ws.on('message', ...) — the AUTHORITATIVE catch-up path
2//
3// After the client receives the SUBSCRIBED ack from Step 4, engine.ts
4// calls the adapter's sync() method, which sends this frame and waits
5// for a matching SYNC_COMPLETE. This is what actually gates whether Flux
6// refreshes its local cache clock.
7//
8// The rows array must contain objects shaped exactly like this —
9// the adapter's normalizeDeltaRow() expects op, data, and timestamp:
10 
11if (frame.action === 'SYNC_REQUEST' && frame.channel) {
12  const rows = await queryChangesSince(frame.channel, frame.last_sync_anchor)
13 
14  ws.send(JSON.stringify({
15    action:  'SYNC_COMPLETE',
16    channel: frame.channel,
17    rows: rows.map((row) => ({
18      op:        row.op,        // 'CREATE' | 'UPDATE' | 'DELETE'
19      data:      row.data,      // normalized row object with string `id`
20      timestamp: row.timestamp, // unix ms
21    })),
22  }))
23}

1# Inside the message loop — same authoritative handshake
2if frame.get("action") == "SYNC_REQUEST" and frame.get("channel"):
3    channel = frame["channel"]
4    rows = await query_changes_since(channel, frame.get("last_sync_anchor", 0))
5    await ws.send_json({
6        "action": "SYNC_COMPLETE",
7        "channel": channel,
8        "rows": [
9            {"op": r["op"], "data": r["data"], "timestamp": r["timestamp"]}
10            for r in rows
11        ],
12    })

1// Inside ws.on('message', ...) — deletion reconciliation
2//
3// If a client was offline longer than your delta log retains history,
4// some deletions fall outside what SYNC_COMPLETE can report. reconcile()
5// is the safety net — the client sends its full list of locally-cached
6// IDs for a channel, and your server confirms which of them no longer
7// exist in the authoritative collection.
8 
9if (frame.action === 'RECONCILE_REQUEST' && frame.channel) {
10  const localIds    = Array.isArray(frame.localIds) ? frame.localIds : []
11  const existingIds = await queryExistingIds(frame.channel, localIds)
12  const deletedIds  = localIds.filter((id) => !existingIds.includes(id))
13 
14  ws.send(JSON.stringify({
15    action:     'RECONCILE_COMPLETE',
16    channel:    frame.channel,
17    deletedIds,
18  }))
19}

1# Inside the message loop — same reconciliation handling
2if frame.get("action") == "RECONCILE_REQUEST" and frame.get("channel"):
3    local_ids = frame.get("localIds", [])
4    existing_ids = await query_existing_ids(frame["channel"], local_ids)
5    deleted_ids = [i for i in local_ids if i not in existing_ids]
6    await ws.send_json({
7        "action": "RECONCILE_COMPLETE",
8        "channel": frame["channel"],
9        "deletedIds": deleted_ids,
10    })

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 is what makes deletions
7// catchable at all: a plain query on the live collection can never
8// tell "row was deleted" from "row never existed."
9const FluxChangelogSchema = new mongoose.Schema({
10  tableName: { type: String, required: true },
11  rowId:     { type: String, required: true },
12  op:        { type: String, required: true, enum: ['CREATE', 'UPDATE', 'DELETE'] },
13  rowData:   { type: mongoose.Schema.Types.Mixed, default: null },
14  changedAt: { type: Date, required: true, default: Date.now },
15})
16 
17// Enforces single-state: findOneAndUpdate with upsert on this key
18// replaces the existing entry instead of creating a second one.
19FluxChangelogSchema.index({ tableName: 1, rowId: 1 }, { unique: true })
20// What queryChangesSince() actually scans.
21FluxChangelogSchema.index({ tableName: 1, changedAt: 1 })
22// Optional retention — a row untouched for 14 days is pruned
23// automatically. A client offline longer than this for a specific row
24// falls outside delta coverage and needs reconcile() as the fallback.
25FluxChangelogSchema.index(
26  { changedAt: 1 },
27  { expireAfterSeconds: 60 * 60 * 24 * 14 }
28)
29 
30module.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 
5# Assume async Motor client and models are configured elsewhere
6changelog_collection = db.flux_changelogs
7model_registry = {
8    "items": db.items,
9    "notifications": db.notifications,
10    "tasks": db.tasks,
11}
12 
13def normalize_row_data(row_id: str, raw_data: Any) -> Any:
14    if not raw_data or not isinstance(raw_data, dict):
15        return raw_data
16    data = dict(raw_data)
17    resolved = str(data.get("id") or data.get("_id") or row_id or "")
18    if resolved:
19        data["id"] = resolved
20    return data
21 
22async def log_change(table_name: str, row_id: str, op: str, raw_data: Any, updated_fields: dict = None):
23    clean_id = str(row_id or raw_data.get("id") or raw_data.get("_id") or "")
24    if not clean_id:
25        return
26    if op == "UPDATE":
27        sparse = dict(updated_fields) if updated_fields else (dict(raw_data) if raw_data else {})
28        sparse.pop("_id", None)
29        payload = {"id": clean_id, **sparse}
30    elif op == "DELETE":
31        payload = {"id": clean_id}
32    else:
33        payload = normalize_row_data(clean_id, raw_data)
34    
35    await changelog_collection.update_one(
36        {"tableName": table_name, "rowId": clean_id},
37        {"$set": {"tableName": table_name, "rowId": clean_id, "op": op, "rowData": payload, "changedAt": datetime.utcnow()}},
38        upsert=True,
39    )
40 
41async def query_changes_since(table_name: str, since_ms: int) -> List[dict]:
42    valid_ms = max(0, int(since_ms)) if isinstance(since_ms, (int, float)) else 0
43    since_date = datetime.utcfromtimestamp(valid_ms / 1000)
44    raw_rows = await changelog_collection.find({
45        "tableName": table_name,
46        "changedAt": {"$gt": since_date},
47    }).sort("changedAt", -1).to_list(length=None)
48    
49    latest_by_row_id: Dict[str, dict] = {}
50    for row in raw_rows:
51        rid = row["rowId"]
52        if rid not in latest_by_row_id:
53            latest_by_row_id[rid] = row
54    
55    deduplicated = sorted(latest_by_row_id.values(), key=lambda r: r["changedAt"])
56    deduplicated = deduplicated[:2000]
57    
58    return [{
59        "op": r["op"],
60        "data": normalize_row_data(r["rowId"], r.get("rowData")),
61        "timestamp": int(r["changedAt"].timestamp() * 1000),
62    } for r in deduplicated]
63 
64async def query_existing_ids(table_name: str, ids: List[str]) -> List[str]:
65    model = model_registry.get(table_name)
66    if not model or not ids:
67        return []
68    from bson import ObjectId
69    found = model.find({"_id": {"$in": [ObjectId(i) for i in ids]}}).projection({"_id": 1})
70    return [str(doc["_id"]) async for doc in found]

1// server/watchers/items.js
2const Item = require('../models/Item')
3const { broadcastToWs } = require('../broadcast')
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    // Extract sparse fields for UPDATEs so the wire payload is minimal
18    let updatedFields = null
19    if (operationType === 'update' && updateDescription?.updatedFields) {
20      updatedFields = { ...updateDescription.updatedFields }
21      delete updatedFields.updatedAt
22      delete updatedFields.__v
23    }
24 
25    const rowData = fullDocument
26      ? { ...fullDocument.toObject?.() ?? fullDocument, id: rowId }
27      : { id: rowId }
28 
29    // 1. Persist to changelog so sync() and reconcile() have data
30    await logChange('items', rowId, op, rowData, updatedFields)
31 
32    // 2. Build wire payload — sparse when possible
33    const wireData = op === 'UPDATE' && updatedFields && Object.keys(updatedFields).length > 0
34      ? { id: rowId, ...updatedFields }
35      : rowData
36 
37    const event = {
38      entity:    'item',
39      id:        rowId,
40      op,
41      data:      wireData,
42      timestamp: Date.now(),
43      source:    'flux-test-server',
44    }
45 
46    // 3. Push to currently-connected live clients
47    broadcastToWs('items', event)
48  })
49 
50  changeStream.on('error', (err) => console.error('[changestream] items error:', err))
51}
52 
53module.exports = { watchItemChanges }

1# server/watchers/items.py
2from db_client import db
3from broadcast import broadcast_to_ws
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            await broadcast_to_ws("items", event)

1// server/index.js
2require('dotenv').config()
3 
4const express  = require('express')
5const http     = require('http')
6const cors     = require('cors')
7const { WebSocketServer } = require('ws')
8const { connectDb } = require('./db')
9 
10const { handleWsConnection } = require('./ws-connection')
11const { handleSyncAnchors }  = require('./sync-anchors')
12const { watchItemChanges }   = require('./watchers/items')
13 
14const app    = express()
15const server = http.createServer(app)
16const wss    = new WebSocketServer({ server })
17 
18app.use(cors({ origin: process.env.CLIENT_ORIGIN?.split(',') }))
19app.use(express.json())
20 
21wss.on('connection', async (ws, req) => {
22  // Eager catch-up (Step 5) runs first, before the message handler
23  // that owns the authoritative SYNC_REQUEST/RECONCILE_REQUEST handling.
24  await handleSyncAnchors(ws, req)
25  handleWsConnection(ws, req)
26})
27 
28const PORT = process.env.PORT || 3001
29 
30connectDb().then(() => {
31  watchItemChanges()
32 
33  server.listen(PORT, '0.0.0.0', () => {
34    console.log(`[server] Listening on 0.0.0.0:${PORT}`)
35  })
36})

1# server/main.py
2import asyncio
3from fastapi import FastAPI, WebSocket
4from ws_connection import handle_ws_connection
5from watchers.items import watch_item_changes
6 
7app = FastAPI()
8 
9@app.websocket("/ws")
10async def ws_endpoint(ws: WebSocket):
11    await handle_ws_connection(ws)
12 
13@app.on_event("startup")
14async def startup():
15    asyncio.create_task(watch_item_changes())
16 
17# 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 / SYNC_COMPLETE)
4// or single delta objects (live events). Your store's setState must
5// handle both shapes and both operation types.
6 
7function createStoreAdapter(set, get) {
8  return {
9    setState: (data) => {
10      if (!data) return
11 
12      // ── 1. Array payload — bootstrap, IDB hydration, or full sync ──
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 live 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}

1wscat -c "ws://localhost:3001"

1{"action":"SUBSCRIBE","channel":"items","event":"UPDATE"}