WebSocket

1import { createFlux } from '@tsworldtech/flux'
2import { createWebSocketAdapter } from '@tsworldtech/flux_websocket'
3 
4const flux = createFlux({
5  adapter: createWebSocketAdapter({
6    url: process.env.NEXT_PUBLIC_WS_URL || 'ws://localhost:3001',
7 
8    // Optional: passed straight to the WebSocket constructor.
9    // protocols: ['my-protocol'],
10 
11    // Optional: how far back (ms) your SYNC_REQUEST handler can reliably
12    // report ALL changes, deletions included. Leave unset if it can't.
13    deltaRetentionMs: 14 * 24 * 60 * 60 * 1000,
14  }),
15})

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// Sends a SUBSCRIBE frame for the 'items' channel the moment this
16// registration runs, and an UNSUBSCRIBE frame the moment it's torn down.

1// 1. Sent automatically the moment flux.register() runs:
2{ "action": "SUBSCRIBE", "channel": "items", "event": "UPDATE" }
3 
4// ⚠️ REQUIRED SERVER ACKNOWLEDGMENT: Backend must respond with a SUBSCRIBED frame.
5// Without this ack, delta catch-up and cache clock refreshes are never triggered:
6{ "action": "SUBSCRIBED", "channel": "items" }
7 
8// 2. Sent automatically when flux.unregister() runs or component unmounts:
9{ "action": "UNSUBSCRIBE", "channel": "items", "event": "UPDATE" }

1// ── Authoritative Sync Handshake ──────────────────────────────────────────
2// Sent by adapter on channel subscription to pull changes since last_sync_anchor:
3{ "action": "SYNC_REQUEST", "channel": "items", "last_sync_anchor": 1718000000000 }
4 
5// Backend ALWAYS answers, even when nothing changed:
6{
7  "action": "SYNC_COMPLETE",
8  "channel": "items",
9  "rows": [
10    // every row needs an id and a server-clock timestamp
11    { "op": "UPDATE", "data": { "id": "662f...", "price": 42 }, "timestamp": 1718000001234 },
12    // a delete needs only the id
13    { "op": "DELETE", "id": "662g...", "timestamp": 1718000002345 }
14  ],
15  "asOf": 1718000003000          // optional: server-clock ms the answer is complete through
16}
17 
18// The other valid answers:
19//   { "action": "SYNC_COMPLETE", "channel": "items", "rows": [], "asOf": ... }   nothing changed
20//   { "action": "SYNC_COMPLETE", "channel": "items", "tooOld": true }            Flux refetches
21//   { "action": "SYNC_COMPLETE", "channel": "items", "error": "message" }        Flux refetches
22 
23// ── Deletion Reconciliation Handshake ─────────────────────────────────────
24// Sent by adapter to verify local IDs that might have been deleted offline:
25{ "action": "RECONCILE_REQUEST", "channel": "items", "localIds": ["id1", "id2"] }
26 
27// Backend returns missing IDs:
28{ "action": "RECONCILE_COMPLETE", "channel": "items", "deletedIds": ["id2"] }

1// Flux needs to know whether a catch-up PROVED the data is current.
2// The adapter reads your SYNC_COMPLETE frame 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 SYNC_COMPLETE within 30 seconds      -> failed   (Flux refetches)
11//   socket closed 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 incoming message needs "channel" at the top level (it is the
2// only routing key), plus the row's id and a server-clock timestamp:
3{
4  "channel":   "items",          // required. Routes the frame to the registration
5  "id":        "662f...",        // required. Or put it in data.id / data._id
6  "timestamp": 1718000001234,    // required. Your server's Date.now() for the change
7  "data":      { /* ...row or sparse diff */ },   // required for CREATE / UPDATE
8  "op":        "UPDATE"          // optional. Defaults to UPDATE; send "DELETE" explicitly
9}
10 
11// "entity" and "source" are optional. Flux stamps the channel name itself.
12//
13// A frame with no id, no finite timestamp, or no row (on a create/update)
14// is dropped and logged. Flux never invents an id for it.
15//
16// Routing happens on "channel" alone — the adapter keeps exactly one
17// callback per channel. Unlike the Socket.IO adapter, "event" inside the
18// payload isn't used for routing; only the outer "channel" key is.