Plucks waiting on a device follow one set of rules

apps/ios/lib/data.ts

@@ -6,7 +6,7 @@ import { Directory, File, Paths } from "expo-file-system";
 import * as Network from "expo-network";
 import { useEffect, useState, useSyncExternalStore } from "react";
 import { AppState } from "react-native";
-import { createQueue, Refused, Unreachable } from "@pluck/client";
+import { createDrain, createQueue, Unreachable } from "@pluck/client";
 import { KEEP, MAX_BYTES, onWifi, overCap, plan, toKeep, type Kept } from "./keep";
 import { kv } from "./kv";
 import { api, authHeader, getSession } from "./session";
@@ -169,32 +169,32 @@ export async function toSync(): Promise<number> {
 }
 
 type SharedRecord = { body?: CreateItemBody; file?: { name: string; contentType: string; path: string } };
+type Share = { f: File; record: SharedRecord | null }; // null: unreadable
+
+// The share extension's leftovers go by the same rules as the phone's queue. The extension writes them and the app
+// only takes them away (with their file, if any), so the store reads and removes rather than rewriting a list.
+const shares = createDrain<Share>({
+  read: async () => {
+    const dir = pendingDir();
+    const files = dir?.exists ? dir.list().filter((f): f is File => f instanceof File && f.name.endsWith(".json")) : [];
+    return Promise.all(files.map(async (f) => ({ f, record: ((await f.text().then(JSON.parse).catch(() => null)) as SharedRecord | null) })));
+  },
+  remove: async ({ f, record }) => {
+    if (f.exists) f.delete();
+    if (record?.file) removeShared(record.file.path);
+  },
+});
 
