replay()

1// Replay one store's queued mutations
2await flux.replay('profile')
3 
4// Replay every store's queue — this is also what bootstrap() calls
5// on the leader tab immediately after a successful fetch
6await flux.replayAll()
7 
8// Both return nothing, but the lower-level call reports what happened.
9// flux.retryFailed() (see "Retrying and discarding" below) returns:
10//   { status: 'completed' | 'skipped' | 'offline' | 'aborted',
11//     replayed: number,    // entries accepted or confirmed as duplicates
12//     failed:   number }   // entries newly marked failed in this run

1// You rarely need to call replay() yourself. Automatic replay happens
2// only on a reconnect, plus once at session start:
3//
4// Session start     — 1200ms after mount. Pending entries only.
5// Back online       — 'online' event, 2000ms delay. Reconnect run.
6// Resume            — pageshow from the iOS back/forward cache, or the tab
7//                     becoming visible, 3000ms delay. Counts as a reconnect
8//                     ONLY if the app saw itself offline since its last run,
9//                     because frozen pages often miss the 'online' event.
10// Service worker    — FLUX_QUEUE_SYNC_SUCCESS message, 500ms delay.
11//                     Pending entries only.
12// Tab hidden        — registers a background sync tag, but only when
13//                     entries are pending. Failed entries never register one.
14//
15// Plain window focus, plain tab visibility changes, and server errors
16// do not start a replay.
17//
18// Manual calls are for cases outside these triggers — a "Retry" button
19// in a failed changes panel, for example.

1// The reason decides which entries a run sends. Pass it explicitly
2// when you call replay() yourself:
3 
4await flux.replay('notes')                          // default: 'session-start'
5await flux.replay('notes', { reason: 'reconnect' }) // pending + retryable failed
6await flux.replay('notes', { reason: 'manual', entryIds: [entry.id] })
7 
8// For failed entries you will almost always want these instead:
9await flux.retryFailed('notes', [entry.id])
10await flux.retryFailed('notes')

1// Before a replay run does anything else, replayQueue() calls
2// isNetworkReachable(probeUrl) and bails out immediately if it
3// returns false — this happens BEFORE the Web Lock is even
4// requested, so an offline device never contends for the lock or
5// burns the cooldown window on a run that can't succeed anyway.
6//
7// Why not just trust the browser's 'online' event? navigator.onLine
8// and the online/offline events only reflect whether the device has
9// *a* network interface up — a phone can report "online" while sitting
10// on a captive Wi-Fi portal, a VPN that's mid-reconnect, or a LAN with
11// no route to your actual server. isNetworkReachable() sends a real
12// HEAD request (with retry) to confirm there's a live path to YOUR
13// backend specifically, not just to some network somewhere.
14//
15// A failed probe is not a failed write. Nothing is marked failed and
16// no failed panel appears; the run reports status 'offline' and the
17// next reconnect tries again. A server that answers the probe with
18// anything below 500 counts as reachable.
19//
20// The probe runs for manual retries too. If the user presses Retry
21// while the device is unreachable, retryFailed() resolves with
22// { status: 'offline' } and changes nothing.
23//
24// probeUrl is optional on ReplayConfig. If you don't set one, point
25// your queueConfig at a small dedicated endpoint rather than reusing
26// a heavier API route — the probe should be as cheap as possible
27// since it runs at the start of every replay run, well before the
28// actual replay payload is built.

1// app/api/flux_probe/route.ts — a minimal reachability endpoint,
2// intended purely as a target for isNetworkReachable(). It does no
3// work beyond confirming the request reached your server: no auth,
4// no database, nothing that could itself become the reason a replay
5// gets skipped.
6 
7export const dynamic = 'force-dynamic'
8 
9export async function HEAD() {
10  return new Response(null, { status: 204 })
11}
12 
13export async function GET() {
14  return new Response(null, { status: 204 })
15}

