SSE

1import { createFlux } from '@tsworldtech/flux'
2import { createSseAdapter } from '@tsworldtech/flux_sse'
3 
4const flux = createFlux({
5  adapter: createSseAdapter({
6    // ONE endpoint for every channel. Flux opens a single stream to it.
7    url: `${process.env.NEXT_PUBLIC_API_URL}/api/stream`,
8 
9    // Optional POST endpoint for deletion reconciliation (SSE is one-way).
10    // A string, or a function that returns one per channel.
11    reconcileUrl: (channel) => `${process.env.NEXT_PUBLIC_API_URL}/api/reconcile/${channel}`,
12 
13    // Optional: how far back (ms) your backfill can reliably report ALL
14    // changes, deletions included. Leave unset if it can't.
15    deltaRetentionMs: 14 * 24 * 60 * 60 * 1000,
16 
17    // Optional: passed straight to the EventSource constructor,
18    // e.g. { withCredentials: true } for cookie auth across origins.
19    // options: { withCredentials: true },
20  }),
21})

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 
15flux.register({
16  store: useOrdersStore,
17  channel: 'orders',
18  event: 'UPDATE',
19  idbKey: 'orders:all',
20  ttl: 'medium',
21  ingestionType: 'COLLECTION_ALL',
22})
23 
24// Both channels ride ONE EventSource:
25//
26//   GET /api/stream?channels={"items":0,"orders":0}
27//
28// Rows for each channel arrive on that stream tagged with the channel
29// name, and Flux routes each one to the right store.

1// What the adapter requests. "channels" is URL-encoded JSON:
2//
3//   GET {url}?channels={"items":1767225600000,"orders":0}
4//
5// Each value is that channel's sync anchor: the server-clock time (ms)
6// the client already has data through. 0 means "send everything".
7//
8// If your url already has a query string, the adapter appends with "&".
9// That is where an auth token goes (see Authentication below):
10//
11//   url: `/api/stream?token=${token}`
12//   -> /api/stream?token=abc&channels={"items":0}

1// THE SERVER CONTRACT
2//
3// 1. Read "channels" from the query string. It is JSON:
4//      { "channelName": anchorMs, ... }
5//
6// 2. For every channel, in any order, do:
7//      a. Send every change newer than that channel's anchor (the backfill).
8//      b. Then send ONE control event, "flux:backfill", to say the backfill
9//         for that channel is finished.
10//      c. From then on, send live changes for that channel.
11//
12// 3. Every change is an SSE event:
13//      - the event NAME is the same value as "event" in flux.register()
14//        (for example 'UPDATE')
15//      - the data is JSON containing:
16//          channel     the channel name                       (required)
17//          timestamp   server-clock ms of the change           (required)
18//                      ("changed_at" as a date string also works)
19//          id          the row's stable identifier             (required)
20//                      Flux looks at data.id, data._id, then a
21//                      top-level row_id / id.
22//          op          'CREATE' | 'UPDATE' | 'DELETE'          (optional)
23//                      A missing op means UPDATE, so send 'DELETE'
24//                      explicitly for deletions. 'INSERT' is read as 'CREATE'.
25//          data        the row (or row_data). If you leave it out, the
26//                      message itself is treated as the row. A DELETE
27//                      needs only the id.
28//
29//    A message with no "channel" cannot be routed and is dropped.
30//
31//    A message with no id, no finite timestamp, or (for a create/update)
32//    no row is treated as LOST. If it arrives during a backfill, that
33//    channel's catch-up reports failed and Flux refetches the channel.
34//    Flux never invents an id for a row, because a made-up id would turn
35//    a delete into a silent no-op.
36//
37// 4. The backfill marker is an SSE event named "flux:backfill":
38//      { "channel": "items", "asOf": 1767225600000 }
39//    asOf is the server-clock time the backfill is complete through.
40//    It is optional but recommended (see "What catch-up proves").
41//
42//    If you cannot answer from that anchor (it is older than you keep
43//    history), send instead:
44//      { "channel": "items", "tooOld": true }
45//    and Flux refetches that channel from scratch.
46//
47// 5. The backfill must include deletions. If it cannot, leave
48//    deltaRetentionMs unset, and deletion catch-up is best-effort.

