WebSocket Backend Setup

1// The shape every LIVE event you send must match. It is the same shape for
2// every realtime transport.
3{
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 frame with no id, no finite timestamp, or no object data on a non DELETE
13// is dropped and counted as lost. The adapter never invents an id.

1// For LIVE pushes the adapter routes by the outer "channel" key:
2//   { channel, ...normalizedEvent }
3{
4  channel:   "items",
5  entity:    "item",
6  id:        "662f...",
7  op:        "UPDATE",
8  data:      { id: "662f...", ...fullRow },
9  timestamp: 1718000001234,
10  source:    "my_ws_server",
11}
12//
13// The client keeps exactly ONE callback per channel. Only the outer
14// "channel" key is used for routing.
15//
16// PROTOCOL envelopes (SUBSCRIBED, SYNC_COMPLETE, RECONCILE_COMPLETE) carry an
17// "action" field and are handled by the adapter internally.

1// server/auth.js
2const jwt = require('jsonwebtoken')
3const { URL } = require('url')
4 
5// A browser WebSocket cannot send custom headers, so the token travels in the
6// connection URL: ws://host:3001?token=...  (the adapter appends its own
7// syncAnchors parameter after it).
8function authenticate(req) {
9  const url = new URL(req.url, 'http://' + req.headers.host)
10  const token = url.searchParams.get('token')
11  if (!token) return null
12 
13  try {
14    return String(jwt.verify(token, process.env.JWT_SECRET).sub)
15  } catch {
16    return null
17  }
18}
19 
20module.exports = { authenticate }

1# server/auth.py
2import os
3import jwt
4 
5# A browser WebSocket cannot send custom headers, so the token travels in the
6# connection URL: ws://host:3001/ws?token=...
7def authenticate(token: str | None):
8    if not token:
9        return None
10    try:
11        return str(jwt.decode(token, os.environ["JWT_SECRET"], algorithms=["HS256"])["sub"])
12    except Exception:
13        return None

1// server/broadcast.js: per channel WebSocket registry with an audience filter
2const wsClients = new Map() // channel to Set of WebSocket. Each ws carries ws.userId
3 
4function registerWsClient(channel, ws) {
5  if (!wsClients.has(channel)) wsClients.set(channel, new Set())
6  wsClients.get(channel).add(ws)
7}
8 
9function removeWsClient(channel, ws) {
10  const clients = wsClients.get(channel)
11  if (clients) {
12    clients.delete(ws)
13    if (clients.size === 0) wsClients.delete(channel)
14  }
15}
16 
17// Pushes ONE event to the sockets subscribed to a channel.
18//   ownerId null     a GLOBAL row: every subscribed socket receives it
19//   ownerId 'u_123'  a PER USER row: only sockets authenticated as that user
20// Call this the moment your data changes: a change stream, a DB trigger
21// listener, an ORM hook, a queue consumer.
22function broadcastToWs(channel, event, ownerId = null) {
23  const clients = wsClients.get(channel)
24  if (!clients) return
25 
26  const frame = JSON.stringify({ channel, ...event })
27 
28  clients.forEach((ws) => {
29    if (ownerId && ws.userId !== String(ownerId)) return
30    try {
31      if (ws.readyState === 1) ws.send(frame) // WebSocket.OPEN
32    } catch {
33      // A dead socket send throws synchronously. The socket's own close
34      // handler removes it from the registry.
35    }
36  })
37}
38 
39module.exports = { registerWsClient, removeWsClient, broadcastToWs }

1# server/broadcast.py
2from typing import Dict, Optional, 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 
17# owner_id None: a GLOBAL row, every subscribed socket receives it.
18# owner_id set:  a PER USER row, only sockets authenticated as that user.
19async def broadcast_to_ws(channel: str, event: dict, owner_id: Optional[str] = None) -> None:
20    clients = ws_clients.get(channel)
21    if not clients:
22        return
23    frame = {"channel": channel, **event}
24    dead = []
25    for ws in clients:
26        if owner_id and getattr(ws.state, "user_id", None) != str(owner_id):
27            continue
28        try:
29            await ws.send_json(frame)
30        except Exception:
31            dead.append(ws)
32    for ws in dead:
33        remove_ws_client(channel, ws)

1// server/ws_connection.js
2const { registerWsClient, removeWsClient } = require('./broadcast')
3const { queryChangesSince, queryExistingIds } = require('./lib/changelog')
4 
5function handleWsConnection(ws, req) {
6  console.log('[ws] Client connected, user: ' + (ws.userId || 'anonymous'))
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 the client never starts its catch up.
24      ws.send(JSON.stringify({ action: 'SUBSCRIBED', channel: frame.channel }))
25    }
26 
27    // UNSUBSCRIBE
28    if (frame.action === 'UNSUBSCRIBE' && frame.channel) {
29      removeWsClient(frame.channel, ws)
30      subscribedChannels.delete(frame.channel)
31      console.log('[ws] Unsubscribed from channel: ' + frame.channel)
32    }
33 
34    // SYNC_REQUEST: the authoritative catch up. Rows are filtered to this
35    // user's own rows plus global rows.
36    if (frame.action === 'SYNC_REQUEST' && frame.channel) {
37      try {
38        const { rows, asOf, tooOld } = await queryChangesSince(
39          frame.channel,
40          frame.last_sync_anchor,
41          ws.userId
42        )
43        ws.send(JSON.stringify({ action: 'SYNC_COMPLETE', channel: frame.channel, rows, asOf, tooOld }))
44        console.log('[ws] SYNC_REQUEST "' + frame.channel + '" sent ' + rows.length + ' row(s), tooOld: ' + tooOld)
45      } catch (err) {
46        // Report the failure. An empty rows array here would be read as
47        // "confirmed current" and the client would skip a refetch it needs.
48        console.error('[ws] SYNC_REQUEST failed for "' + frame.channel + '":', err)
49        ws.send(JSON.stringify({
50          action:  'SYNC_COMPLETE',
51          channel: frame.channel,
52          error:   String((err && err.message) || err),
53        }))
54      }
55    }
56 
57    // RECONCILE_REQUEST: the deletion safety net, scoped to this user.
58    if (frame.action === 'RECONCILE_REQUEST' && frame.channel) {
59      try {
60        const localIds    = Array.isArray(frame.localIds) ? frame.localIds : []
61        const existingIds = await queryExistingIds(frame.channel, localIds, ws.userId)
62        // null means "cannot answer", so report no deletions rather than all.
63        const deletedIds  = existingIds === null ? [] : localIds.filter((id) => !existingIds.includes(id))
64 
65        ws.send(JSON.stringify({ action: 'RECONCILE_COMPLETE', channel: frame.channel, deletedIds }))
66      } catch (err) {
67        console.error('[ws] RECONCILE_REQUEST failed for "' + frame.channel + '":', err)
68        ws.send(JSON.stringify({ action: 'RECONCILE_COMPLETE', channel: frame.channel, deletedIds: [] }))
69      }
70    }
71  })
72 
73  const cleanup = () => {
74    subscribedChannels.forEach((channel) => removeWsClient(channel, ws))
75    subscribedChannels.clear()
76  }
77 
78  ws.on('close', () => {
79    cleanup()
80    console.log('[ws] Client disconnected')
81  })
82  ws.on('error', cleanup)
83}
84 
85module.exports = { handleWsConnection }