1// Every replay run is Web Locks protected under flux:queue:{storeName} —
2// Tab A holds the lock for the duration of its replay; Tab B waits,
3// then acquires and re-reads the queue, finding it already drained.
4 
5// Two guards prevent redundant AUTOMATIC runs on top of the lock:
6//   - in-flight guard   — blocks a second concurrent replay on the same store
7//   - 4s cooldown guard — after a run starts, blocks rapid re-triggering
8//                          from stacked trigger events (e.g. online + resume firing together)
9 
10// Manual runs — retryFailed(), resolveUnifiedConflict() — bypass both
11// guards. The user explicitly triggered them. The lock still queues a
12// manual run behind any run already in progress.
13 
14// Every request has a timeout (requestTimeoutMs, default 30000). The
15// lock is held while a request is in flight, so a hung connection can
16// no longer block the queue in every tab. A timeout is treated as "no
17// response": the entries stay pending and are sent again on the next
18// reconnect. It is not a failure.

1// Every send ends in exactly one of these outcomes, per entry:
2//
3//   2xx                        removed. onSuccess / onReplaySuccess fire.
4//   duplicate answer           removed. onDuplicate fires. The optimistic
5//                              UI change stays (the server has the write).
6//   error answer               KEPT, marked failed. onFailed + onFailure fire.
7//     (any 4xx, any 5xx,       The user retries or discards.
8//      429, and 409)
9//   no response                KEPT, still pending. Nothing fires.
10//     (offline, timeout,       The run stops and the next reconnect
11//      dropped connection,     sends the entry again.
12//      aborted)
13//
14// A 409 is an ordinary failed status. Earlier versions treated it as
15// success and deleted the entries. If your endpoint uses 409 to mean
16// "I already have this", opt in with idempotency.duplicateStatus.
17//
18// Failures are per request. Entries go out in batches, and a failed
19// batch does not stop later batches. The exceptions that stop the run:
20// no response, and 429.

1// A failed entry carries a failure record:
2//
3// entry.status   // 'failed'  (entries that haven't failed are 'pending')
4// entry.failure = {
5//   kind:           'rejected'      // other 4xx: the server said no
6//                 | 'server'        // 5xx
7//                 | 'auth'          // 401 / 403
8//                 | 'rate_limited'  // 429
9//                 | 'revalidation', // handshake pass 1 got a server error
10//   status:         422,            // HTTP status (0 for 'revalidation' with no status)
11//   message:        'Title is required',  // short text from the response body, if any
12//   at:             1767225600000,  // time of the latest failure
13//   attempts:       2,              // how many times this entry has failed
14//   retryAfterSec:  30,             // only on a 429 that sent Retry-After
15// }
16//
17// message comes from the response body's "message" or "error" field,
18// or from short plain text. HTML error pages are ignored and the HTTP
19// status text is used instead.

1// A reconnect run sends:
2//   - every pending entry, and
3//   - failed entries that could plausibly succeed now:
4//       'server' (5xx), 'auth' (401/403), 'rate_limited' (429, only
5//       once its Retry-After window has passed), 'revalidation',
6//       and status 408.
7//
8// It does NOT resend definitive rejections (400, 409, 422, ...). They
9// would fail the same way every time, so they wait for the user.
10//
11// Failed entries are never retried by a session-start run. After a
12// reload they show up in your failed panel and wait.
13//
14// A failed entry that is not being sent also holds back LATER pending
15// entries with the same dedupeKey, since they may depend on it. Entries
16// with a different dedupeKey, or none, are sent normally.

