Socket.IO Backend Setup

1// The shape every LIVE event you emit 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// An event 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// createSocketIOAdapter listens for a socket EVENT NAME shaped exactly like
2// "<channel>:<event>". This is the one place your server and your client's
3// flux.register() calls have to agree with each other.
4//
5// A registration like this on the client:
6//   flux.register({ channel: 'items', event: 'UPDATE', ... })
7//
8// means the adapter attaches a listener for the socket event:
9//   'items:UPDATE'
10//
11// GLOBAL rows go to everyone who is connected:
12io.emit('items:UPDATE', event)
13 
14// PER USER rows go only to that user's room (see Step 3 for how a socket
15// joins it):
16io.to('user:' + ownerId).emit('items:UPDATE', event)

1// server/auth.js
2const jwt = require('jsonwebtoken')
3 
4// Socket.IO clients CAN send a token with the handshake:
5//   io(url, { auth: { token } })
6// Verify it in a connection middleware, once, before the socket is accepted.
7function authMiddleware(socket, next) {
8  const token = socket.handshake.auth && socket.handshake.auth.token
9 
10  if (!token) {
11    socket.data.userId = null // anonymous: global rows only
12    return next()
13  }
14 
15  try {
16    socket.data.userId = String(jwt.verify(token, process.env.JWT_SECRET).sub)
17    next()
18  } catch {
19    next(new Error('invalid token')) // rejects the connection
20  }
21}
22 
23module.exports = { authMiddleware }

1# server/auth.py
2import os
3import jwt
4 
5# Socket.IO clients CAN send a token with the handshake: io(url, { auth: { token } }).
6# Verify it in the connect handler, once, before the socket is accepted.
7def verify_token(token):
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 "invalid"

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

1// server/socket_handlers.js
2//
3// Per user rows are routed with rooms. When a socket connects, it joins a room
4// named after its user, and a per user change is emitted to that room only.
5// Global rows are emitted to everyone with io.emit().
6//
7// The adapter ALWAYS emits 'FLUX_SYNC_ANCHOR' on connect, even on a cold
8// start. It is a bulk session signal. The per channel catch up is
9// flux_sync_request below.
10 
11const { queryChangesSince, queryExistingIds } = require('./lib/changelog')
12 
13function registerSocketHandlers(io) {
14  io.on('connection', (socket) => {
15    const userId = socket.data.userId
16 
17    // Join the user's private room so per user rows can reach this socket
18    if (userId) socket.join('user:' + userId)
19 
20    console.log('[socketio] Client connected: ' + socket.id + ', user: ' + (userId || 'anonymous'))
21 
22    socket.on('FLUX_SYNC_ANCHOR', (anchors) => {
23      console.log('[socketio] FLUX_SYNC_ANCHOR received:', anchors)
24      // Optional: server side session tracking or logging.
25    })
26 
27    // Catch up. Server contract: on 'flux_sync_request' { channel, last_sync_anchor }
28    // ALWAYS answer with '<channel>:flux_sync_complete'. Rows are filtered to
29    // this user's own rows plus global rows.
30    socket.on('flux_sync_request', async ({ channel, last_sync_anchor }) => {
31      try {
32        const { rows, asOf, tooOld } = await queryChangesSince(channel, last_sync_anchor, userId)
33        socket.emit(channel + ':flux_sync_complete', { rows, asOf, tooOld })
34        console.log('[socketio] flux_sync_request "' + channel + '" sent ' + rows.length + ' row(s), tooOld: ' + tooOld)
35      } catch (err) {
36        // Report the failure. An empty rows array here would be read as
37        // "confirmed current" and the client would skip a refetch it needs.
38        console.error('[socketio] flux_sync_request failed for "' + channel + '":', err)
39        socket.emit(channel + ':flux_sync_complete', { error: String((err && err.message) || err) })
40      }
41    })
42 
43    // Reconciliation. Server contract: on 'flux_reconcile_request' { channel, localIds }
44    // emit '<channel>:flux_reconcile_complete' with { deletedIds }.
45    socket.on('flux_reconcile_request', async ({ channel, localIds }) => {
46      try {
47        const ids = Array.isArray(localIds) ? localIds : []
48        const existingIds = await queryExistingIds(channel, ids, userId)
49        // null means "cannot answer", so report no deletions rather than all.
50        const deletedIds = existingIds === null ? [] : ids.filter((id) => !existingIds.includes(id))
51 
52        socket.emit(channel + ':flux_reconcile_complete', { deletedIds })
53      } catch (err) {
54        console.error('[socketio] flux_reconcile_request failed for "' + channel + '":', err)
55        socket.emit(channel + ':flux_reconcile_complete', { deletedIds: [] })
56      }
57    })
58 
59    socket.on('disconnect', () => {
60      console.log('[socketio] Client disconnected: ' + socket.id)
61    })
62  })
63}
64 
65module.exports = { registerSocketHandlers }

