offlineQueue.ts

offlineQueue.tsapps/web/src/stores/offlineQueue.tsraw ← stores11,899 B← Back to the app
1"use client";
2 
3/**
4 * The offline outbox store — TASK-0095 R1 (the app half of `apps/web/stores/
5 * offlineQueue.ts`). Wraps the platform-agnostic `OutboxQueue`
6 * (`@factory/collaboration/outbox`) with:
7 *
8 *   - a persistence backend (IndexedDB — the spec's "a plain `idb` store").
9 *     `idb` is only a TRANSITIVE of y-indexeddb (not resolvable under pnpm's
10 *     strict layout), so we use the raw IndexedDB API directly — ZERO new
11 *     dependencies (recorded against the spec's `idb` assumption);
12 *   - a Zustand mirror of the pending/failed counts (the banner's input —
13 *     re-renders on `onCountChange`);
14 *   - the `enqueueMutation` promise (an offline mutation suspends until its
15 *     replay produces a real 2xx body, then resolves with it; a 4xx rejects);
16 *   - the `online`-event drain (replay on reconnect — the spec's "the
17 *     provider's online event → the outbox worker drains FIFO per resource");
18 *   - `clear()` on logout (the recorded rule: a signed-out user's queued
19 *     mutations are discarded).
20 *
21 * The TRANSPORT (the actual HTTP send) is injected by `lib/api.ts` via
22 * `registerOutboxTransport` — this module does NOT import `api.ts` (no cycle:
23 * api.ts → offlineQueue.ts one direction). The transport sends each item with
24 * its OWN `Idempotency-Key` (the item carries the key; a 5xx retry reuses it,
25 * a user Retry gets a fresh one from `OutboxQueue.retry`).
26 */
27import { create } from "zustand";
28import {
29  MemoryOutboxStore,
30  OutboxQueue,
31  type OutboxItem,
32  type OutboxResult,
33  type OutboxStore,
34} from "@factory/collaboration/outbox";
35 
36/* ---------------------------------------------------------------- store */
37 
38/** The raw IndexedDB outbox backend (DB "factory-outbox", store "items"). */
39const OUTBOX_DB = "factory-outbox";
40const OUTBOX_STORE = "items";
41 
42function createOutboxStore(): OutboxStore {
43  // No IndexedDB (SSR / private mode / jsdom) → the in-memory store. The
44  // state machine is identical; persistence is a browser concern.
45  if (typeof indexedDB === "undefined") return new MemoryOutboxStore();
46 
47  const open = (): Promise<IDBDatabase> =>
48    new Promise((resolve, reject) => {
49      const req = indexedDB.open(OUTBOX_DB, 1);
50      req.onupgradeneeded = () => {
51        if (!req.result.objectStoreNames.contains(OUTBOX_STORE)) {
52          req.result.createObjectStore(OUTBOX_STORE, { keyPath: "key" });
53        }
54      };
55      req.onsuccess = () => resolve(req.result);
56      req.onerror = () => reject(req.error);
57    });
58 
59  const run = async <T>(mode: IDBTransactionMode, fn: (store: IDBObjectStore) => IDBRequest): Promise<T> => {
60    const db = await open();
61    return new Promise<T>((resolve, reject) => {
62      const tx = db.transaction(OUTBOX_STORE, mode);
63      const req = fn(tx.objectStore(OUTBOX_STORE));
64      req.onsuccess = () => resolve(req.result as T);
65      req.onerror = () => reject(req.error);
66      tx.oncomplete = () => db.close();
67    });
68  };
69 
70  return {
71    load: async () => {
72      try {
73        const all = await run<OutboxItem[]>("readonly", (s) => s.getAll() as IDBRequest<OutboxItem[]>);
74        return all;
75      } catch {
76        return []; // storage unavailable → start empty (fail-open, no crash)
77      }
78    },
79    save: async (items: readonly OutboxItem[]) => {
80      try {
81        await run("readwrite", (s) => {
82          const clearReq = s.clear();
83          for (const item of items) s.put(item);
84          return clearReq;
85        });
86      } catch {
87        /* storage unavailable → in-memory state still holds (non-fatal) */
88      }
89    },
90  };
91}
92 
93/* --------------------------------------------------------------- queue */
94 
95/** The registered transport (lib/api.ts's send — sends an item with its key). */
96type OutboxTransport = (item: OutboxItem) => Promise<OutboxResult>;
97const transportRef: { current: OutboxTransport | null } = { current: null };
98 
99/** The app-wide outbox (one queue per tab; the store mirrors its counts). */
100const queue = new OutboxQueue({
101  store: createOutboxStore(),
102  now: () => Date.now(),
103  send: (item) => {
104    const t = transportRef.current;
105    if (!t) return Promise.resolve({ ok: false, error: "outbox transport not registered" });
106    return t(item);
107  },
108  onCountChange: (c) => {
109    useOutboxStore.setState({ pending: c.pending, failed: c.failed, ready: true });
110  },
111  onEvicted: () => {
112    // A day's worth of offline mutations overflowed the 500-cap — the oldest
113    // was dropped. Surface it (the toast material the spec calls for).
114    try {
115      // Lazy (avoids a cycle at module load; ui-store has no outbox import).
116      void import("@/stores/ui-store").then(({ toast }) =>
117        toast("error", "Offline storage is full", "The oldest queued change was discarded."),
118      );
119    } catch {
120      /* toast is best-effort */
121    }
122  },
123});
124 
125/** Register the transport (lib/api.ts calls this once at module load). */
126export function registerOutboxTransport(transport: OutboxTransport): void {
127  transportRef.current = transport;
128}
129 
130/* ------------------------------------------- in-flight mutation promises */
131 
132/**
133 * The pending mutation callbacks (an offline mutation's caller promise).
134 * Resolved when the item's replay produces a real 2xx body; rejected on a 4xx
135 * (the item moves to the failed list). A 5xx/network keeps it PENDING (the
136 * item retries; it resolves on a later success).
137 */
138interface PendingMutation {
139  resolve: (body: unknown) => void;
140  reject: (err: unknown) => void;
141}
142const pendingMutations = new Map<string, PendingMutation>();
143 
144/* ------------------------------------------------------------ resource */
145 
146const RESOURCE_ROOTS = new Set([
147  "pages",
148  "blocks",
149  "workspaces",
150  "comments",
151  "databases",
152  "records",
153  "templates",
154  "notifications",
155  "invites",
156  "files",
157  "teams",
158  "members",
159]);
160 
161/**
162 * Derive the per-resource ordering key for a mutation (the spec's `resource:
163 * {type, id}` — a later mutation to the SAME resource never replays before an
164 * earlier one; a different resource can interleave). Heuristic over the path
165 * (the api client is generic; the caller does not pass a resource):
166 *   - `/pages/{id}`            → {type:"page",     id}          (title/icon)
167 *   - `/blocks/{id}`           → {type:"block",    id}          (block edit)
168 *   - `/databases/{id}/...`    → {type:"database", id}          (record/props)
169 *   - `POST /comments`         → {type:"comment",  body.pageId} (per-page order)
170 *   - a create to a collection → {type: <root>,    __new__|scope}
171 */
172export function deriveOutboxResource(
173  path: string,
174  body?: Record<string, unknown> | null,
175): { type: string; id: string } {
176  const segs = (path.split("?")[0] ?? "").split("/").filter(Boolean);
177  const rootIdx = segs.findIndex((s) => RESOURCE_ROOTS.has(s));
178  if (rootIdx === -1) {
179    // Not a known root (an auth/search/etc. mutation) → order by the whole path.
180    return { type: "unknown", id: segs.join("/") || "root" };
181  }
182  const root = segs[rootIdx] ?? "unknown";
183  const id = segs[rootIdx + 1];
184  if (id !== undefined) {
185    // `/root/{id}` or `/root/{id}/sub` → the resource is the id (mutations to
186    // the same entity are ordered; sub-collections order under their parent).
187    return { type: root, id };
188  }
189  // A create to the collection `/root` (no id yet) → scope by the body's
190  // parent (e.g. a comment's pageId) so same-parent creates stay in order.
191  const scope =
192    (typeof body?.pageId === "string" && body.pageId) ||
193    (typeof body?.workspaceId === "string" && body.workspaceId) ||
194    (typeof body?.databaseId === "string" && body.databaseId) ||
195    "__new__";
196  return { type: root, id: scope };
197}
198 
199/* ------------------------------------------------------------- enqueue */
200 
201export interface EnqueueMutationInput {
202  method: OutboxItem["method"];
203  path: string;
204  body?: Record<string, unknown>;
205  /** The per-resource ordering key (overrides the path-derived one). */
206  resource?: { type: string; id: string };
207}
208 
209/**
210 * Enqueue an OFFLINE mutation. Returns a promise that:
211 *   - RESOLVES with the real 2xx body when the replay succeeds (the caller's
212 *     contract is preserved — it gets the server's response, just later);
213 *   - REJECTS with the ApiError on a 4xx (the item → the failed list).
214 * The item's `key` is its idempotency token (the transport sends it as the
215 * `Idempotency-Key` header; the API replays a seen key without re-running).
216 * A Retry re-queues via `OutboxQueue.retry` (a FRESH key — the old key is
217 * dead), so no key pinning is needed here.
218 */
219export function enqueueMutation<T = unknown>(input: EnqueueMutationInput): Promise<T> {
220  const resource = input.resource ?? deriveOutboxResource(input.path, input.body);
221  const item = queue.enqueue(input.method, input.path, input.body, resource);
222  return new Promise<T>((resolve, reject) => {
223    pendingMutations.set(item.key, {
224      resolve: (body) => resolve(body as T),
225      reject,
226    });
227  });
228}
229 
230/** Resolve the caller promise for a successfully-replayed item (2xx). */
231export function settleMutationSuccess(item: OutboxItem, body: unknown): void {
232  const p = pendingMutations.get(item.key);
233  if (p) {
234    pendingMutations.delete(item.key);
235    p.resolve(body);
236  }
237}
238 
239/** Reject the caller promise for a 4xx-replayed item (→ the failed list). */
240export function settleMutationFailure(item: OutboxItem, err: unknown): void {
241  const p = pendingMutations.get(item.key);
242  if (p) {
243    pendingMutations.delete(item.key);
244    p.reject(err);
245  }
246}
247 
248/** Drop a caller promise (a Discard / a clear) without resolving it. */
249export function dropMutation(item: OutboxItem): void {
250  pendingMutations.delete(item.key);
251}
252 
253/* ---------------------------------------------------------------- store */
254 
255interface OutboxState {
256  ready: boolean;
257  pending: number;
258  failed: number;
259  load: () => Promise<void>;
260  drain: () => Promise<void>;
261  retry: (key: string) => void;
262  discard: (key: string) => void;
263  clear: () => Promise<void>;
264  pendingItems: () => readonly OutboxItem[];
265  failedItems: () => readonly OutboxItem[];
266}
267 
268export const useOutboxStore = create<OutboxState>()((set) => ({
269  ready: false,
270  pending: 0,
271  failed: 0,
272  load: async () => {
273    await queue.init(); // load persisted items (splits pending vs failed)
274    set({ ready: true });
275  },
276  drain: async () => {
277    await queue.drain();
278  },
279  retry: (key) => {
280    const item = [...queue.failedItems, ...queue.pendingItems].find((i) => i.key === key);
281    if (!item) return;
282    dropMutation(item); // the original caller already settled (a 4xx) — a fresh attempt
283    queue.retry(item);
284    void queue.drain();
285  },
286  discard: (key) => {
287    const item = [...queue.failedItems, ...queue.pendingItems].find((i) => i.key === key);
288    if (!item) return;
289    dropMutation(item);
290    queue.discard(item);
291  },
292  clear: async () => {
293    // A signed-out user's queued mutations are discarded (the recorded rule:
294    // logout = an explicit abandon). Drop the caller promises (they are orphaned).
295    for (const item of [...queue.pendingItems, ...queue.failedItems]) dropMutation(item);
296    await queue.clear();
297    set({ pending: 0, failed: 0 });
298  },
299  pendingItems: () => queue.pendingItems,
300  failedItems: () => queue.failedItems,
301}));
302 
303/* --------------------------------------------------- drain on reconnect */
304 
305/**
306 * Wire the `online` event → drain (the spec's replay-on-reconnect). Also
307 * drains once on `load` (a returning tab with a persisted outbox). Idempotent
308 * (a module-level guard); called from the app provider / a hook.
309 */
310let listenersWired = false;
311export function initOutbox(): void {
312  if (listenersWired) return;
313  listenersWired = true;
314  if (typeof window === "undefined") return;
315  void useOutboxStore.getState().load();
316  window.addEventListener("online", () => {
317    void useOutboxStore.getState().drain();
318  });
319}
320