1// Manual retry — sends exactly the entries you name, whatever their
2// status, ignoring the cooldown.
3const outcome = await flux.retryFailed('notes', [entry.id])
4 
5// Or retry every failed entry on a channel:
6await flux.retryFailed('notes')
7 
8// outcome is null if the channel has no queue (or no user is active
9// for a user-scoped queue).
10if (outcome?.status === 'offline') {
11  // device could not reach your server — nothing changed
12}
13// Entries that fail again are marked failed again (attempts + 1) and
14// onFailed fires for each one.
15 
16// Discard — deletes the entry and fires onRollback so you can revert
17// the optimistic change. refetch: true also invalidates the channel's
18// cached copy and reloads it from the server (use while online).
19await flux.discardFailed('notes', entry.id, { refetch: true })

1// replay() never decides the strategy for you — it just executes
2// whichever one you set on queueConfig.replayStrategy at register() time.
3// The two strategies exist because "did this write succeed" and
4// "is this write still safe to apply" are different questions,
5// and only one of them is worth paying an extra request for.
6 
7flux.register({
8  channel: 'waitlist',
9  queueConfig: {
10    storeName: 'waitlist',
11    replayEndpoint: '/api/waitlist',
12    method: 'POST',
13    replayStrategy: 'fire_and_forget', // default — no existing record to clash with
14  },
15})
16 
17flux.register({
18  channel: 'profile',
19  scope: 'user',
20  queueConfig: {
21    storeName: 'profile',
22    replayEndpoint: '/api/profile',
23    method: 'PATCH',
24    replayStrategy: 'handshake',       // there IS an existing record to clash with
25    revalidateFn: myRevalidateFn,
26  },
27})

1// fire_and_forget — one request per batch, no server round trip to
2// check state first.
3//
4// Queued entries for a given method and endpoint are grouped and sent in
5// batches of up to maxBatchSize (default 50). A batch with one entry
6// sends that entry's payload as the body; a batch with several sends an
7// array. Every request carries:
8//
9//   X-Flux-Batch:       true
10//   X-Flux-Count:       number of entries in this request
11//   X-Flux-Entry-Keys:  one key per entry, in body order: "entryId:revision"
12//
13// There is nothing to reconcile because these writes don't overwrite
14// anything — a waitlist signup, a contact form, a log line. Whatever
15// landed last is simply another row, not a collision.
16//
17// The server answers once per request. Without any extra setup, that
18// answer applies to the whole batch: 2xx removes every entry in it, and
19// an error marks every entry in it failed. See "Idempotency" below to
20// let the server answer per entry instead.

1flux.register({
2  channel: 'notes',
3  queueConfig: {
4    storeName:      'notes_mutation_queue',
5    replayEndpoint: '/api/notes',
6    method:         'POST',
7 
8    // Opt in. Without this block, Flux ignores per-item answers and
9    // treats the batch response as one result.
10    idempotency: {
11      duplicateStatus: 409,           // or a list: [409, 208]. Optional.
12      // keysHeader: 'X-Flux-Entry-Keys',  // default
13    },
14 
15    // The server already has this write. The entry is removed and the
16    // optimistic UI change must stay.
17    onDuplicate: (entry) => {
18      markNoteConfirmed(entry.payload.id)
19    },
20  },
21})

