Socket.IO Backend Setup

1// The shape every event you emit 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, Socket.IO).
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// createSocketIOAdapter listens for a socket EVENT NAME shaped exactly
2// like "<channel>:<event>" — this is the one place your server and your
3// client's 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// Your server emits under that exact string. io.emit() reaches every
12// connected socket project-wide; only clients listening for this exact
13// event name act on it. Unrelated event names cost the receiving client
14// nothing.
15io.emit('items:UPDATE', event)

1// server/socket-handlers.js
2//
3// The adapter ALWAYS emits 'FLUX_SYNC_ANCHOR' on every connect(),
4// even on a cold start (all-zero anchors). This is a secondary/bulk
5// signal — the primary per-channel catch-up handshake is flux_sync_request
6// (see Step 5 below).
7//
8// FIXED: previously gated behind hasAnchors, only emitting when at least
9// one channel had a non-zero anchor. That's inconsistent with sync() below,
10// which ALWAYS emits flux_sync_request with last_sync_anchor regardless of
11// value. FLUX_SYNC_ANCHOR now always fires so the server has signal that
12// a session connected, but the actual row backfill happens per-channel.
13 
14const { queryChangesSince, queryExistingIds } = require('../lib/changelog')
15 
16function registerSocketHandlers(io) {
17  io.on('connection', (socket) => {
18    console.log('[socketio] Client connected:', socket.id)
19 
20    socket.on('FLUX_SYNC_ANCHOR', (anchors) => {
21      console.log('[socketio] FLUX_SYNC_ANCHOR received:', anchors)
22      // Optional: use for server-side session tracking or logging.
23      // The authoritative per-channel catch-up is flux_sync_request below.
24    })
25 
26    // ── Primary catch-up: flux_sync_request ───────────────────────────
27    //
28    // Server contract: on 'flux_sync_request' { channel, last_sync_anchor },
29    // emit '${channel}:flux_sync_complete' with { rows: Array<{
30    //   op, data (or row_data), timestamp (or changed_at), id (or row_id)
31    // }> }.
32    socket.on('flux_sync_request', async ({ channel, last_sync_anchor }) => {
33      try {
34        const rows = await queryChangesSince(channel, last_sync_anchor)
35        socket.emit(`${channel}:flux_sync_complete`, { rows })
36        console.log(
37          `[socketio] flux_sync_request "${channel}" since ${last_sync_anchor} — sent ${rows.length} row(s)`
38        )
39      } catch (err) {
40        console.error(`[socketio] flux_sync_request failed for "${channel}":`, err)
41        socket.emit(`${channel}:flux_sync_complete`, { rows: [] })
42      }
43    })
44 
45    // ── Reconciliation ────────────────────────────────────────────────
46    //
47    // Server contract: on 'flux_reconcile_request' { channel, localIds },
48    // emit '${channel}:flux_reconcile_complete' with { deletedIds: string[] }.
49    socket.on('flux_reconcile_request', async ({ channel, localIds }) => {
50      try {
51        const ids = Array.isArray(localIds) ? localIds : []
52        const existingIds = await queryExistingIds(channel, ids)
53        const deletedIds = ids.filter((id) => !existingIds.includes(id))
54 
55        socket.emit(`${channel}:flux_reconcile_complete`, { deletedIds })
56      } catch (err) {
57        console.error(`[socketio] flux_reconcile_request failed for "${channel}":`, err)
58        socket.emit(`${channel}:flux_reconcile_complete`, { deletedIds: [] })
59      }
60    })
61 
62    socket.on('disconnect', () => {
63      console.log('[socketio] Client disconnected:', socket.id)
64    })
65  })
66}
67 
68module.exports = { registerSocketHandlers }

1# server/socket_handlers.py
2import time
3from lib.changelog import query_changes_since, query_existing_ids
4 
5def register_handlers(sio):
6    @sio.on("connect")
7    async def on_connect(sid, environ):
8        print(f"[socketio] Client connected: {sid}")
9 
10    @sio.on("FLUX_SYNC_ANCHOR")
11    async def on_sync_anchor(sid, anchors):
12        print(f"[socketio] FLUX_SYNC_ANCHOR received: {anchors}")
13 
14    @sio.on("flux_sync_request")
15    async def on_sync_request(sid, data):
16        channel = data.get("channel")
17        anchor = data.get("last_sync_anchor", 0)
18        try:
19            rows = await query_changes_since(channel, anchor)
20            await sio.emit(f"{channel}:flux_sync_complete", {"rows": rows}, to=sid)
21            print(f'[socketio] flux_sync_request "{channel}" since {anchor} — sent {len(rows)} row(s)')
22        except Exception as err:
23            print(f'[socketio] flux_sync_request failed for "{channel}":', err)
24            await sio.emit(f"{channel}:flux_sync_complete", {"rows": []}, to=sid)
25 
26    @sio.on("flux_reconcile_request")
27    async def on_reconcile_request(sid, data):
28        channel = data.get("channel")
29        local_ids = data.get("localIds", [])
30        try:
31            existing = await query_existing_ids(channel, local_ids)
32            deleted_ids = [i for i in local_ids if i not in existing]
33            await sio.emit(f"{channel}:flux_reconcile_complete", {"deletedIds": deleted_ids}, to=sid)
34        except Exception as err:
35            print(f'[socketio] flux_reconcile_request failed for "{channel}":', err)
36            await sio.emit(f"{channel}:flux_reconcile_complete", {"deletedIds": []}, to=sid)
37 
38    @sio.on("disconnect")
39    async def on_disconnect(sid):
40        print(f"[socketio] Client disconnected: {sid}")

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 all three transports (WS, SSE, Socket.IO).
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/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, updateDescription } = 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    // Extract sparse fields for UPDATEs so the wire payload is minimal
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. Persist to changelog so sync() and reconcile() have data
29    await logChange('items', rowId, op, rowData, updatedFields)
30 
31    // 2. Build wire payload — sparse when possible
32    const wireData = op === 'UPDATE' && updatedFields && Object.keys(updatedFields).length > 0
33      ? { id: rowId, ...updatedFields }
34      : rowData
35 
36    const event = {
37      entity:    'item',
38      id:        rowId,
39      op,
40      data:      wireData,
41      timestamp: Date.now(),
42      source:    'flux-test-server',
43    }
44 
45    // 3. Broadcast live event — Socket.IO dispatches by event name
46    io.emit('items:UPDATE', event)
47  })
48 
49  changeStream.on('error', (err) => console.error('[changestream] items error:', err))
50}
51 
52module.exports = { watchItemChanges }