-// Runs when the app comes to the foreground: the phone's queue, then the share extension's leftovers.
-// One run at a time: opening the app and pulling to refresh must not send the same share twice.
-let syncing: Promise<number> | null = null;
-export const syncPending = (): Promise<number> => (syncing ??= sync().finally(() => (syncing = null)));
-
-async function sync(): Promise<number> {
+// Runs when the app comes to the foreground: the phone's queue, then the share extension's leftovers. Each runs one
+// at a time: opening the app and pulling to refresh must not send the same share twice.
+export async function syncPending(): Promise<number> {
   await queue.drain((body) => api("/api/v1/items", { json: body }));
-  const dir = pendingDir();
-  if (dir?.exists) {
-    for (const f of dir.list()) {
-      if (!(f instanceof File) || !f.name.endsWith(".json")) continue;
-      let record: SharedRecord | null = null;
-      try {
-        record = JSON.parse(await f.text()) as SharedRecord;
-        // The file's name is the key: if the upload below fails, sending the body again makes no second Note.
-        if (record.body) await api("/api/v1/items", { json: { ...record.body, clientKey: f.name } });
-        if (record.file) await uploadShared(record.file);
-        f.delete();
-      } catch (e) {
-        if (record && !(e instanceof Refused)) break; // offline, signed out, busy or down: keep it for next time
-        f.delete(); // refused or unreadable: retrying will not help
-        if (record?.file) removeShared(record.file.path);
-      }
-    }
-  }
+  await shares(async ({ f, record }) => {
+    if (!record) return; // unreadable: nothing to send, so it goes
+    // The file's name is the key: if the upload below fails, sending the body again makes no second Note.
+    if (record.body) await api("/api/v1/items", { json: { ...record.body, clientKey: f.name } });
+    if (record.file) await uploadShared(record.file);
+  });
   void queryClient.invalidateQueries({ queryKey: ["items"] });
   void queryClient.invalidateQueries({ queryKey: keys.pending });
   return toSync();

apps/web/components/ItemList.tsx

@@ -290,11 +290,11 @@ export function ItemList({ view, initial, total: initialTotal, tags: initialTags
     setRows((cur) => [...drafts, ...cur]);
   };
   // A re-pluck returns an existing item: it moves to the top instead of appearing twice.
-  const onPlucked = (d: PluckDraft, { results }: PluckResponse) => {
+  const onPlucked = (key: string, { results }: PluckResponse) => {
     if (view !== "inbox") return;
     const ids = new Set(results.map((r) => r.item.id));
     setTotal((t) => t + results.filter((r) => r.created).length);
-    setRows((cur) => [...results.map((r) => r.item), ...cur.filter((r) => r.draft !== d.key && !ids.has(r.id))]);
+    setRows((cur) => [...results.map((r) => r.item), ...cur.filter((r) => r.draft !== key && !ids.has(r.id))]);
   };
   const onFailed = (d: PluckDraft) => setRows((cur) => cur.filter((r) => r.draft !== d.key));
 

apps/web/components/TopBar.tsx

@@ -1,6 +1,7 @@
 "use client";
 
-import { extractUrls, type PluckResponse, type TextKind } from "@pluck/shared-schemas";
+import { answer, createQueue, type Queued } from "@pluck/client";
+import { extractUrls, type CreateItemBody, type PluckResponse, type TextKind } from "@pluck/shared-schemas";
 import { useTranslations } from "next-intl";
 import Link from "next/link";
 import { useEffect, useRef, useState, type FormEvent } from "react";
@@ -13,38 +14,37 @@ type Props = {
   // The empty inbox's look: 1.5px ink border and a "Pluck it" button, pointing at step 1.
   emphasis?: boolean;
   onPluck?: (draft: PluckDraft) => void;
-  onPlucked?: (draft: PluckDraft, result: PluckResponse) => void;
+  onPlucked?: (key: string, result: PluckResponse) => void;
   onFailed?: (draft: PluckDraft) => void;
 };
 
 const KINDS: TextKind[] = ["note", "todo", "thing"];
 
 const QUEUE_KEY = "pluck:queue";
-type Queued = { draft: PluckDraft; body: Record<string, unknown> };
 
-// Plucks made offline wait in this browser and go when the network is back.
-function readQueue(): Queued[] {
-  try {
-    return JSON.parse(localStorage.getItem(QUEUE_KEY) ?? "[]") as Queued[];
-  } catch {
-    return [];
-  }
-}
-function writeQueue(q: Queued[]) {
-  try {
-    if (q.length) localStorage.setItem(QUEUE_KEY, JSON.stringify(q));
-    else localStorage.removeItem(QUEUE_KEY);
-  } catch {
-    // storage blocked: the pluck stays on screen as a draft row until reload
-  }
-}
-
-// One flush at a time, across every open tab: mount, the online event and a re-render can all start one. A page that
-// closes mid-send leaves its pluck queued though it may have landed, so each carries a clientKey and a resend of a
-// Note never makes a second one (a link is safe either way). A Web Lock where the browser has one, else one per page.
-let flushing: Promise<void> | null = null;
+// Plucks made offline wait in this browser and go when the network is back. A page that closes mid-send leaves its
+// pluck queued though it may have landed; the queue sends each with a clientKey, so a resend never makes a second Note.
+const queue = createQueue({
+  read: async () => {
+    try {
+      // Entries queued before the shared queue were { draft, body }: the draft's key is their id.
+      const q = JSON.parse(localStorage.getItem(QUEUE_KEY) ?? "[]") as (Queued & { draft?: { key: string } })[];
+      return q.map((x) => (x.id || !x.draft ? x : { id: x.draft.key, body: x.body, at: 0 }));
+    } catch {
+      return [];
+    }
+  },
+  write: async (q) => {
+    try {
+      if (q.length) localStorage.setItem(QUEUE_KEY, JSON.stringify(q));
+      else localStorage.removeItem(QUEUE_KEY);
+    } catch {
+      // storage blocked: the pluck stays on screen as a draft row until reload
+    }
+  },
+});
 
-const post = (body: Record<string, unknown>) =>
+const post = (body: CreateItemBody) =>
   fetch("/api/v1/items", { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(body) });
 
 // The paste box (Flow 3): one link, a block of links, or a sentence that becomes a Note, Todo or Thing. No modal.
@@ -60,24 +60,16 @@ export function TopBar({ sample, emphasis = false, onPluck, onPlucked, onFailed
 
   // Send anything queued while offline, now and whenever the network comes back.
   useEffect(() => {
-    async function send() {
-      for (const q of readQueue()) {
-        const res = await post(q.body).catch(() => null);
-        // Offline, signed out, busy or down: it stays queued, and so does everything after it.
-        if (!res || res.status === 401 || res.status === 429 || res.status >= 500) return;
-        writeQueue(readQueue().filter((x) => x.draft.key !== q.draft.key)); // sent, or refused for good
-        const result = res.ok ? ((await res.json().catch(() => null)) as PluckResponse | null) : null;
-        if (result) onPlucked?.(q.draft, result);
-      }
-    }
+    // One drain at a time across every open tab: mount, the online event and a re-render can all start one.
+    // A Web Lock where the browser has one, else the queue's own guard, one per page.
     function flush() {
-      if (!navigator.onLine || flushing) return;
-      flushing = (navigator.locks
-        ? navigator.locks.request(QUEUE_KEY, { ifAvailable: true }, async (lock) => {
-            if (lock) await send();
-          }).then(() => undefined)
-        : send()
-      ).finally(() => (flushing = null));
+      if (!navigator.onLine) return;
+      const drain = () =>
+        queue.drain(async (body, q) => {
+          const result = await answer<PluckResponse | null>(await post(body)); // a gateway error page is not JSON
+          if (result && q.id) onPlucked?.(q.id, result);
+        });
+      void (navigator.locks ? navigator.locks.request(QUEUE_KEY, { ifAvailable: true }, (lock) => lock && drain()) : drain());
     }
     flush();
     window.addEventListener("online", flush);
@@ -94,7 +86,7 @@ export function TopBar({ sample, emphasis = false, onPluck, onPlucked, onFailed
     setError(null);
     setNotice(null);
     const draft: PluckDraft = { key: `${Date.now()}-${Math.random().toString(36).slice(2, 10)}`, urls, text: isText ? text : null, kind: isText ? kind : null };
-    const body = isText
+    const body: CreateItemBody = isText
       ? { text, kind, source: "web", ...(due && kind !== "note" ? { dueAt: new Date(`${due}T09:00:00`).toISOString() } : {}) }
       : { text, source: "web" };
     setValue("");
@@ -105,7 +97,7 @@ export function TopBar({ sample, emphasis = false, onPluck, onPlucked, onFailed
       if (!navigator.onLine) throw new TypeError("offline");
       res = await post(body);
     } catch {
-      writeQueue([...readQueue(), { draft, body: { ...body, clientKey: draft.key } }]); // the key makes a resend safe
+      await queue.add(body, draft.key); // the draft's row and the queued pluck share a key
       setNotice(t("queued"));
       return;
     }
@@ -115,7 +107,7 @@ export function TopBar({ sample, emphasis = false, onPluck, onPlucked, onFailed
       if (!json) return;
       const result = json as PluckResponse;
       if (result.rejected.length) setError(t("rejected", { count: result.rejected.length, reason: result.rejected[0]!.message }));
-      return onPlucked?.(draft, result);
+      return onPlucked?.(draft.key, result);
     }
     setError(json?.rejected?.[0]?.message ?? json?.error?.message ?? t("failed"));
     setValue(text);

apps/web/next.config.ts

@@ -11,7 +11,7 @@ try {
 const posthog = process.env.POSTHOG_HOST || "https://eu.i.posthog.com";
 
 const config: NextConfig = {
-  transpilePackages: ["@pluck/ui", "@pluck/shared-schemas", "@pluck/db", "@pluck/extract", "@pluck/blob", "@pluck/i18n"],
+  transpilePackages: ["@pluck/ui", "@pluck/shared-schemas", "@pluck/db", "@pluck/extract", "@pluck/blob", "@pluck/i18n", "@pluck/client"],
   // Parsers with large or native-ish dependency trees run from node_modules instead of the bundle.
   serverExternalPackages: ["defuddle", "linkedom", "unpdf", "@mozilla/readability", "undici"],
   // Monorepo root, so env and tracing resolve from there.

apps/web/package.json

@@ -20,6 +20,7 @@
     "@better-auth/oauth-provider": "1.7.12",
     "@modelcontextprotocol/server": "2.4.0",
     "@pluck/blob": "workspace:*",
+    "@pluck/client": "workspace:*",
     "@pluck/db": "workspace:*",
     "@pluck/extract": "workspace:*",
     "@pluck/i18n": "workspace:*",

docs/architecture.md

@@ -35,7 +35,7 @@ One Next.js app (`apps/web`) is the website, the signed-in app, the REST API for
 | `packages/extract` | Guarded fetch (`fetch.ts`), page and PDF reading (`extract.ts`, Defuddle then Readability), email (`email.ts`) and imports. Its fixtures are served by `pnpm stack` so tests never touch the internet. |
 | `packages/blob` | S3 through the AWS SDK; MinIO locally and for self-hosting. |
 | `packages/shared-schemas` | Every type as a Zod schema, used by the server and every client. |
-| `packages/client` | What the iPhone app and the extension share: `createClient` (bearer auth, refresh a minute early, errors mapped to `Unreachable`, `SignedOut` or `Refused`), `createQueue` (the offline queue) and PKCE. |
+| `packages/client` | What the iPhone app and the extension share: `createClient` (bearer auth, refresh a minute early, errors mapped to `Unreachable`, `SignedOut` or `Refused` by `answer`), `createQueue` (the offline queue) and PKCE. The web's offline queue is `createQueue` too, over `localStorage`; the iPhone app drains the share extension's App Group files with `createDrain`, the same keep and drop rules. |
 | `packages/i18n` | The words, in ten languages, one catalog per surface. See [i18n.md](i18n.md). |
 | `packages/ui` | Tokens, the logo, the icon and the Kind glyphs. |
 | `apps/extension` | WXT. The popup, options and a background worker that drains the queue every minute. |
@@ -99,12 +99,6 @@ These are the places where the code is shallower than it needs to be: a rule wri
 - **Problem:** this is where a newcomer gets lost. Two retry policies live in two files, and which one governs a failed extraction is not obvious from either. The cycle between the library and the jobs is a sign the seam is in the wrong place.
 - **Shape:** one `lib/library/keeping.ts` with `keepNow(userId, item, { html?, transcript?, timeoutMs })`, `keepLater(userId, itemIds)` and `keepFile(userId, item)`. It owns the retry policy and the mapping from what `packages/extract` returns to what `packages/db` stores; `jobs.ts` becomes a plain table of handlers. Pass the "run the queue now" kick in, instead of the library importing jobs. Tests cross `keepNow` with the fixture server: a permanent 4xx fails at once, a transient failure retries the stated number of times.
 
-### 4. One offline queue for every client (M)
-
-- **Files:** `packages/client/src/queue.ts` (`createQueue`, used by the extension and the iPhone app, well tested), `apps/web/components/TopBar.tsx` (its own `localStorage` queue), `apps/ios/lib/data.ts` (a second drain loop over the share extension's App Group files), `apps/ios/targets/share/PluckClient.swift` (writes those files).
-- **Problem:** the web keeps its own copy of the queue. Two flushes at once are now kept apart by a Web Lock (`TopBar.tsx`), the guard `client.test.ts` tests for the shared queue ("two drains at once send each note once"), but it is still a second implementation of the same rules. The iPhone app drains the App Group files with a third, hand-written copy.
-- **Shape:** make `createQueue<T>(store)` generic over what an entry carries, then add two store adapters: `localStorage` for the web and the App Group folder for the iPhone. Two adapters make it a real seam. Swift cannot share TypeScript, so pin the file format with a JSON fixture that a TS test parses and the Swift code names in a comment.
-
 ### 5. One session module for the extension and the iPhone app (M)
 
 - **Files:** `apps/extension/lib/auth.ts`, `apps/ios/lib/session.ts`, `apps/ios/targets/share/PluckClient.swift`, and the readers of raw storage keys in `apps/extension/entrypoints/popup/Popup.tsx` and `lib/pluck.ts`.

packages/client/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@pluck/client",
-  "description": "The API client, offline queue and PKCE shared by the Pluck iPhone app and browser extension.",
+  "description": "The API client, offline queue and PKCE shared by the Pluck iPhone app and browser extension; the web uses the queue too.",
   "repository": { "type": "git", "url": "https://github.com/codu-code/pluck.git", "directory": "packages/client" },
   "version": "0.0.0",
   "private": true,

packages/client/src/api.ts

@@ -38,17 +38,20 @@ export function createClient({ server, auth, onSignedOut, fetch: f = fetch, time
     } catch {
       throw new Unreachable("Couldn't reach Pluck.");
     }
-    if (res.status === 401) {
-      await onSignedOut();
-      throw new SignedOut("Sign in again to pluck.");
-    }
-    if (res.status >= 500 || res.status === 429) throw new Unreachable("Couldn't reach Pluck.");
-    const body = (await res.json().catch(() => null)) as ({ error?: { code: string; message: string; details?: unknown } } & T) | null;
-    if (!res.ok) throw new Refused(body?.error?.code ?? "error", body?.error?.message ?? "That did not work.", res.status, body?.error?.details);
-    return body as T;
+    if (res.status === 401) await onSignedOut();
+    return answer<T>(res);
   };
 }
 
+// What a response means: its JSON, or the error the queue and the screens act on. The web reads its own fetches with it.
+export async function answer<T>(res: Response): Promise<T> {
+  if (res.status === 401) throw new SignedOut("Sign in again to pluck.");
+  if (res.status >= 500 || res.status === 429) throw new Unreachable("Couldn't reach Pluck.");
+  const body = (await res.json().catch(() => null)) as ({ error?: { code: string; message: string; details?: unknown } } & T) | null;
+  if (!res.ok) throw new Refused(body?.error?.code ?? "error", body?.error?.message ?? "That did not work.", res.status, body?.error?.details);
+  return body as T;
+}
+
 export type Tokens = { access: string; refresh: string | null; expiresAt: number };
 
 // The OAuth token endpoint. Null means the server said no (the code or refresh token is spent: sign in again).

packages/client/src/client.test.ts

@@ -77,6 +77,21 @@ describe("the offline queue under load", () => {
     expect(sent).toEqual(["a note", "another"]);
   });
 
+  test.each([
+    ["signed out", new SignedOut("sign in")],
+    ["offline or down", new Unreachable("down")],
+    ["anything else unexpected", new TypeError("fetch failed")],
+  ])("%s: the pluck is kept, with its key, for next time", async (_, error) => {
+    const { box, q } = store();
+    await q.add({ text: "a note" }, "draft-1");
+    const keys: (string | undefined)[] = [];
+    expect(await q.drain(async (b) => { keys.push(b.clientKey); throw error; })).toBe(1);
+    expect(box.saved).toMatchObject([{ id: "draft-1", body: { text: "a note" } }]);
+    await q.drain(async (b) => void keys.push(b.clientKey));
+    expect(keys).toEqual(["draft-1", "draft-1"]); // the same key every time, so a resend makes no second Note
+    expect(box.saved).toEqual([]);
+  });
+
   test("the first pluck that cannot be sent stops the drain", async () => {
     const { q } = store();
     for (const u of ["a", "b", "c"]) await q.add({ url: `https://${u}.test` });

packages/client/src/index.ts

@@ -1,3 +1,3 @@
-export { createClient, Refused, SignedOut, tokenRequest, Unreachable, type Client, type ClientDeps, type Init, type Tokens } from "./api";
-export { createQueue, type Queued, type QueueStore } from "./queue";
+export { answer, createClient, Refused, SignedOut, tokenRequest, Unreachable, type Client, type ClientDeps, type Init, type Tokens } from "./api";
+export { createDrain, createQueue, type Queued, type QueueStore } from "./queue";
 export { b64url, codeFrom, pkcePair } from "./pkce";