1// THE SERVER CONTRACT
2//
3// 1. Read the keys. Flux sends X-Flux-Entry-Keys on every request, in the
4//    same order as the payloads in the body. Each key is
5//    "entryId:revision". Split on the LAST colon. The revision changes
6//    whenever the user edits a still-queued entry, so an edited write
7//    is never mistaken for an older copy of itself.
8//
9// 2. Dedupe on the key. If you have already processed that key, don't
10//    apply the write again.
11//
12// 3. Answer in one of two ways:
13//
14//    Whole request already known
15//      Respond with the status you set as duplicateStatus. Every entry in
16//      the request is removed and onDuplicate fires for each.
17//
18//    Per entry (works on ANY status, including 200)
19//      Respond with a JSON body:
20//
21//      {
22//        "failed":     [{ "id": "entryId", "status": 422, "message": "Title is required" }],
23//        "duplicates": ["entryId", "entryId"]
24//      }
25//
26//      "id" is the entry id WITHOUT the revision.
27//      A bare id string in "failed" is accepted and counts as a failure
28//      with no message.
29//
30// WHAT FLUX DOES WITH THE ANSWER
31//
32//   listed in failed       kept, marked failed with its own status/message
33//   listed in duplicates   removed, onDuplicate fires, optimistic UI stays
34//   in both lists          counts as failed
35//   not listed, 2xx        removed as succeeded
36//   not listed, non-2xx    marked failed with the request's status
37//   id Flux didn't send    ignored
38//
39// Retrying a failed entry sends the same entry id with the same key. If
40// the server had actually applied it before the error, it answers
41// "duplicate" and Flux cleans the entry up.
42//
43// IMPORTANT: if you return per-item results, every failed item must be
44// listed. An unlisted entry in a 2xx response is treated as succeeded,
45// the same trust Flux places in any 2xx.
46//
47// A 429 is always handled as rate limiting and per-item results in that
48// response are ignored.
49//
50// No response at all (a dropped connection) is the case this exists
51// for: the server may have applied the batch, Flux can't know, so the
52// entries stay pending and are sent again on the next reconnect with
53// the same keys.

1// app/api/notes/route.ts — an endpoint that implements the contract.
2// The db helpers are placeholders: use your own storage.
3 
4import { NextResponse } from 'next/server'
5import { db } from '@/lib/db'
6 
7export async function POST(req: Request) {
8  const keys  = (req.headers.get('X-Flux-Entry-Keys') ?? '').split(',').filter(Boolean)
9  const body  = await req.json()
10  const items = Array.isArray(body) ? body : [body]
11 
12  const failed: Array<{ id: string; status: number; message: string }> = []
13  const duplicates: string[] = []
14 
15  for (let i = 0; i < items.length; i++) {
16    const key     = keys[i]                                // "entryId:revision"
17    const entryId = key.slice(0, key.lastIndexOf(':'))     // split on the LAST colon
18 
19    if (await db.idempotency.exists(key)) {
20      duplicates.push(entryId)                             // already applied
21      continue
22    }
23 
24    try {
25      await db.notes.upsert(items[i])
26      await db.idempotency.record(key)
27    } catch (err: any) {
28      failed.push({ id: entryId, status: 422, message: err?.message ?? 'Could not save' })
29    }
30  }
31 
32  // 200 is fine even when some items failed — Flux reads the lists.
33  return NextResponse.json({ failed, duplicates })
34}

1// handshake — one revalidation request for the whole endpoint group,
2// then the safe entries are sent in batches of up to maxBatchSize. This
3// is the strategy for anything that OVERWRITES an existing server
4// record: dashboard edits, settings, profile fields. Sending those blind
5// risks silently clobbering a change that happened on the server (or
6// from another device) while this client was offline.
7 
8// Pass 1 — bulk probe (read only, the server changes nothing)
9//   flux builds one manifest for the whole endpoint group:
10//     [{ id: dedupeKey ?? entry.id, timestamp: entry.timestamp + clockSkewMs }, ...]
11//   and calls your revalidateFn(entries, manifest) ONCE for the group.
12//   Your function checks each id against current server state and returns:
13//     {
14//       conflictedIds:      string[],
15//       serverStates:       Record<string, any | null>,
16//       alreadyAppliedIds?: string[],
17//     }
18 
19// In-memory split — no extra network cost:
20//   entries in alreadyAppliedIds  -> removed as done, onDuplicate fires
21//   entries NOT in conflictedIds  -> safe, queued for Pass 2 as-is
22//   entries IN conflictedIds      -> routed through the conflict chain below
23 
24// Conflict resolution priority, per conflicted entry:
25//   1. queueConfig.onConflict(entry, serverState) if defined
26//        -> 'proceed'      : send local payload anyway, overwrite server
27//        -> 'apply_merged' : send resolution.mergedPayload instead
28//        -> 'drop'         : falls through to step 2/3
29//   2. engine's onConflictUnified, if configured
30//        -> entry is held, still pending in storage, and surfaced to
31//           your universal conflict UI via getActiveConflicts() /
32//           subscribeToConflicts()
33//        -> nothing is sent for this entry until you call
34//           resolveUnifiedConflict()
35//   3. queueConfig.conflictStrategy as a last-resort default
36//        -> 'reject'  : drop, server wins (safest, and the default)
37//        -> 'replace' : drop, server wins
38//        -> 'merge'   : requires onConflict returning 'apply_merged' —
39//                        without it, flux warns and drops rather than
40//                        guessing at a merge
41//
42// If serverState is null the record was deleted on the server while the
43// entry sat queued. 'proceed' and 'apply_merged' make no sense then, so
44// Flux forces 'drop'.
45//
46// A merged payload is saved on the entry before it is sent, so a retry
47// sends exactly what was approved.
48 
49// Pass 2 — execution
50//   every "safe" and "resolved" payload from the split above is sent
51//   in batches of up to maxBatchSize, using the same answer handling as
52//   fire_and_forget: 2xx removes the batch, an error answer marks the
53//   whole batch failed. Handshake does not read per-item results.