1# server/ws_connection.py
2from starlette.websockets import WebSocket, WebSocketDisconnect
3from auth import authenticate
4from broadcast import register_ws_client, remove_ws_client
5from lib.changelog import query_changes_since, query_existing_ids
6 
7async def handle_ws_connection(ws: WebSocket):
8    # Identify the user BEFORE accepting. The token is in the connection URL.
9    ws.state.user_id = authenticate(ws.query_params.get("token"))
10    await ws.accept()
11    print("[ws] Client connected, user:", ws.state.user_id or "anonymous")
12    subscribed_channels = set()
13 
14    try:
15        while True:
16            frame = await ws.receive_json()
17            action = frame.get("action")
18            channel = frame.get("channel")
19 
20            if action == "SUBSCRIBE" and channel:
21                register_ws_client(channel, ws)
22                subscribed_channels.add(channel)
23                print("[ws] Subscribed to channel:", channel)
24                # REQUIRED. Without this ack the client never starts its catch up.
25                await ws.send_json({"action": "SUBSCRIBED", "channel": channel})
26 
27            if action == "UNSUBSCRIBE" and channel:
28                remove_ws_client(channel, ws)
29                subscribed_channels.discard(channel)
30 
31            if action == "SYNC_REQUEST" and channel:
32                try:
33                    result = await query_changes_since(
34                        channel, frame.get("last_sync_anchor", 0), ws.state.user_id
35                    )
36                    await ws.send_json({"action": "SYNC_COMPLETE", "channel": channel, **result})
37                except Exception as err:
38                    # Report the failure. An empty rows array would be read as
39                    # "confirmed current".
40                    print("[ws] SYNC_REQUEST failed:", err)
41                    await ws.send_json({"action": "SYNC_COMPLETE", "channel": channel, "error": str(err)})
42 
43            if action == "RECONCILE_REQUEST" and channel:
44                local_ids = frame.get("localIds", [])
45                try:
46                    existing = await query_existing_ids(channel, local_ids, ws.state.user_id)
47                    deleted_ids = [] if existing is None else [i for i in local_ids if i not in existing]
48                except Exception as err:
49                    print("[ws] RECONCILE_REQUEST failed:", err)
50                    deleted_ids = []
51                await ws.send_json({"action": "RECONCILE_COMPLETE", "channel": channel, "deletedIds": deleted_ids})
52 
53    except WebSocketDisconnect:
54        for channel in subscribed_channels:
55            remove_ws_client(channel, ws)
56        print("[ws] Client disconnected")

