From 8c2eefabde9eb95d8f7ea2c03b3469bc8857a4ed Mon Sep 17 00:00:00 2001 From: NiallJoeMaher Date: Tue, 27 Oct 2026 07:20:09 +0000 Subject: [PATCH] Plucks waiting on a device follow one set of rules: the web, the extension, the iPhone app and its shares all drain through createQueue's Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01Per4f8yL5A7nL9frZVDJj7 Signed-off-by: NiallJoeMaher --- apps/ios/lib/data.ts | 50 +++++++++--------- apps/web/components/ItemList.tsx | 4 +- apps/web/components/TopBar.tsx | 82 ++++++++++++++---------------- apps/web/next.config.ts | 2 +- apps/web/package.json | 1 + docs/architecture.md | 8 +-- packages/client/package.json | 2 +- packages/client/src/api.ts | 19 ++++--- packages/client/src/client.test.ts | 15 ++++++ packages/client/src/index.ts | 4 +- packages/client/src/queue.ts | 51 +++++++++++-------- pnpm-lock.yaml | 5 +- 12 files changed, 128 insertions(+), 115 deletions(-) diff --git a/apps/ios/lib/data.ts b/apps/ios/lib/data.ts index 6b4b323..ae06d43 100644 --- a/apps/ios/lib/data.ts +++ b/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 { } 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({ + 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 | null = null; -export const syncPending = (): Promise => (syncing ??= sync().finally(() => (syncing = null))); - -async function sync(): Promise { +// 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 { 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(); diff --git a/apps/web/components/ItemList.tsx b/apps/web/components/ItemList.tsx index 6362632..f6ea75e 100644 --- a/apps/web/components/ItemList.tsx +++ b/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)); diff --git a/apps/web/components/TopBar.tsx b/apps/web/components/TopBar.tsx index 54aef57..78d0025 100644 --- a/apps/web/components/TopBar.tsx +++ b/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 }; -// 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 | 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) => +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(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); diff --git a/apps/web/next.config.ts b/apps/web/next.config.ts index 32ebfeb..045f3d8 100644 --- a/apps/web/next.config.ts +++ b/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. diff --git a/apps/web/package.json b/apps/web/package.json index b7b934b..09e9717 100644 --- a/apps/web/package.json +++ b/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:*", diff --git a/docs/architecture.md b/docs/architecture.md index 3510183..5af471d 100644 --- a/docs/architecture.md +++ b/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(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`. diff --git a/packages/client/package.json b/packages/client/package.json index 877bfb4..e8fa7a0 100644 --- a/packages/client/package.json +++ b/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, diff --git a/packages/client/src/api.ts b/packages/client/src/api.ts index b3271ba..44dfe71 100644 --- a/packages/client/src/api.ts +++ b/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(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(res: Response): Promise { + 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). diff --git a/packages/client/src/client.test.ts b/packages/client/src/client.test.ts index 5bb3ae0..b1700eb 100644 --- a/packages/client/src/client.test.ts +++ b/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` }); diff --git a/packages/client/src/index.ts b/packages/client/src/index.ts index 074d222..9709261 100644 --- a/packages/client/src/index.ts +++ b/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"; diff --git a/packages/client/src/queue.ts b/packages/client/src/queue.ts index fe11fbc..8c39168 100644 --- a/packages/client/src/queue.ts +++ b/packages/client/src/queue.ts @@ -7,33 +7,40 @@ export type QueueStore = { read(): Promise; write(q: Queued[]): Promis const keyOf = (q: Queued) => q.id ?? String(q.at); // entries queued before ids existed go by their time -export function createQueue(store: QueueStore) { +// The rules for everything waiting on a device, whatever holds it. Sends in order, one at a time. An entry leaves once +// it is sent, or refused (retrying will not change that). Anything else (offline, signed out, busy, down) keeps it and +// stops the drain: the rest would fail the same way. One drain runs at a time; a second call gets the first's result. +export function createDrain(store: { read(): Promise; remove(e: E): Promise }) { let draining: Promise | null = null; + return (send: (e: E) => Promise): Promise => + (draining ??= (async () => { + try { + for (const e of await store.read()) { + try { + await send(e); + } catch (err) { + if (!(err instanceof Refused)) break; + } + await store.remove(e); + } + return (await store.read()).length; + } finally { + draining = null; + } + })()); +} + +export function createQueue(store: QueueStore) { + // Removing writes back a fresh read, so plucks queued meanwhile stay. + const drain = createDrain({ read: store.read, remove: async (q) => store.write((await store.read()).filter((x) => keyOf(x) !== keyOf(q))) }); return { count: async () => (await store.read()).length, - add: async (body: CreateItemBody, now = Date.now()) => { - const q = [...(await store.read()), { id: `${now}-${Math.random().toString(36).slice(2, 10)}`, body, at: now }]; + add: async (body: CreateItemBody, id = `${Date.now()}-${Math.random().toString(36).slice(2, 10)}`) => { + const q = [...(await store.read()), { id, body, at: Date.now() }]; await store.write(q); return q.length; }, - // Sends in order, one at a time. An entry leaves the queue once it is sent, or refused (retrying will not change - // that), by writing back a fresh read, so plucks queued meanwhile stay. The first one that cannot be sent stops - // the drain: the rest would fail the same way. One drain runs at a time; a second call gets the first's result. - drain: (send: (body: CreateItemBody) => Promise): Promise => - (draining ??= (async () => { - try { - for (const q of await store.read()) { - try { - await send(q.id ? { ...q.body, clientKey: q.id } : q.body); // the key makes a resend safe - } catch (e) { - if (!(e instanceof Refused)) break; - } - await store.write((await store.read()).filter((x) => keyOf(x) !== keyOf(q))); - } - return (await store.read()).length; - } finally { - draining = null; - } - })()), + drain: (send: (body: CreateItemBody, q: Queued) => Promise) => + drain((q) => send(q.id ? { ...q.body, clientKey: q.id } : q.body, q)), // the key makes a resend safe }; } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index cf59353..64e5b48 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -241,6 +241,9 @@ importers: '@pluck/blob': specifier: workspace:* version: link:../../packages/blob + '@pluck/client': + specifier: workspace:* + version: link:../../packages/client '@pluck/db': specifier: workspace:* version: link:../../packages/db @@ -7640,9 +7643,7 @@ snapshots: metro-runtime: 0.87.1 transitivePeerDependencies: - '@babel/core' - - bufferutil - supports-color - - utf-8-validate '@react-native/normalize-colors@0.79.7': {}