1// Conflicts and failures are different things and are never mixed:
2//
3//                   Conflict                      Failed
4//   Found by        Pass 1, per entry             Pass 2 answer, or a Pass 1 server error
5//   Sent to server  No                            Yes
6//   Unit            one entry                     the whole batch that was sent
7//   Resolved in     your conflict UI              your failed changes panel
8//                   (proceed / drop / merge)      (retry / discard)
9//
10// WHAT A FAILURE DOES AT EACH STAGE
11//
12// Pass 1, no response (device offline)
13//   Nothing changed on the server or in the queue. Entries stay pending,
14//   no panel appears, and the run stops.
15//
16// Pass 1, server answered with an error (revalidateFn threw while the
17// server was reachable)
18//   Conflicts can't be determined, and sending without checking would
19//   defeat the handshake, so Pass 2 does not run. The group is marked
20//   failed with kind 'revalidation'. It is retried on the next reconnect.
21//
22// Pass 2, no response
23//   The entries stay pending and the run stops. Because the server may
24//   have received the batch, return alreadyAppliedIds from revalidateFn
25//   so the next run recognises your own write instead of reporting a
26//   false conflict against it.
27//
28// Pass 2, server answered with an error
29//   The batch is marked failed (a 409 here is just a failed status).
30//   Entries held back as conflicts are not part of it.
31//
32// A manual retry of a handshake store runs Pass 1 again for the
33// entries being retried, so a retried entry can surface a conflict.
34//
35// ONE EXCEPTION TO "NOTHING IS DELETED"
36// Conflict resolution that ends in 'drop' removes the entry: a drop
37// from onConflict, the default 'reject' / 'replace' strategies, a
38// 'merge' with no onConflict, a missing mergedPayload, or a null
39// serverState. Those drops do not mark the entry failed, do not fire
40// onFailed, and do not fire onRollback. If you need the user to see
41// and decide on every conflict, configure onConflictUnified.

1// revalidateFn is required once replayStrategy is 'handshake' — flux
2// warns and falls back to fire_and_forget for that group if it's
3// missing, rather than freezing the queue on a config mistake.
4 
5async function revalidateFn(
6  entries: QueueEntry[],
7  manifest: Array<{ id: string; timestamp: number }>
8) {
9  // Look up current server state for each manifest id, compare against
10  // the (clock-skew-corrected) timestamp, and report back which ids
11  // are stale relative to the server.
12  const res = await fetch('/api/profile/revalidate', {
13    method: 'POST',
14    body: JSON.stringify({ manifest }),
15  })
16 
17  // Throw if the server could not answer. Flux checks whether the device
18  // is offline: if so, entries simply stay pending; if the server is
19  // reachable, the group is marked failed with kind 'revalidation'.
20  if (!res.ok) throw Object.assign(new Error('Revalidation failed'), { status: res.status })
21 
22  return res.json() as Promise<{
23    // ids the server considers conflicting, including records that
24    // no longer exist on the server at all
25    conflictedIds: string[]
26 
27    // id -> current server state. A deleted record maps to null (not
28    // an absent key), so "no conflict" and "conflict, record is gone"
29    // stay distinguishable.
30    serverStates: Record<string, any | null>
31 
32    // Optional. Records the server already has — for example your own
33    // earlier write whose response was lost. Flux removes these as done
34    // instead of resending them or reporting a conflict against yourself.
35    alreadyAppliedIds?: string[]
36  }>
37}