1// Your answer to SYNC_REQUEST. The adapter turns it into ONE outcome.
2{
3  action:  'SYNC_COMPLETE',
4  channel: 'items',
5  rows: [
6    { op: 'UPDATE', id: '662f...', data: { id: '662f...', ...fullRow }, timestamp: 1718000001234 },
7    { op: 'DELETE', id: '771a...', data: { id: '771a...' },             timestamp: 1718000002345 },
8  ],
9  asOf:   1718000099000,   // server clock ms the answer is complete through (optional)
10  tooOld: false,           // true: you cannot answer from this anchor, the client refetches
11  error:  undefined,       // a string: you could not answer, treated as a failure
12}
13 
14// Outcome the adapter reports:
15//   rows applied              applied
16//   rows is empty             confirmed current (a quiet channel is a valid answer)
17//   tooOld is true            too old, the client refetches the channel
18//   error is set, no reply    failed, the gap stays open and the client refetches
19//   any row lost on the way   failed, never claimed as covered
20//
21// NEVER answer a failed query with an empty rows array. That reads as
22// "nothing changed" and the client would believe it is up to date.

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 makes deletions catchable at all. A plain
6// query on the live collection can never tell "deleted" from "never existed."
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 
20// Enforces single state: upserts on this key replace the existing entry.
21FluxChangelogSchema.index({ tableName: 1, rowId: 1 }, { unique: true })
22// What queryChangesSince() scans: one table, one audience, newer than an anchor.
23FluxChangelogSchema.index({ tableName: 1, userId: 1, changedAt: 1 })
24// Cleanup after 14 days. queryChangesSince() refuses anchors older than 13
25// days, so a catch up is never answered from a half purged window.
26FluxChangelogSchema.index({ changedAt: 1 }, { expireAfterSeconds: 60 * 60 * 24 * 14 })
27 
28module.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 catch up 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 event carries no document, so the owner comes from the earlier
39  // entry. If the row was never logged it becomes a global tombstone with
40  // only an id.
41  const owner = ownerId ? String(ownerId) : (existing?.userId ?? null)
42 
43  const payload   = finalOp === 'DELETE' ? { id: cleanId } : normalizeRowData(cleanId, rowData)
44  const changedAt = new Date()
45 
46  await FluxChangelog.findOneAndUpdate(
47    { tableName, rowId: cleanId },
48    { tableName, rowId: cleanId, userId: owner, op: finalOp, rowData: payload, changedAt },
49    { upsert: true, setDefaultsOnInsert: true }
50  )
51 
52  return { ownerId: owner, timestamp: changedAt.getTime() }
53}
54 
55// Answers a catch up for ONE user. Returns { rows, asOf, tooOld }.
56//   - the user receives their own rows plus global rows, never anyone else's
57//   - an anchor older than the retention window, or no anchor at all (0),
58//     is answered tooOld so the client refetches
59//   - more rows than ROW_LIMIT is answered tooOld, never a partial delta
60async function queryChangesSince(tableName, sinceMs, userId = null) {
61  const now  = Date.now()
62  const asOf = now - OVERLAP_MS
63 
64  const parsed = Number(sinceMs)
65  const since  = Number.isFinite(parsed) && parsed > 0 ? Math.floor(parsed) : 0
66 
67  if (since < now - RETENTION_MS) return { rows: [], asOf, tooOld: true }
68 
69  const audience = userId ? [null, String(userId)] : [null]
70 
71  const found = await FluxChangelog.find({
72    tableName,
73    changedAt: { $gt: new Date(since) },
74    userId:    { $in: audience },
75  })
76    .sort({ changedAt: 1, _id: 1 })
77    .limit(ROW_LIMIT + 1)
78    .lean()
79 
80  if (found.length > ROW_LIMIT) return { rows: [], asOf, tooOld: true }
81 
82  const rows = found.map((r) => ({
83    op:        r.op,
84    id:        r.rowId,
85    data:      r.op === 'DELETE' ? { id: r.rowId } : normalizeRowData(r.rowId, r.rowData),
86    timestamp: r.changedAt.getTime(),
87  }))
88 
89  return { rows, asOf, tooOld: false }
90}
91 
92// Which of these ids still exist for THIS user? Returns null when the server
93// cannot answer (unknown table, or a per user table with no user), and the
94// caller must then report NO deletions.
95async function queryExistingIds(tableName, ids, userId = null) {
96  const entry = tables[tableName]
97  if (!entry || !Array.isArray(ids) || ids.length === 0) return null
98  if (entry.ownerField && !userId) return null
99 
100  const validIds = ids.filter((id) => mongoose.isValidObjectId(id))
101  const filter   = { _id: { $in: validIds } }
102  if (entry.ownerField) filter[entry.ownerField] = userId
103 
104  const found = await entry.model.find(filter).select('_id').lean()
105  return found.map((doc) => doc._id.toString())
106}
107 
108module.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 
44# Records one change. Returns {"ownerId", "timestamp"} so the caller broadcasts
45# with the SAME owner and timestamp that catch up will later report.
46async def log_change(table_name: str, row_id: str, op: str, row_data: Any, owner_id: Optional[str] = None):
47    clean_id = str(row_id or "")
48    if not clean_id:
49        return None
50 
51    existing = await changelog.find_one({"tableName": table_name, "rowId": clean_id}, {"op": 1, "userId": 1})
52 
53    # A row the client has not seen yet stays a CREATE even if edited again.
54    final_op = "CREATE" if (op == "UPDATE" and existing and existing["op"] == "CREATE") else op
55 
56    # A delete event carries no document, so the owner comes from the earlier entry.
57    owner = str(owner_id) if owner_id else (existing.get("userId") if existing else None)
58 
59    payload = {"id": clean_id} if final_op == "DELETE" else normalize_row_data(clean_id, row_data)
60    changed_at = datetime.now(timezone.utc)
61 
62    await changelog.update_one(
63        {"tableName": table_name, "rowId": clean_id},
64        {"$set": {"tableName": table_name, "rowId": clean_id, "userId": owner,
65                  "op": final_op, "rowData": payload, "changedAt": changed_at}},
66        upsert=True,
67    )
68    return {"ownerId": owner, "timestamp": to_ms(changed_at)}
69 
70# Answers a catch up for ONE user. Returns {"rows", "asOf", "tooOld"}.
71async def query_changes_since(table_name: str, since_ms: Any, user_id: Optional[str] = None) -> dict:
72    now = int(time.time() * 1000)
73    as_of = now - OVERLAP_MS
74    since = int(since_ms) if isinstance(since_ms, (int, float)) and since_ms > 0 else 0
75 
76    if since < now - RETENTION_MS:
77        return {"rows": [], "asOf": as_of, "tooOld": True}
78 
79    audience = [None, str(user_id)] if user_id else [None]
80    cursor = (
81        changelog.find({
82            "tableName": table_name,
83            "changedAt": {"$gt": datetime.fromtimestamp(since / 1000, tz=timezone.utc)},
84            "userId": {"$in": audience},
85        })
86        .sort([("changedAt", 1), ("_id", 1)])
87        .limit(ROW_LIMIT + 1)
88    )
89    found = await cursor.to_list(length=ROW_LIMIT + 1)
90 
91    if len(found) > ROW_LIMIT:
92        return {"rows": [], "asOf": as_of, "tooOld": True}
93 
94    rows = [{
95        "op": r["op"],
96        "id": r["rowId"],
97        "data": {"id": r["rowId"]} if r["op"] == "DELETE" else normalize_row_data(r["rowId"], r.get("rowData")),
98        "timestamp": to_ms(r["changedAt"]),
99    } for r in found]
100 
101    return {"rows": rows, "asOf": as_of, "tooOld": False}
102 
103# Which of these ids still exist for THIS user? Returns None when the server
104# cannot answer, and the caller must then report NO deletions.
105async def query_existing_ids(table_name: str, ids: List[str], user_id: Optional[str] = None) -> Optional[List[str]]:
106    entry = tables.get(table_name)
107    if not entry or not ids:
108        return None
109    if entry["owner_field"] and not user_id:
110        return None
111 
112    valid = [ObjectId(i) for i in ids if ObjectId.is_valid(i)]
113    query: Dict[str, Any] = {"_id": {"$in": valid}}
114    if entry["owner_field"]:
115        query[entry["owner_field"]] = user_id
116 
117    found = entry["collection"].find(query, {"_id": 1})
118    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 } = 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. Persist to the changelog so catch up and reconcile have data
26    const logged = await logChange('items', rowId, op, fullRow, ownerId)
27    if (!logged) return
28 
29    // 2. Push to live clients. Same owner and timestamp the changelog holds.
30    broadcastToWs(
31      'items',
32      {
33        entity:    'item',
34        id:        rowId,
35        op,
36        data:      op === 'DELETE' ? { id: rowId } : fullRow,
37        timestamp: logged.timestamp,
38        source:    'my_ws_server',
39      },
40      logged.ownerId
41    )
42  })
43 
44  changeStream.on('error', (err) => console.error('[changestream] items error:', err))
45}
46 
47module.exports = { watchItemChanges }

