outbox.ts · Source

outbox.tspackages/collaboration/src/outbox.tsraw ← src15,353 B← Back to the app
1/**
2 * @factory/collaboration — the offline outbox (TASK-0095 R1).
3 *
4 * Every REST mutation the app makes while the realtime path cannot deliver
5 * it (offline / reconnecting) is enqueued here instead of sent:
6 * `{key (idempotency key), method, path, body, queuedAt, resource}`. On
7 * reconnect the {@link OutboxQueue.drain} worker replays them:
8 *
9 *   - **per-resource ordering** — a later mutation to the same resource
10 *     never replays before an earlier one (per-resource FIFO chain;
11 *     different resources interleave, `maxConcurrentResources` chains in
12 *     parallel — the recorded concurrency, default 3).
13 *   - **idempotency** — every replay sends `Idempotency-Key: <key>`; the
14 *     API middleware (apps/api, same task) checks a 24 h Redis keyspace
15 *     `idem:{userId}:{key}` BEFORE execution and replays the stored
16 *     response on a hit — a double-replay (crash mid-drain) creates no
17 *     duplicate. The key is `key_` + 32 hex (unpredictable — a collision
18 *     is a non-event, recorded in the task Security section).
19 *   - **failure classification** — 4xx → the item moves to the **failed**
20 *     list (the "Review failed changes" dialog: per-item Retry/Discard);
21 *     5xx / network → retry with backoff `min(1 s · 2^attempts, 30 s)`;
22 *     `200/201/…` (incl. an idempotent hit) → done, removed.
23 *   - **retry = a fresh key** — "a retry is a new attempt, the old key is
24 *     dead" (the server's 24 h stored response for the old key must not be
25 *     replayed for a deliberately re-sent mutation).
26 *   - **GC** — pending items older than 24 h are garbage-collected
27 *     (the task's offline-file-upload rule generalizes; recorded: FAILED
28 *     items are NOT auto-GC'd — they need an explicit user Discard).
29 *   - **cap** — 500 pending items max; enqueuing beyond the cap evicts
30 *     the OLDEST pending item (oldest is the least likely to replay
31 *     cleanly after a day offline — recorded; the eviction is surfaced
32 *     via `onEvicted`).
33 *
34 * The queue is deterministic (injected clock/timers) and storage-agnostic
35 * (`OutboxStore` — apps/web implements it with `idb`, the y-indexeddb
36 * sibling; tests use a memory store). LWW-with-consent (title 409) and
37 * property LWW are app-level decisions ON TOP of the 4xx→failed
38 * classification — the queue never auto-resolves conflicts (recorded).
39 */
40 
41import type { OutboxItem, OutboxResult, OutboxStore } from "./types.js";
42 
43/** The outbox's public types (the app imports them via `./outbox`). */
44export type { OutboxItem, OutboxResult, OutboxStore } from "./types.js";
45 
46/** The outbox cap (the task's 500-item limit). */
47export const OUTBOX_MAX_ITEMS = 500;
48 
49/** Items pending longer than this are GC'd (the task's 24 h rule). */
50export const OUTBOX_GC_MS = 24 * 60 * 60 * 1000;
51 
52/** Backoff bounds: 1 s → 30 s (the task's 1 s → 30 s max). */
53export const OUTBOX_BACKOFF_MIN_MS = 1000;
54export const OUTBOX_BACKOFF_MAX_MS = 30_000;
55 
56/** Default parallel resource chains (the recorded concurrency). */
57export const OUTBOX_DEFAULT_CONCURRENCY = 3;
58 
59/** `key_` + 32 hex — unpredictable (crypto when present, fallback random). */
60export function generateIdempotencyKey(): string {
61  const c = (globalThis as Record<string, unknown>).crypto as
62    | { getRandomValues?: (arr: Uint8Array) => Uint8Array }
63    | undefined;
64  if (c !== undefined && typeof c.getRandomValues === "function") {
65    const buf = new Uint8Array(16);
66    c.getRandomValues(buf);
67    return (
68      "key_" +
69      Array.from(buf, (b) => b.toString(16).padStart(2, "0")).join("")
70    );
71  }
72  let out = "";
73  for (let i = 0; i < 32; i += 1) {
74    out += Math.floor(Math.random() * 16).toString(16);
75  }
76  return "key_" + out;
77}
78 
79/** The per-resource chain key (ordering scope). */
80export function resourceKey(resource: { type: string; id: string }): string {
81  return `${resource.type}:${resource.id}`;
82}
83 
84/** Backoff delay for an item at `attempts` (1 s · 2^n, capped at 30 s). */
85export function backoffForAttempt(attempts: number): number {
86  const n = Math.max(0, attempts);
87  return Math.min(OUTBOX_BACKOFF_MIN_MS * 2 ** n, OUTBOX_BACKOFF_MAX_MS);
88}
89 
90export interface OutboxQueueOptions {
91  /** The persistence backend (idb in the app, a memory store in tests). */
92  readonly store: OutboxStore;
93  /** Sends one item with the Idempotency-Key header (the api client). */
94  readonly send: (item: OutboxItem) => Promise<OutboxResult>;
95  /** Injectable clock (ms) — determinism (AGENTS.md rule 25). */
96  readonly now?: () => number;
97  /** Parallel resource chains during a drain (default 3). */
98  readonly maxConcurrentResources?: number;
99  /** Pending/failed counts changed (the banner's count input). */
100  readonly onCountChange?: (info: { pending: number; failed: number }) => void;
101  /** An item was evicted by the 500-cap (the toast material). */
102  readonly onEvicted?: (item: OutboxItem) => void;
103  /** A drain finished (observability — the task's "drains with counts"). */
104  readonly onDrain?: (result: DrainResult) => void;
105  /** Initial items (skips the store load; tests). */
106  readonly initial?: readonly OutboxItem[];
107}
108 
109export interface DrainResult {
110  readonly sent: number;
111  readonly failed: number;
112  readonly retrying: number;
113  readonly pending: number;
114  readonly failedCount: number;
115}
116 
117/**
118 * The outbox queue (the app's `offlineQueue` Zustand store wraps this —
119 * the store owns the UI, the queue owns the state machine).
120 */
121export class OutboxQueue {
122  private readonly store: OutboxStore;
123  private readonly send: (item: OutboxItem) => Promise<OutboxResult>;
124  private readonly now: () => number;
125  private readonly maxConcurrentResources: number;
126  // `| undefined` (not `?`): exactOptionalPropertyTypes — the constructor
127  // may legitimately assign `undefined` (option absent = callback off).
128  private readonly onCountChange: ((info: { pending: number; failed: number }) => void) | undefined;
129  private readonly onEvicted: ((item: OutboxItem) => void) | undefined;
130  private readonly onDrain: ((result: DrainResult) => void) | undefined;
131 
132  private pending: OutboxItem[] = [];
133  private failed: OutboxItem[] = [];
134  private draining = false;
135  private loaded = false;
136 
137  constructor(opts: OutboxQueueOptions) {
138    this.store = opts.store;
139    this.send = opts.send;
140    this.now = opts.now ?? (() => Date.now());
141    this.maxConcurrentResources = opts.maxConcurrentResources ?? OUTBOX_DEFAULT_CONCURRENCY;
142    this.onCountChange = opts.onCountChange;
143    this.onEvicted = opts.onEvicted;
144    this.onDrain = opts.onDrain;
145    if (opts.initial !== undefined) {
146      for (const item of opts.initial) {
147        (item.lastError !== undefined ? this.failed : this.pending).push(item);
148      }
149      this.loaded = true;
150    }
151  }
152 
153  /** Load persisted items (call once before use; a no-op when `initial` was set). */
154  async init(): Promise<void> {
155    if (this.loaded) return;
156    const items = await this.store.load();
157    for (const item of items) {
158      (item.lastError !== undefined ? this.failed : this.pending).push(item);
159    }
160    this.loaded = true;
161    this.persist();
162  }
163 
164  get pendingItems(): readonly OutboxItem[] {
165    return this.pending;
166  }
167 
168  get failedItems(): readonly OutboxItem[] {
169    return this.failed;
170  }
171 
172  /** The banner's pending count (CRDT-pending is summed by the app). */
173  count(): number {
174    return this.pending.length;
175  }
176 
177  failedCount(): number {
178    return this.failed.length;
179  }
180 
181  /**
182   * Enqueue a mutation for replay. Evicts the oldest pending item when
183   * the 500 cap is reached (recorded). Returns the item (the caller
184   * shows the optimistic UI keyed on it).
185   */
186  enqueue(
187    method: OutboxItem["method"],
188    path: string,
189    body: Record<string, unknown> | undefined,
190    resource: { type: string; id: string },
191  ): OutboxItem {
192    // GC first — a day-old queue makes room for the new mutation.
193    this.gc();
194    const item: OutboxItem = {
195      key: generateIdempotencyKey(),
196      method,
197      path,
198      ...(body !== undefined ? { body } : {}),
199      queuedAt: this.now(),
200      resource,
201      attempts: 0,
202    };
203    this.pending.push(item);
204    while (this.pending.length > OUTBOX_MAX_ITEMS) {
205      const evicted = this.pending.shift();
206      if (evicted !== undefined) this.onEvicted?.(evicted);
207    }
208    this.persist();
209    this.emitCount();
210    return item;
211  }
212 
213  /**
214   * Drop peers silent forever: remove pending items queued > 24 h ago
215   * (the task's GC rule). FAILED items are kept (explicit Discard only).
216   * Returns the removed items.
217   */
218  gc(): readonly OutboxItem[] {
219    const cutoff = this.now() - OUTBOX_GC_MS;
220    const kept: OutboxItem[] = [];
221    const removed: OutboxItem[] = [];
222    for (const item of this.pending) {
223      (item.queuedAt < cutoff ? removed : kept).push(item);
224    }
225    if (removed.length > 0) {
226      this.pending = kept;
227      this.persist();
228      this.emitCount();
229    }
230    return removed;
231  }
232 
233  /**
234   * A 4xx failed mutation: retry with a FRESH key (the old key is dead —
235   * the server's stored response for it must not replay, recorded in the
236   * task). The item re-enters the pending chain head of its resource.
237   */
238  retry(item: OutboxItem): OutboxItem {
239    const idx = this.failed.findIndex((i) => i.key === item.key);
240    if (idx < 0) return item;
241    this.failed.splice(idx, 1);
242    const fresh: OutboxItem = {
243      ...item,
244      key: generateIdempotencyKey(),
245      attempts: 0,
246      queuedAt: this.now(),
247    };
248    delete (fresh as { nextAttemptAt?: number }).nextAttemptAt;
249    delete (fresh as { lastError?: string }).lastError;
250    this.pending.push(fresh);
251    this.persist();
252    this.emitCount();
253    return fresh;
254  }
255 
256  /** The user's explicit Discard (the confirm dialog precedes the call). */
257  discard(item: OutboxItem): void {
258    const fIdx = this.failed.findIndex((i) => i.key === item.key);
259    if (fIdx >= 0) {
260      this.failed.splice(fIdx, 1);
261    } else {
262      const pIdx = this.pending.findIndex((i) => i.key === item.key);
263      if (pIdx >= 0) this.pending.splice(pIdx, 1);
264    }
265    this.persist();
266    this.emitCount();
267  }
268 
269  /**
270   * Replay pending mutations: per-resource FIFO chains, up to
271   * `maxConcurrentResources` chains in parallel (the recorded
272   * concurrency). Items whose backoff window is still open are left for
273   * the next drain (they are NOT re-sent early). Idempotent hits count
274   * as sent (the server replayed the stored response — no re-run).
275   */
276  async drain(): Promise<DrainResult> {
277    if (this.draining) {
278      return {
279        sent: 0,
280        failed: 0,
281        retrying: 0,
282        pending: this.pending.length,
283        failedCount: this.failed.length,
284      };
285    }
286    this.draining = true;
287    try {
288      const sent = { count: 0 };
289      const failedMoves = { count: 0 };
290      const retrying = { count: 0 };
291 
292      // Group by resource (queuedAt order preserved within each chain).
293      const chains = new Map<string, OutboxItem[]>();
294      const sorted = [...this.pending].sort((a, b) => a.queuedAt - b.queuedAt);
295      for (const item of sorted) {
296        const key = resourceKey(item.resource);
297        const list = chains.get(key) ?? [];
298        list.push(item);
299        chains.set(key, list);
300      }
301      const keys = [...chains.keys()];
302 
303      // Run up to N chains in parallel; each chain is strictly FIFO.
304      const runChain = async (key: string): Promise<void> => {
305        const items = chains.get(key);
306        if (items === undefined) return;
307        for (const item of items) {
308          // Backoff window still open → wait for the next drain.
309          if (item.nextAttemptAt !== undefined && this.now() < item.nextAttemptAt) {
310            retrying.count += 1;
311            continue;
312          }
313          let result: OutboxResult;
314          try {
315            result = await this.send(item);
316          } catch (err) {
317            // Network failure (the send threw, not resolved) → retryable.
318            result = { ok: false, error: err instanceof Error ? err.message : String(err) };
319          }
320          if (result.ok) {
321            this.removePending(item);
322            sent.count += 1;
323          } else if (result.status !== undefined && result.status >= 400 && result.status < 500) {
324            // 4xx → the failed list (a user decision: Retry/Discard).
325            this.moveToFailed(item, result.error ?? `HTTP ${result.status}`);
326            failedMoves.count += 1;
327          } else {
328            // 5xx / network → retry with backoff (key stays — the server
329            // never executed, or the response was not stored: a retry of
330            // the SAME key is safe, the middleware checks before run).
331            this.pending = this.pending.map((p) =>
332              p.key === item.key
333                ? {
334                    ...p,
335                    attempts: p.attempts + 1,
336                    nextAttemptAt: this.now() + backoffForAttempt(p.attempts + 1),
337                    ...(result.error !== undefined ? { lastError: result.error } : {}),
338                  }
339                : p,
340            );
341            retrying.count += 1;
342          }
343        }
344      };
345 
346      // A simple bounded pool over the resource chains.
347      const queue = [...keys];
348      const workers = Array.from(
349        { length: Math.min(this.maxConcurrentResources, queue.length) },
350        async () => {
351          for (;;) {
352            const key = queue.shift();
353            if (key === undefined) return;
354            await runChain(key);
355          }
356        },
357      );
358      await Promise.all(workers);
359 
360      this.persist();
361      this.emitCount();
362      const result: DrainResult = {
363        sent: sent.count,
364        failed: failedMoves.count,
365        retrying: retrying.count,
366        pending: this.pending.length,
367        failedCount: this.failed.length,
368      };
369      this.onDrain?.(result);
370      return result;
371    } finally {
372      this.draining = false;
373    }
374  }
375 
376  /** Logout = explicit abandon (the recorded rule): clear the outbox. */
377  async clear(): Promise<void> {
378    this.pending = [];
379    this.failed = [];
380    await this.persist();
381    this.emitCount();
382  }
383 
384  private removePending(item: OutboxItem): void {
385    const idx = this.pending.findIndex((i) => i.key === item.key);
386    if (idx >= 0) this.pending.splice(idx, 1);
387  }
388 
389  private moveToFailed(item: OutboxItem, error: string): void {
390    this.removePending(item);
391    this.failed = this.failed.filter((i) => i.key !== item.key);
392    this.failed.push({ ...item, lastError: error });
393  }
394 
395  private async persist(): Promise<void> {
396    try {
397      await this.store.save([...this.pending, ...this.failed]);
398    } catch {
399      // Persistence failure is non-fatal (the in-memory queue is the
400      // source of truth for the session; a reload loses unsaved
401      // enqueues — recorded: the app surfaces a reload warning).
402    }
403  }
404 
405  private emitCount(): void {
406    this.onCountChange?.({ pending: this.pending.length, failed: this.failed.length });
407  }
408}
409 
410/** A memory `OutboxStore` for tests (the app uses `idb`). */
411export class MemoryOutboxStore implements OutboxStore {
412  items: OutboxItem[] = [];
413  async load(): Promise<readonly OutboxItem[]> {
414    return this.items;
415  }
416  async save(items: readonly OutboxItem[]): Promise<void> {
417    this.items = [...items];
418  }
419}
420