1// A reference route (Express). The two helpers are placeholders:
2// subscribeToChanges() and loadChangesSince() are YOUR storage layer.
3// Every change they return needs: id, event, op, timestamp (server-clock
4// ms) and data (the row; not needed for a delete).
5import type { Request, Response } from 'express'
6 
7app.get('/api/stream', async (req: Request, res: Response) => {
8  const anchors: Record<string, number> = JSON.parse(String(req.query.channels ?? '{}'))
9  const channels = Object.keys(anchors)
10 
11  res.set({
12    'Content-Type':  'text/event-stream',
13    'Cache-Control': 'no-cache, no-transform',
14    Connection:      'keep-alive',
15  })
16  res.flushHeaders()
17 
18  const send = (event: string, data: unknown) =>
19    res.write('event: ' + event + '\n' + 'data: ' + JSON.stringify(data) + '\n\n')
20 
21  // Subscribe to live changes FIRST and hold them until the backfill is
22  // done, so a change that happens during the backfill is never lost.
23  const held: Array<() => void> = []
24  let backfilling = true
25 
26  const unsubscribe = subscribeToChanges(channels, (change) => {
27    const emit = () =>
28      send(change.event, {
29        channel:   change.channel,
30        id:        change.id,
31        timestamp: change.timestamp,
32        op:        change.op,
33        data:      change.row,
34      })
35    if (backfilling) held.push(emit)
36    else emit()
37  })
38 
39  // Comment lines keep proxies and load balancers from closing an idle stream.
40  const ping = setInterval(() => res.write(': ping\n\n'), 25_000)
41 
42  let closed = false
43  req.on('close', () => {
44    closed = true
45    clearInterval(ping)
46    unsubscribe()
47  })
48 
49  for (const channel of channels) {
50    if (closed) return
51 
52    // loadChangesSince -> { tooOld: boolean, rows: [...], asOf: number }
53    const result = await loadChangesSince(channel, anchors[channel])
54 
55    if (result.tooOld) {
56      send('flux:backfill', { channel, tooOld: true })
57      continue
58    }
59 
60    for (const row of result.rows) {
61      send(row.event, {
62        channel,
63        id:        row.id,
64        timestamp: row.timestamp,
65        op:        row.op,
66        data:      row.data,
67      })
68    }
69 
70    send('flux:backfill', { channel, asOf: result.asOf })
71  }
72 
73  backfilling = false
74  held.forEach((emit) => emit())
75})

1// Flux needs to know whether a catch-up PROVED the data is current.
2// The adapter reads your backfill marker and reports one of:
3//
4//   marker received, rows came before it    -> applied
5//   marker received, no rows                -> confirmed-current
6//   marker with tooOld: true                -> too-old   (Flux refetches)
7//   a row before the marker was unusable    -> failed    (Flux refetches)
8//     (no id, no finite timestamp, no row, or a handler that threw)
9//   backfill began later than Flux needs    -> failed    (Flux refetches)
10//   no marker within 30 seconds             -> failed    (Flux refetches)
11//   stream 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 on the marker is optional but recommended:
17//   with asOf     Flux records the channel as current through that
18//                 server-clock time
19//   without asOf  Flux uses the moment the catch-up started (server clock)
20//                 minus a 5-second safety margin, so it never trusts the
21//                 device clock
22//
23// "failed" is never a data loss. It means "I can't prove this channel is
24// current", so Flux keeps the gap open and refetches rather than claim
25// freshness it can't back up. The marker is what lets a quiet channel be
26// confirmed current without a refetch.

1// An EventSource URL can't change after it opens, so a channel registered
2// after the stream is running isn't on it yet. The adapter handles this:
3//
4//   flux.register({ channel: 'notes', ... })   // stream already open
5//
6//   -> the adapter reopens the stream once, with every channel on it
7//   -> registrations that arrive together (app start-up, a route change
8//      mounting several stores) are batched into ONE reopen, not one each
9//
10// Channels that were already live reopen from the newest position the
11// adapter has seen for them, not from the old anchor, so a reopen does not
12// replay history you already have. Your server just sees a fresh request
13// with new anchors.

1// Because browser EventSource cannot send messages client -> server,
2// deletion reconciliation is handled via standard HTTP POST when
3// reconcileUrl is configured:
4//
5// 1. The adapter posts the cached local ids for ONE channel:
6//      POST /api/reconcile/items
7//      Body: { "channel": "items", "localIds": ["id_1", "id_2", "id_3"] }
8//
9//    Lists longer than 5,000 ids are sent as several sequential requests
10//    of up to 5,000 ids each.
11//
12// 2. The server checks its database and returns the ids it no longer has:
13//      Response: { "deletedIds": ["id_2"] }
14//
15// 3. Flux removes each of those ids from the store and from disk.
16//
17// A request that times out (15 seconds), fails, or returns a non-2xx
18// status just stops there: whatever earlier chunks already confirmed is
19// kept, and nothing is deleted on a guess.

1// One stream means one failure domain.
2//
3// If the stream closes and the browser gives up, every channel on it
4// stops together, and Flux treats every channel as having a gap until a
5// reconnect and catch-up (or a refetch) closes it. That is the trade for
6// holding one connection instead of one per channel.
7//
8// The browser's own retry (while the stream is "connecting") reuses the
9// ORIGINAL url, so it asks for the original anchors again and the server
10// may resend rows you already have. Applying the same row twice is
11// harmless, so Flux doesn't try to filter them.
12//
13// A very long channel list makes a long URL. The adapter logs a warning
14// past about 7,000 characters, because browsers and proxies may reject
15// URLs that long. Keep channel names short, or register fewer channels
16// on the SSE adapter.

1// Rebuild the adapter (or the url) when the token changes, since the
2// url string is fixed for the stream's lifetime.
3createSseAdapter({
4  url: `https://api.myapp.com/api/stream?token=${shortLivedToken}`,
5})
6 
7// Same-origin cookies need no change at all. For cross-origin cookies:
8createSseAdapter({
9  url: 'https://api.myapp.com/api/stream',
10  options: { withCredentials: true },
11})