1# server/watchers/items.py
2from db_client import db
3from broadcast import broadcast_to_ws
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. A
22            # delete has no document, so log_change recovers the owner from
23            # the earlier entry.
24            owner_id = str(full_doc["userId"]) if full_doc and full_doc.get("userId") else None
25 
26            logged = await log_change("items", row_id, op, full_row, owner_id)
27            if not logged:
28                continue
29 
30            await broadcast_to_ws(
31                "items",
32                {
33                    "entity": "item",
34                    "id": row_id,
35                    "op": op,
36                    "data": {"id": row_id} if op == "DELETE" else full_row,
37                    "timestamp": logged["timestamp"],
38                    "source": "my_ws_server",
39                },
40                logged["ownerId"],
41            )

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 { authenticate }       = require('./auth')
11const { handleWsConnection } = require('./ws_connection')
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', (ws, req) => {
22  // Identify the user once, on the upgrade request. null means anonymous:
23  // global rows only, no per user rows.
24  ws.userId = authenticate(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, merged collections)
4// or single event objects (live events and catch up rows). Every event now
5// carries a FULL row, and a DELETE carries { id }. Your store's setState
6// must handle both shapes and all three operations.
7 
8function createStoreAdapter(set, get) {
9  return {
10    setState: (data) => {
11      if (!data) return
12 
13      // 1. Array payload: bootstrap, IDB hydration, or a merged collection
14      if (Array.isArray(data)) {
15        set({ items: [...data], isHydrated: true, isLive: true })
16        return
17      }
18 
19      // 2. Object payload: a single event
20      if (typeof data === 'object') {
21        const current = get().items
22 
23        // DELETE
24        if (data.op === 'DELETE' || data.eventType === 'DELETE') {
25          const id = String(data.data?.id ?? data.id ?? data._id ?? '')
26          if (id) set({ items: current.filter((i) => i.id !== id) })
27          return
28        }
29 
30        const rawRecord = data.data ?? data.new ?? data
31        const recordId  = String(rawRecord?.id ?? rawRecord?._id ?? data.id ?? '')
32        if (!recordId) return
33 
34        // UPDATE: the row is complete, merging keeps any local only fields
35        const idx = current.findIndex((i) => i.id === recordId)
36        if (idx !== -1) {
37          const next = [...current]
38          next[idx] = { ...next[idx], ...rawRecord, id: recordId }
39          set({ items: next, isHydrated: true, isLive: true })
40        } else {
41          // CREATE: prepend the new record
42          set({
43            items: [{ id: recordId, ...rawRecord }, ...current],
44            isHydrated: true,
45            isLive: true,
46          })
47        }
48      }
49    }
50  }
51}

1wscat -c "ws://localhost:3001?token=YOUR_TEST_TOKEN"

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

1{"action":"SYNC_REQUEST","channel":"items","last_sync_anchor":1718000000000}