1# server/socket_handlers.py
2from auth import verify_token
3from lib.changelog import query_changes_since, query_existing_ids
4 
5# Per user rows are routed with rooms. A socket joins a room named after its
6# user, and a per user change is emitted to that room only. Global rows are
7# emitted to everyone.
8 
9def register_handlers(sio):
10    @sio.on("connect")
11    async def on_connect(sid, environ, auth):
12        token = auth.get("token") if isinstance(auth, dict) else None
13        user_id = verify_token(token)
14 
15        if user_id == "invalid":
16            raise ConnectionRefusedError("invalid token")  # rejects the connection
17 
18        await sio.save_session(sid, {"user_id": user_id})
19        if user_id:
20            await sio.enter_room(sid, "user:" + user_id)
21        print("[socketio] Client connected:", sid, "user:", user_id or "anonymous")
22 
23    @sio.on("FLUX_SYNC_ANCHOR")
24    async def on_sync_anchor(sid, anchors):
25        print("[socketio] FLUX_SYNC_ANCHOR received:", anchors)
26 
27    @sio.on("flux_sync_request")
28    async def on_sync_request(sid, data):
29        channel = data.get("channel")
30        anchor = data.get("last_sync_anchor", 0)
31        session = await sio.get_session(sid)
32        try:
33            result = await query_changes_since(channel, anchor, session.get("user_id"))
34            await sio.emit(channel + ":flux_sync_complete", result, to=sid)
35        except Exception as err:
36            # Report the failure. An empty rows array would be read as
37            # "confirmed current".
38            print("[socketio] flux_sync_request failed for", channel, err)
39            await sio.emit(channel + ":flux_sync_complete", {"error": str(err)}, to=sid)
40 
41    @sio.on("flux_reconcile_request")
42    async def on_reconcile_request(sid, data):
43        channel = data.get("channel")
44        local_ids = data.get("localIds", [])
45        session = await sio.get_session(sid)
46        try:
47            existing = await query_existing_ids(channel, local_ids, session.get("user_id"))
48            # None means "cannot answer", so report no deletions rather than all.
49            deleted = [] if existing is None else [i for i in local_ids if i not in existing]
50        except Exception as err:
51            print("[socketio] flux_reconcile_request failed for", channel, err)
52            deleted = []
53        await sio.emit(channel + ":flux_reconcile_complete", {"deletedIds": deleted}, to=sid)
54 
55    @sio.on("disconnect")
56    async def on_disconnect(sid):
57        print("[socketio] Client disconnected:", sid)

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 makes deletions catchable. A plain query on the
6// 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 
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 catch up 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 emits
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 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 catch up 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/watchers/items.js
2const Item = require('../models/Item')
3const { logChange } = require('../lib/changelog')
4 
5function watchItemChanges(io) {
6  const changeStream = Item.watch([], { fullDocument: 'updateLookup' })
7 
8  changeStream.on('change', async (change) => {
9    const { operationType, documentKey, fullDocument } = change
10    const opMap = { insert: 'CREATE', update: 'UPDATE', replace: 'UPDATE', delete: 'DELETE' }
11    const op = opMap[operationType]
12    if (!op) return
13 
14    const rowId = documentKey._id.toString()
15 
16    // Always the FULL row. updateLookup returns the whole document even for
17    // a partial update, so the wire never carries a partial record.
18    const fullRow = fullDocument ? { ...fullDocument, id: rowId } : { id: rowId }
19 
20    // The owner of a per user row. null for a global collection. A delete has
21    // no document, so logChange() recovers the owner from the earlier entry.
22    const ownerId = fullDocument?.userId ? String(fullDocument.userId) : null
23 
24    // 1. Persist to the changelog so catch up and reconcile have data
25    const logged = await logChange('items', rowId, op, fullRow, ownerId)
26    if (!logged) return
27 
28    const event = {
29      entity:    'item',
30      id:        rowId,
31      op,
32      data:      op === 'DELETE' ? { id: rowId } : fullRow,
33      timestamp: logged.timestamp,
34      source:    'my_socketio_server',
35    }
36 
37    // 2. Emit. Same owner and timestamp the changelog holds.
38    if (logged.ownerId) {
39      io.to('user:' + logged.ownerId).emit('items:UPDATE', event) // PER USER
40    } else {
41      io.emit('items:UPDATE', event)                              // GLOBAL
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 lib.changelog import log_change, normalize_row_data
4 
5OP_MAP = {"insert": "CREATE", "update": "UPDATE", "replace": "UPDATE", "delete": "DELETE"}
6 
7async def watch_item_changes(sio):
8    async with db.items.watch(full_document="updateLookup") as stream:
9        async for change in stream:
10            op = OP_MAP.get(change["operationType"])
11            if not op:
12                continue
13 
14            row_id = str(change["documentKey"]["_id"])
15            full_doc = change.get("fullDocument")
16 
17            # Always the FULL row. update_lookup returns the whole document.
18            full_row = normalize_row_data(row_id, full_doc) if full_doc else {"id": row_id}
19 
20            # The owner of a per user row. None for a global collection.
21            owner_id = str(full_doc["userId"]) if full_doc and full_doc.get("userId") else None
22 
23            logged = await log_change("items", row_id, op, full_row, owner_id)
24            if not logged:
25                continue
26 
27            event = {
28                "entity": "item",
29                "id": row_id,
30                "op": op,
31                "data": {"id": row_id} if op == "DELETE" else full_row,
32                "timestamp": logged["timestamp"],
33                "source": "my_socketio_server",
34            }
35 
36            # Same owner and timestamp the changelog holds.
37            if logged["ownerId"]:
38                await sio.emit("items:UPDATE", event, room="user:" + logged["ownerId"])  # PER USER
39            else:
40                await sio.emit("items:UPDATE", event)                                    # GLOBAL

1// server/index.js
2require('dotenv').config()
3 
4const express = require('express')
5const http = require('http')
6const { Server: SocketIOServer } = require('socket.io')
7const { connectDb } = require('./db')
8 
9const { authMiddleware } = require('./auth')
10const { registerSocketHandlers } = require('./socket_handlers')
11const { watchItemChanges } = require('./watchers/items')
12 
13const app = express()
14const server = http.createServer(app)
15 
16// Socket.IO CORS must be set on the Socket.IO server constructor. Express's
17// cors() middleware does NOT cover the Engine.IO handshake.
18const io = new SocketIOServer(server, {
19  cors: {
20    origin: process.env.CLIENT_ORIGIN?.split(',').map((o) => o.trim()) || [],
21  },
22})
23 
24app.use(express.json())
25 
26// Verify the handshake token before any socket is accepted
27io.use(authMiddleware)
28 
29// FLUX_SYNC_ANCHOR, flux_sync_request, flux_reconcile_request
30registerSocketHandlers(io)
31 
32const PORT = process.env.PORT || 3001
33 
34connectDb().then(() => {
35  // Start listening for data changes BEFORE accepting connections.
36  watchItemChanges(io)
37 
38  server.listen(PORT, '0.0.0.0', () => {
39    console.log('[server] Socket.IO ready on http://0.0.0.0:' + PORT)
40  })
41})

1# server/main.py
2import asyncio
3import socketio
4from fastapi import FastAPI
5from socket_handlers import register_handlers
6from watchers.items import watch_item_changes
7 
8sio = socketio.AsyncServer(
9    async_mode="asgi",
10    cors_allowed_origins=["https://app.example.com"],  # list your real client origins
11)
12 
13app = FastAPI()
14asgi_app = socketio.ASGIApp(sio, other_asgi_app=app)
15 
16register_handlers(sio)
17 
18@app.on_event("startup")
19async def startup():
20    asyncio.create_task(watch_item_changes(sio))
21 
22# uvicorn main:asgi_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 must
6// 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}

1// test_connection.js, run with: node test_connection.js
2const { io } = require('socket.io-client')
3 
4const socket = io('http://localhost:3001', {
5  auth: { token: process.env.TEST_TOKEN }, // a real user's token
6})
7 
8socket.on('connect', () => {
9  console.log('[test] connected:', socket.id)
10 
11  // Anchor within the last 13 days. 0 should come back tooOld.
12  socket.emit('flux_sync_request', {
13    channel: 'items',
14    last_sync_anchor: Date.now() - 60 * 60 * 1000,
15  })
16})
17 
18socket.on('items:flux_sync_complete', (payload) => {
19  console.log('[test] sync complete:', {
20    rows: payload.rows ? payload.rows.length : 0,
21    asOf: payload.asOf,
22    tooOld: payload.tooOld,
23    error: payload.error,
24  })
25})
26 
27socket.on('items:UPDATE', (event) => {
28  console.log('[test] live event received:', event.id, event.op)
29})
30 
31socket.on('connect_error', (err) => console.log('[test] rejected:', err.message))
32socket.on('disconnect', () => console.log('[test] disconnected'))

1# test_connection.py, run with: TEST_TOKEN=... python test_connection.py
2import os
3import time
4import socketio
5 
6sio = socketio.Client()
7 
8@sio.event
9def connect():
10    print("[test] connected:", sio.sid)
11    # Anchor within the last 13 days. 0 should come back tooOld.
12    sio.emit("flux_sync_request", {
13        "channel": "items",
14        "last_sync_anchor": int(time.time() * 1000) - 60 * 60 * 1000,
15    })
16 
17@sio.on("items:flux_sync_complete")
18def on_sync_complete(data):
19    print("[test] sync complete:", {
20        "rows": len(data.get("rows", [])),
21        "asOf": data.get("asOf"),
22        "tooOld": data.get("tooOld"),
23        "error": data.get("error"),
24    })
25 
26@sio.on("items:UPDATE")
27def on_update(event):
28    print("[test] live event received:", event["id"], event["op"])
29 
30@sio.event
31def disconnect():
32    print("[test] disconnected")
33 
34sio.connect("http://localhost:3001", auth={"token": os.environ.get("TEST_TOKEN")})
35sio.wait()