1queueConfig: {
2  // ── Where and how ─────────────────────────────────────────────
3  storeName:      'notes_mutation_queue',
4  replayEndpoint: '/api/notes',
5  method:         'POST',                  // 'POST' | 'PUT' | 'PATCH'
6  probeUrl:       '/api/flux_probe',
7 
8  // ── Strategy ──────────────────────────────────────────────────
9  replayStrategy:   'fire_and_forget',     // or 'handshake'
10  conflictStrategy: 'reject',              // handshake fallback
11  revalidateFn,                            // handshake only
12  onConflict,                              // handshake only
13 
14  // ── Dedupe ────────────────────────────────────────────────────
15  dedupeField:    'id',
16  dedupeStrategy: 'replace',               // or 'reject'
17  mergeHandler,
18 
19  // ── Results ───────────────────────────────────────────────────
20  onSuccess:       (id) => {},             // an entry was accepted
21  onReplaySuccess: (responseBody, entry) => {},
22  onFailed:        (entry, failure) => {}, // an entry became failed (use this)
23  onFailure:       (id, status) => {},     // legacy, fires alongside onFailed
24  onDuplicate:     (entry) => {},          // the server already has it
25  onRollback:      (storeName, error) => {}, // the USER discarded a failed entry
26  onComplete:      (replayed, failed) => {}, // failed = newly failed this run
27 
28  // ── Idempotency (fire_and_forget only, opt-in) ────────────────
29  idempotency: { duplicateStatus: 409 },
30 
31  // ── Limits ────────────────────────────────────────────────────
32  maxBatchSize:     50,                    // entries per request
33  requestTimeoutMs: 30000,                 // a timeout counts as "no response"
34}

1// replayQueue() internally accepts a userId used to filter which
2// entries it's allowed to touch — resolved from the TARGET STORE'S
3// OWN registered scope, never unconditionally from whoever happens
4// to be logged in when replay() is called.
5 
6// A 'global'-scoped queue always passes null, so it keeps replaying
7// correctly regardless of session state:
8await flux.replay('waitlist') // scope: 'global' on this store's registration
9 
10// A 'user'-scoped queue passes the active userId, so it only ever
11// replays that tenant's own queued entries:
12await flux.replay('profile') // scope: 'user' on this store's registration
13 
14// The failed-writes API follows the same rule: a failed panel shows
15// the signed-in user's failed entries plus global-scope ones, never
16// another user's.

1// Each scope has its own AbortController in the engine. If
2// clearUserSession() fires mid-replay — say, while a handshake's
3// execution request is in flight — that scope's fetch rejects
4// immediately rather than landing against a torn-down session.
5 
6// The entry is not marked failed. An aborted request is "no response",
7// so nothing is removed and nothing is counted as a failure.
8await flux.clearUserSession() // aborts any in-flight 'user'-scope replay

1// For handshake replay specifically, entry.timestamp is corrected
2// by flux.clockSkewMs (computed during bootstrap()) before the
3// manifest is built and sent to revalidateFn — this prevents false
4// conflict detection on devices with drifting clocks.
5const correctedTimestamp = entry.timestamp + flux.clockSkewMs