1# server/watchers/items.py
2from db_client import db
3from lib.changelog import log_change
4import time
5 
6OP_MAP = {"insert": "CREATE", "update": "UPDATE", "replace": "UPDATE", "delete": "DELETE"}
7 
8async def watch_item_changes(sio):
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            full_doc = change.get("fullDocument")
16            update_desc = change.get("updateDescription")
17            row_id = str(change["documentKey"]["_id"])
18 
19            updated_fields = None
20            if change["operationType"] == "update" and update_desc:
21                updated_fields = dict(update_desc.get("updatedFields", {}))
22                updated_fields.pop("updatedAt", None)
23                updated_fields.pop("__v", None)
24 
25            row_data = {**full_doc, "id": str(full_doc["_id"])} if full_doc else {"id": row_id}
26            await log_change("items", row_id, op, row_data, updated_fields)
27 
28            wire_data = updated_fields if (op == "UPDATE" and updated_fields) else row_data
29            event = {
30                "entity": "item",
31                "id": row_id,
32                "op": op,
33                "data": wire_data,
34                "timestamp": int(time.time() * 1000),
35                "source": "flux-test-server",
36            }
37 
38            await sio.emit("items:UPDATE", event)

1// server/index.js
2require('dotenv').config()
3 
4const express = require('express')
5const http = require('http')
6const cors = require('cors')
7const { Server: SocketIOServer } = require('socket.io')
8const { connectDb } = require('./db')
9 
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 —
17// Express's 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// Register all Socket.IO event handlers (FLUX_SYNC_ANCHOR,
27// flux_sync_request, flux_reconcile_request)
28registerSocketHandlers(io)
29 
30const PORT = process.env.PORT || 3001
31 
32connectDb().then(() => {
33  // Start listening for data changes BEFORE accepting connections.
34  watchItemChanges(io)
35 
36  server.listen(PORT, '0.0.0.0', () => {
37    console.log(`[server] Socket.IO ready on http://0.0.0.0:${PORT}`)
38  })
39})

1# server/main.py
2import asyncio
3import socketio
4from fastapi import FastAPI
5from fastapi.middleware.cors import CORSMiddleware
6from socket_handlers import register_handlers
7from watchers.items import watch_item_changes
8 
9sio = socketio.AsyncServer(
10    async_mode="asgi",
11    cors_allowed_origins="*",  # scope this down in production
12)
13 
14app = FastAPI()
15asgi_app = socketio.ASGIApp(sio, other_asgi_app=app)
16 
17register_handlers(sio)
18 
19@app.on_event("startup")
20async def startup():
21    asyncio.create_task(watch_item_changes(sio))
22 
23# 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) or single delta
4// objects (live events & sync complete). Your store's setState must
5// handle both shapes and operations.
6 
7function createStoreAdapter(set, get) {
8  return {
9    setState: (data) => {
10      if (!data) return
11 
12      // ── 1. Array payload — bootstrap 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 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}

1// test-connection.js — run with: node test-connection.js
2const { io } = require('socket.io-client')
3 
4const socket = io('http://localhost:3001')
5 
6socket.on('connect', () => {
7  console.log('[test] connected:', socket.id)
8 
9  // Trigger a sync manually to verify the handshake
10  socket.emit('flux_sync_request', {
11    channel: 'items',
12    last_sync_anchor: 0,
13  })
14})
15 
16socket.on('items:flux_sync_complete', ({ rows }) => {
17  console.log('[test] sync complete — rows:', rows.length)
18})
19 
20socket.on('items:UPDATE', (event) => {
21  console.log('[test] live event received:', event.id, event.op)
22})
23 
24socket.on('disconnect', () => {
25  console.log('[test] disconnected')
26})

1# test_connection.py — run with: python test_connection.py
2import socketio
3 
4sio = socketio.Client()
5 
6@sio.event
7def connect():
8    print("[test] connected:", sio.sid)
9    sio.emit("flux_sync_request", {
10        "channel": "items",
11        "last_sync_anchor": 0,
12    })
13 
14@sio.on("items:flux_sync_complete")
15def on_sync_complete(data):
16    print("[test] sync complete — rows:", len(data.get("rows", [])))
17 
18@sio.on("items:UPDATE")
19def on_update(event):
20    print("[test] live event received:", event["id"], event["op"])
21 
22@sio.event
23def disconnect():
24    print("[test] disconnected")
25 
26sio.connect("http://localhost:3001")
27sio.wait()