Socket.IO

1import { io } from 'socket.io-client'
2import { createFlux } from '@tsworldtech/flux'
3import { createSocketIOAdapter } from '@tsworldtech/flux_socketio'
4 
5const socket = io(process.env.NEXT_PUBLIC_SOCKET_URL!, {
6  autoConnect: false, // let Flux control when the connection opens
7})
8 
9const flux = createFlux({
10  adapter: createSocketIOAdapter(socket, {
11    // Optional: how far back (ms) your flux_sync_request handler can
12    // reliably report ALL changes, deletions included. Leave unset if
13    // it can't.
14    deltaRetentionMs: 14 * 24 * 60 * 60 * 1000,
15  }),
16})

1flux.register({
2  store: useItemsStore,
3  channel: 'items',
4  event: 'UPDATE',
5  idbKey: 'items:all',
6  ttl: 'medium',
7  ingestionType: 'COLLECTION_ALL',
8  diffBeforeUpdate: true,
9  hydrateState: (store, data) => {
10    // store.getState().setState handles both array hydrations and single delta frames
11    store.getState().setState(data)
12  },
13})
14 
15// This registration listens for the live socket event named 'items:UPDATE' —
16// your server needs to emit under that exact string. See below.

1// channel + event combine into one socket event name: "channel:event"
2flux.register({ channel: 'orders',  event: 'CREATE', /* ... */ })
3flux.register({ channel: 'orders',  event: 'UPDATE', /* ... */ })
4flux.register({ channel: 'jobs',    event: '*',      /* ... */ }) // all ops
5 
6// Your server side emits must match these strings exactly:
7//   io.emit('orders:CREATE', event)
8//   io.emit('orders:UPDATE', event)
9//   io.emit('jobs:*', event)  // or emit under the specific op that occurred

1// ── 1. Secondary Bulk Signal (Emitted by adapter on connect) ─────────────
2// Emits full anchor map on connect: { [channel]: unixMs }
3socket.emit('FLUX_SYNC_ANCHOR', { items: 1718000000000 })
4 
5// ── 2. Primary Per-Channel Catch-up Handshake ─────────────────────────────
6// Adapter requests deltas for subscribed channel:
7socket.emit('flux_sync_request', { channel: 'items', last_sync_anchor: 1718000000000 })
8 
9// Server ALWAYS answers on the channel-scoped event:
10socket.emit('items:flux_sync_complete', {
11  rows: [
12    // every row needs an id and a server-clock timestamp
13    { op: 'UPDATE', data: { id: '662f...', price: 42 }, timestamp: 1718000001234 },
14    // a delete needs only the id
15    { op: 'DELETE', id: '662g...', timestamp: 1718000002345 },
16  ],
17  asOf: 1718000003000,   // optional: server-clock ms the answer is complete through
18})
19 
20// The other valid answers:
21//   { rows: [], asOf }      nothing changed (this is what confirms a quiet channel)
22//   { tooOld: true }        you cannot answer from that anchor; Flux refetches
23//   { error: 'message' }    you could not answer; Flux refetches
24 
25// ── 3. Deletion Reconciliation Handshake ───────────────────────────
26// Adapter verifies cached local IDs against database state:
27socket.emit('flux_reconcile_request', { channel: 'items', localIds: ['id1', 'id2'] })
28 
29// Server returns IDs that no longer exist:
30socket.emit('items:flux_reconcile_complete', { deletedIds: ['id2'] })

1// Flux needs to know whether a catch-up PROVED the data is current.
2// The adapter reads your flux_sync_complete reply and reports one of:
3//
4//   rows delivered and applied              -> applied
5//   rows: []                                -> confirmed-current
6//   tooOld: true                            -> too-old  (Flux refetches)
7//   error: '...'                            -> failed   (Flux refetches)
8//   a row had no id, no finite timestamp,
9//     or no data (and was not a delete)     -> failed   (Flux refetches)
10//   no reply within 30 seconds              -> failed   (Flux refetches)
11//   connection dropped or channel removed   -> failed
12//
13// Flux also downgrades applied / confirmed-current to failed if a row was
14// dropped before it reached the store, or could not be written to disk.
15//
16// asOf is optional. With it, Flux records the channel as current through
17// that server-clock time. Without it, Flux uses the moment the catch-up
18// started (server clock) minus a 5-second safety margin, so it never
19// trusts the device clock.
20//
21// "failed" is never a data loss. It means "I can't prove this channel is
22// current", so Flux keeps the gap open and refetches.

1// Every live frame your server emits needs this shape:
2{
3  id:        string   // required. Or put it in data.id / data._id
4  timestamp: number   // required. Your server's Date.now() when the change happened
5  data:      any      // required for CREATE / UPDATE: the full row or a sparse diff
6                      // (not needed for DELETE)
7  op?:       'CREATE' | 'UPDATE' | 'DELETE'
8                      // optional. A missing op means UPDATE, so send 'DELETE'
9                      // explicitly for deletions ('INSERT' is read as 'CREATE')
10  entity?:   string   // optional. Flux stamps the channel name
11  source?:   string   // optional
12}
13 
14// A frame with no id, no finite timestamp, or no row (on a create/update)
15// is dropped and logged. Flux never invents an id for it.