diff --git a/apps/web/lib/capture.ts b/apps/web/lib/capture.ts deleted file mode 100644 index 7876e08..0000000 --- a/apps/web/lib/capture.ts +++ /dev/null @@ -1,108 +0,0 @@ -import { getObject, keys, putObject } from "@pluck/blob"; -import { applyExtraction, enqueue, nextUnsorted, recordExtractionFailure, setBlobKeys, type ExtractionResult } from "@pluck/db"; -import { extract, FetchError, guardedFetch, type Extracted } from "@pluck/extract"; -import type { Item } from "@pluck/shared-schemas"; -import { log } from "./log"; - -// The page-keeping pipeline shared by the inline save and the retry worker. - -export const MAX_EXTRACTION_ATTEMPTS = 4; // the inline try plus three retries -const BACKOFF_MS = [30_000, 120_000, 600_000]; -export const backoffMs = (attempt: number) => BACKOFF_MS[Math.min(attempt - 1, BACKOFF_MS.length - 1)]!; - -// A 4xx (other than 408 and 429) or a refused address will not get better by waiting. -export function isPermanent(e: unknown): boolean { - if (!(e instanceof FetchError)) return false; - if (e.status) return e.status >= 400 && e.status < 500 && e.status !== 408 && e.status !== 429; - return /blocked|unsupported|over \d+ bytes/.test(e.message); -} - -const errorText = (e: unknown) => (e instanceof Error ? (e.name === "TimeoutError" ? "Timed out" : e.message) : String(e)); - -const toResult = (x: Extracted): ExtractionResult => ({ - kind: x.kind, preview: x.preview, facts: x.facts, title: x.title, siteName: x.siteName, author: x.author, excerpt: x.excerpt, - canonicalUrl: x.canonicalUrl, lang: x.lang, wordCount: x.wordCount, markdown: x.markdown, extractor: x.extractor, -}); - -// Pictures only: an SVG can carry script, and a self-host may serve its blob store from its own domain. -const picture = (contentType: string) => /^image\/(jpeg|png|webp|gif|avif)\b/i.test(contentType); - -// Raw page, file and cover to the blob store. Covers go through the same guarded fetch as pages. -export async function storeBlobs(userId: string, itemId: string, x: Pick): Promise { - const update: { coverKey?: string; htmlKey?: string } = {}; - if (x.raw && (x.kind !== "image" || picture(x.raw.contentType))) { - const key = x.kind === "pdf" ? keys.file(userId, itemId, "pdf") : x.kind === "image" ? keys.cover(userId, itemId) : keys.html(userId, itemId); - await putObject(key, x.raw.body, x.raw.contentType); - if (x.kind === "image") update.coverKey = key; - else if (x.kind !== "pdf") update.htmlKey = key; - } - if (x.coverUrl && !update.coverKey) { - try { - const img = await guardedFetch(x.coverUrl, { maxBytes: 5_000_000, accept: "image/*", signal: AbortSignal.timeout(8000) }); - if (picture(img.contentType)) { - await putObject(keys.cover(userId, itemId), img.body, img.contentType); - update.coverKey = keys.cover(userId, itemId); - } - } catch (e) { - log("warn", "cover fetch failed", { userId, itemId, error: errorText(e) }); - } - } - if (update.coverKey || update.htmlKey) await setBlobKeys(userId, itemId, update); -} - -type Defer = (task: () => Promise) => void; - -// Captions the browser read go after the page text, so search and the AI tool read what was said. -function withTranscript(x: Extracted, transcript?: string): Extracted { - const t = transcript?.trim(); - if (!t) return x; - return { ...x, markdown: `${x.markdown}\n\n## Transcript\n\n${t}`, wordCount: (x.wordCount ?? 0) + t.split(/\s+/).length }; -} - -// One extraction attempt. On success the item is updated and blobs are stored (deferred when a caller -// is waiting). On failure the attempt is recorded and, unless it is final, a retry is queued. -export async function extractAttempt( - userId: string, - item: Item, - { html, transcript, timeoutMs, defer }: { html?: string | null; transcript?: string; timeoutMs: number; defer?: Defer }, -): Promise { - const started = Date.now(); - try { - // Shown as an article by the person's choice: its body is kept, whatever the link looks like. - const asArticle = item.preview === "reader" && item.previewGuess !== "reader"; - const named = item.titleIsDerived && item.title !== item.url ? item.title : null; // the device's, or an earlier read's - const x = withTranscript(await extract({ url: item.url!, html, title: named }, { signal: AbortSignal.timeout(timeoutMs), asArticle }), transcript); - const updated = (await applyExtraction(userId, item.id, toResult(x))) ?? item; - const blobs = () => - storeBlobs(userId, item.id, x).catch(async (e) => { - log("warn", "blob store failed", { userId, itemId: item.id, error: errorText(e) }); - }); - if (defer) defer(blobs); - else await blobs(); - log("info", "extracted", { userId, itemId: item.id, durationMs: Date.now() - started, extractor: x.extractor }); - return updated; - } catch (e) { - const attempt = item.extraction.attempts + 1; - const final = isPermanent(e) || attempt >= MAX_EXTRACTION_ATTEMPTS; - const updated = (await recordExtractionFailure(userId, item.id, errorText(e), final)) ?? item; - if (!final) await enqueue("extract", { userId, itemId: item.id }, { delayMs: backoffMs(attempt) }); - log("warn", "extraction failed", { userId, itemId: item.id, durationMs: Date.now() - started, attempt, final, error: errorText(e) }); - return updated; - } -} - -// Items read before Previews, sorted again a few at a time from the page kept with them, so the sites see nothing; -// one with no kept page is fetched again. A title or Preview the person chose stays theirs (applyExtraction). -export async function sortBackfill(limit = 10): Promise { - const page = await nextUnsorted(limit); - if (!page) return 0; - for (const u of page) { - try { - const html = u.htmlKey ? (await getObject(u.htmlKey)).toString("utf8") : null; - await applyExtraction(u.userId, u.itemId, toResult(await extract({ url: u.url, html, title: u.title }, { signal: AbortSignal.timeout(10_000) }))); - } catch (e) { - log("warn", "preview backfill skipped an item", { userId: u.userId, itemId: u.itemId, error: errorText(e) }); // it keeps its old look - } - } - return page.length; -} diff --git a/apps/web/lib/job-runner.ts b/apps/web/lib/job-runner.ts new file mode 100644 index 0000000..ae2f9a5 --- /dev/null +++ b/apps/web/lib/job-runner.ts @@ -0,0 +1,39 @@ +import { claimJob, completeJob, failJob, retryJob, type Job } from "@pluck/db"; +import { log } from "./log"; + +// The loop that runs queued jobs, apart from the handler table (jobs.ts) so the library can kick it without a cycle. +export type Handlers = Record) => Promise>; + +const MAX_ATTEMPTS = 5; + +// A handler that throws is retried with backoff up to five times. +async function runOne(handlers: Handlers, job: Job): Promise { + const started = Date.now(); + const handler = handlers[job.queue]; + if (!handler) return failJob(job.id, `no handler for ${job.queue}`); + // A job the platform killed mid-run never threw: its lease ran out and it came back. Stop after five of those. + if (job.attempts > MAX_ATTEMPTS) return failJob(job.id, "did not finish in time"); + try { + await handler(job.payload); + await completeJob(job.id); + log("info", "job done", { jobId: job.id, queue: job.queue, durationMs: Date.now() - started }); + } catch (e) { + const error = e instanceof Error ? e.message : String(e); + if (job.attempts >= MAX_ATTEMPTS) await failJob(job.id, error); + else await retryJob(job.id, error, 30_000 * 2 ** (job.attempts - 1)); + log("warn", "job failed", { jobId: job.id, queue: job.queue, attempts: job.attempts, error }); + } +} + +// Claims and runs jobs (of one queue, if named) until none are visible or the time budget is spent. +export async function runQueue(handlers: Handlers, { budgetMs = 45_000, max = 500, queue }: { budgetMs?: number; max?: number; queue?: string } = {}): Promise<{ ran: number }> { + const deadline = Date.now() + budgetMs; + let ran = 0; + while (ran < max && Date.now() < deadline) { + const job = await claimJob(60_000, queue); + if (!job) break; + await runOne(handlers, job); + ran++; + } + return { ran }; +} diff --git a/apps/web/lib/jobs.ts b/apps/web/lib/jobs.ts index f4159cf..a3d9899 100644 --- a/apps/web/lib/jobs.ts +++ b/apps/web/lib/jobs.ts @@ -1,9 +1,7 @@ -import { claimJob, completeJob, failJob, getItem, hitRateLimit, retryJob, type Job } from "@pluck/db"; -import { extractAttempt } from "./capture"; -import { finishFile, ingestInbound, runExport, runImport, sendDigest, sendWelcome } from "./library"; -import { log } from "./log"; +import { hitRateLimit } from "@pluck/db"; +import { runQueue, type Handlers } from "./job-runner"; +import { ingestInbound, keepFile, keepQueued, runExport, runImport, sendDigest, sendWelcome } from "./library"; -const MAX_ATTEMPTS = 5; // Queued mail (Monday emails, welcomes) shares one send rate across every instance, under SES's 14 a second, // so sign-in links (which skip the queue) always have room. const QUEUED_MAIL_PER_SECOND = 8; @@ -11,13 +9,13 @@ const mailSlot = async () => { while (!(await hitRateLimit("send:queued", QUEUED_MAIL_PER_SECOND, 1)).ok) await new Promise((r) => setTimeout(r, 250)); }; -// Worker job handlers. A handler that throws is retried with backoff up to five times. -const handlers: Record) => Promise> = { +// Worker job handlers, by queue. Keeping a page has its own retry policy (library/keeping.ts). +const handlers: Handlers = { async extract(p) { - const userId = String(p.userId); - const item = await getItem(userId, String(p.itemId)); - if (!item || item.extraction.state !== "pending") return; // done, failed, re-plucked or gone - await extractAttempt(userId, item, { timeoutMs: 20_000 }); // records the attempt and queues the next one itself + await keepQueued(String(p.userId), String(p.itemId)); + }, + async file(p) { + await keepFile(String(p.userId), String(p.itemId), String(p.key), String(p.contentType), String(p.name)); }, async digest(p) { await mailSlot(); @@ -33,41 +31,10 @@ const handlers: Record) => Promise { - const started = Date.now(); - const handler = handlers[job.queue]; - if (!handler) return failJob(job.id, `no handler for ${job.queue}`); - // A job the platform killed mid-run never threw: its lease ran out and it came back. Stop after five of those. - if (job.attempts > MAX_ATTEMPTS) return failJob(job.id, "did not finish in time"); - try { - await handler(job.payload); - await completeJob(job.id); - log("info", "job done", { jobId: job.id, queue: job.queue, durationMs: Date.now() - started }); - } catch (e) { - const error = e instanceof Error ? e.message : String(e); - if (job.attempts >= MAX_ATTEMPTS) await failJob(job.id, error); - else await retryJob(job.id, error, 30_000 * 2 ** (job.attempts - 1)); - log("warn", "job failed", { jobId: job.id, queue: job.queue, attempts: job.attempts, error }); - } -} - -// Claims and runs jobs until none are visible or the time budget is spent. Used by the cron tick and the worker loop. -export async function runJobs({ budgetMs = 45_000, max = 500 } = {}): Promise<{ ran: number }> { - const deadline = Date.now() + budgetMs; - let ran = 0; - while (ran < max && Date.now() < deadline) { - const job = await claimJob(60_000); - if (!job) break; - await runOne(job); - ran++; - } - return { ran }; -} +// Used by the cron tick and the worker loop. +export const runJobs = (opts?: { budgetMs?: number; max?: number }) => runQueue(handlers, opts); diff --git a/apps/web/lib/library/account.ts b/apps/web/lib/library/account.ts index c58dd5b..8dee061 100644 --- a/apps/web/lib/library/account.ts +++ b/apps/web/lib/library/account.ts @@ -1,9 +1,9 @@ import { TooLarge, deleteObject, deletePrefix, getObject, putObject, signedPutUrl, signedUrl } from "@pluck/blob"; import { - aiConnections, applyExtraction, claimOnce, countItems, createApiKey, disconnectAi, claimImportStart, createImport, deleteAccountData, enqueue, exportData, finishExport, getImport, getMe, importLink, + aiConnections, claimOnce, countItems, createApiKey, disconnectAi, claimImportStart, createImport, deleteAccountData, enqueue, exportData, finishExport, getImport, getMe, importLink, latestExport, recountTags, listApiKeys, revokeApiKey, startExport, updateImport, } from "@pluck/db"; -import { extractFromHtml, parseImport, type ImportSource } from "@pluck/extract"; +import { parseImport, type ImportSource } from "@pluck/extract"; import { MAX_IMPORT_BYTES, normalizeUrl, type AiConnection, type ApiKey, type ImportStatus } from "@pluck/shared-schemas"; import { strToU8, zipSync } from "fflate"; import { sendEmail } from "../email"; @@ -11,6 +11,7 @@ import { exportEmail, welcomeEmail } from "../emails"; import { ApiError } from "../errors"; import { log } from "../log"; import { ORIGIN } from "../origin"; +import { keepImported } from "./keeping"; // ---- Welcome ------------------------------------------------------------------------------------- @@ -105,18 +106,8 @@ export async function runImport(userId: string, importId: string): Promise } const { item, created, tagIds } = await importLink(userId, url, e); touchedTags.push(...tagIds); - if (created) { - const page = e.html ? await extractFromHtml(e.html(), url).catch(() => null) : null; - if (page && page.wordCount > 0) { - await applyExtraction(userId, item.id, { - kind: page.kind, preview: page.preview, facts: page.facts, title: e.title ?? page.title, siteName: page.siteName, author: page.author, excerpt: page.excerpt, - canonicalUrl: page.canonicalUrl, lang: page.lang, wordCount: page.wordCount, markdown: page.markdown, extractor: page.extractor, - }); - } else { - // Spread the fetching out: a thousand imports should not hit the network at once. - await enqueue("extract", { userId, itemId: item.id }, { delayMs: (done % 1000) * 3000 }); - } - } + // Spread the fetching out: a thousand imports should not hit the network at once. + if (created) await keepImported(userId, item, { html: e.html, title: e.title, delayMs: (done % 1000) * 3000 }); done++; } await recountTags(userId, touchedTags); diff --git a/apps/web/lib/library/files.test.ts b/apps/web/lib/library/files.test.ts index 6be3ac7..9a1cf5b 100644 --- a/apps/web/lib/library/files.test.ts +++ b/apps/web/lib/library/files.test.ts @@ -1,7 +1,8 @@ import { readFileSync } from "node:fs"; +import { pendingJobCount } from "@pluck/db"; import { twoUsers } from "@pluck/db/testing"; import { expect, test } from "vitest"; -import { beginFile, finishFile, readItem, readText } from "./index"; +import { beginFile, keepFile, readItem, readText } from "./index"; const PDF = readFileSync(new URL("../../../../packages/extract/fixtures/pages/doc.pdf", import.meta.url)); @@ -11,13 +12,15 @@ test("a PDF shared from the phone: the item at once, the bytes by signed upload, expect(item).toMatchObject({ kind: "pdf", url: null, source: "ios", extraction: { state: "pending" } }); const key = new URL(uploadUrl).pathname.split("/").slice(2).join("/"); - await expect(finishFile(a, item.id, key, "application/pdf", "Estimates.pdf")).rejects.toThrow(); // not uploaded yet: the job retries + await keepFile(a, item.id, key, "application/pdf", "Estimates.pdf"); // not uploaded yet: a failed try, the next one queued + expect((await readItem(a, item.id)).extraction).toEqual({ state: "pending", attempts: 1, lastError: "The file has not arrived." }); + expect(await pendingJobCount("file")).toBe(2); // beginFile's, and the retry const put = await fetch(uploadUrl, { method: "PUT", headers: { "content-type": "application/pdf" }, body: PDF }); expect(put.ok).toBe(true); - await finishFile(b, item.id, key, "application/pdf", "Estimates.pdf"); // someone else's item: nothing happens + await keepFile(b, item.id, key, "application/pdf", "Estimates.pdf"); // someone else's item: nothing happens expect((await readItem(a, item.id)).extraction.state).toBe("pending"); - await finishFile(a, item.id, key, "application/pdf", "Estimates.pdf"); + await keepFile(a, item.id, key, "application/pdf", "Estimates.pdf"); expect((await readItem(a, item.id)).extraction.state).toBe("done"); expect((await readText(a, item.id)).length).toBeGreaterThan(20); }); diff --git a/apps/web/lib/library/files.ts b/apps/web/lib/library/files.ts index be0d772..57d5439 100644 --- a/apps/web/lib/library/files.ts +++ b/apps/web/lib/library/files.ts @@ -1,13 +1,12 @@ -import { TooLarge, deleteObject, getObject, keys, signedPutUrl } from "@pluck/blob"; -import { applyExtraction, createFileItem, enqueue, getItem, recordExtractionFailure, setBlobKeys } from "@pluck/db"; -import { extractFile } from "@pluck/extract"; -import { FILE_TYPES, MAX_FILE_BYTES, type CreateFileItemBody, type Item } from "@pluck/shared-schemas"; +import { keys, signedPutUrl } from "@pluck/blob"; +import { createFileItem, enqueue } from "@pluck/db"; +import { FILE_TYPES, type CreateFileItemBody, type Item } from "@pluck/shared-schemas"; const keyFor = (userId: string, itemId: string, contentType: CreateFileItemBody["contentType"]) => contentType === "application/pdf" ? keys.file(userId, itemId, "pdf") : keys.file(userId, itemId, FILE_TYPES[contentType]); // An Image or PDF from the phone's share sheet. The item lands at once; the bytes go straight to blob storage -// (a background upload that may finish after the extension has gone), and the worker reads them once they are there. +// (a background upload that may finish after the extension has gone), and keepFile reads them once they are there. export async function beginFile(userId: string, body: CreateFileItemBody): Promise<{ item: Item; uploadUrl: string }> { const kind = body.contentType === "application/pdf" ? "pdf" : "image"; const item = await createFileItem(userId, { kind, title: body.name, source: body.source ?? "ios" }); @@ -15,25 +14,3 @@ export async function beginFile(userId: string, body: CreateFileItemBody): Promi await enqueue("file", { userId, itemId: item.id, key, contentType: body.contentType, name: body.name }, { delayMs: 15_000 }); return { item, uploadUrl: await signedPutUrl(key, body.contentType) }; } - -// The worker's half. Throws while the upload has not arrived, so the job retries with backoff. -export async function finishFile(userId: string, itemId: string, key: string, contentType: string, name: string): Promise { - const item = await getItem(userId, itemId); - if (!item || item.extraction.state !== "pending") return; - // shortcut: an upload that never arrives leaves the item pending once the job gives up (five tries, about 15 minutes); mark it failed if that shows up. - let body: Buffer; - try { - body = await getObject(key, { maxBytes: MAX_FILE_BYTES }); // missing yet: throws, and the job tries again later - } catch (e) { - if (!(e instanceof TooLarge)) throw e; - await recordExtractionFailure(userId, itemId, "That file is over 50 MB.", true); - await deleteObject(key); - return; - } - const x = await extractFile({ body, contentType, fileName: name }); - if (x.kind === "image") await setBlobKeys(userId, itemId, { coverKey: key }); - await applyExtraction(userId, itemId, { - kind: x.kind, preview: x.preview, facts: x.facts, title: x.title, siteName: x.siteName, author: x.author, excerpt: x.excerpt, canonicalUrl: null, lang: x.lang, - wordCount: x.wordCount, markdown: x.markdown, extractor: x.extractor, - }); -} diff --git a/apps/web/lib/library/index.ts b/apps/web/lib/library/index.ts index f537a09..72a412a 100644 --- a/apps/web/lib/library/index.ts +++ b/apps/web/lib/library/index.ts @@ -12,3 +12,4 @@ export * from "./email-in"; export * from "./account"; export * from "./tries"; export * from "./files"; +export * from "./keeping"; diff --git a/apps/web/lib/library/items.ts b/apps/web/lib/library/items.ts index 0274ca9..35dd9ca 100644 --- a/apps/web/lib/library/items.ts +++ b/apps/web/lib/library/items.ts @@ -1,6 +1,6 @@ import { deletePrefix, keys as blobKeys, putObject, signedUrl } from "@pluck/blob"; import { - batchTags, batchUpdate, countItems, createHighlight, createTextItem, deleteHighlight, editItem, enqueue, findByUrl, getContent, getItem, getItemBlobKeys, + batchTags, batchUpdate, countItems, createHighlight, createTextItem, deleteHighlight, editItem, findByUrl, getContent, getItem, getItemBlobKeys, hitRateLimit, listHighlights, listItems, listsForItem, markOpened, moveToBin, pluckLink, purgeBin, purgeFromBin, resetExtraction, restoreFromBin, setBlobKeys, setItemTags, updateHighlight, } from "@pluck/db"; @@ -9,7 +9,7 @@ import { type Highlight, type Item, type ItemPage, type ListItemsQuery, type ListMembership, type PluckResponse, type Status, type UpdateItemBody, } from "@pluck/shared-schemas"; import { ApiError } from "../errors"; -import { extractAttempt } from "../capture"; +import { keepLater, keepNow } from "./keeping"; type Defer = (task: () => Promise) => void; const INLINE_TIMEOUT_MS = 5000; @@ -52,27 +52,16 @@ export async function pluckAny(userId: string, body: CreateItemBody, { defer }: results.push({ item, created }); if (needsExtraction) toExtract.push(item); } - await queueExtraction(userId, toExtract, { defer }); + await keepLater(userId, toExtract, { defer }); return { results, rejected }; } -// Many pages are read by the job queue, so a function cut off at 60 s leaves none of them behind; -// `defer` starts on them as soon as the response is out, and the cron tick carries on with the rest. -export async function queueExtraction(userId: string, items: Item[], { defer }: { defer?: Defer } = {}): Promise { - for (const item of items) await enqueue("extract", { userId, itemId: item.id }); - if (!items.length) return; - const { runJobs } = await import("../jobs"); // jobs imports the library, so not at the top - const work = async () => void (await runJobs({ budgetMs: 40_000 })); - if (defer) defer(work); - else await work(); -} - export async function pluck(userId: string, body: CreateItemBody & { url: string }, { defer }: { defer?: Defer } = {}): Promise { const url = normalizeUrl(body.url); if (!url) throw new ApiError("bad_request", "That is not a link Pluck can open.", 400); const { item, created, needsExtraction } = await pluckLink(userId, { url, title: body.title, note: body.note, source: body.source ?? "web" }); const kept = needsExtraction || body.html || body.transcript - ? await extractAttempt(userId, item, { html: body.html, transcript: body.transcript, timeoutMs: INLINE_TIMEOUT_MS, defer }) + ? await keepNow(userId, item, { html: body.html, transcript: body.transcript, timeoutMs: INLINE_TIMEOUT_MS, defer }) : item; // A selection becomes a highlight anchored by its quote; the reader finds it in the kept text. const quote = body.selection?.trim(); @@ -98,7 +87,7 @@ export async function setCoverFromApp(userId: string, itemId: string, body: Uint export async function retryExtraction(userId: string, itemId: string, { defer }: { defer?: Defer } = {}): Promise { const item = await resetExtraction(userId, itemId); if (!item) throw new ApiError("not_found", "No such item.", 404); - return extractAttempt(userId, item, { timeoutMs: INLINE_TIMEOUT_MS, defer }); + return keepNow(userId, item, { timeoutMs: INLINE_TIMEOUT_MS, defer }); } export async function readItem(userId: string, itemId: string): Promise { diff --git a/apps/web/lib/library/keeping.test.ts b/apps/web/lib/library/keeping.test.ts new file mode 100644 index 0000000..5d17793 --- /dev/null +++ b/apps/web/lib/library/keeping.test.ts @@ -0,0 +1,55 @@ +import { getContent, pendingJobCount, pluckLink } from "@pluck/db"; +import { makeJobsVisible, twoUsers } from "@pluck/db/testing"; +import { expect, test } from "vitest"; +import { runJobs } from "../jobs"; +import { keepLater, keepNow, readItem } from "./index"; + +const FIX = "http://127.0.0.1:4180"; +const link = async (userId: string, url: string) => (await pluckLink(userId, { url, source: "web" })).item; + +test("kept now: the page's text is kept and nothing is queued", async () => { + const { a } = await twoUsers(); + const item = await keepNow(a, await link(a, `${FIX}/news-1.html`), { timeoutMs: 5000 }); + expect(item.extraction).toMatchObject({ state: "done" }); + expect(item.title).toBe("Small teams ship when the list is short"); + expect((await getContent(a, item.id))?.textMarkdown).toContain("finishing problem"); + expect(await pendingJobCount("extract")).toBe(0); +}); + +test("kept later: the kick reads the queued pages after the response, and a failure there is retried", async () => { + const { a } = await twoUsers(); + const item = await link(a, `${FIX}/flaky?n=later-${Date.now()}`); + const deferred: (() => Promise)[] = []; + await keepLater(a, [item], { defer: (task) => void deferred.push(task) }); + expect(await pendingJobCount("extract")).toBe(1); + for (const task of deferred) await task(); + expect((await readItem(a, item.id)).extraction).toEqual({ state: "pending", attempts: 1, lastError: "HTTP 503" }); + + await makeJobsVisible(); + await runJobs({ budgetMs: 10_000 }); + const after = await readItem(a, item.id); + expect(after.extraction.state).toBe("done"); + expect(after.title).toMatch(/^Back on its feet/); + expect(await pendingJobCount()).toBe(0); +}); + +test("a page that keeps failing gets four tries, then the failure is final on the Item", async () => { + const { a } = await twoUsers(); + let item = await keepNow(a, await link(a, `${FIX}/error?n=${Date.now()}`), { timeoutMs: 5000 }); + for (let i = 0; i < 3; i++) { + expect(item.extraction).toEqual({ state: "pending", attempts: i + 1, lastError: "HTTP 500" }); + expect(await pendingJobCount("extract")).toBe(1); + await makeJobsVisible(); + await runJobs({ budgetMs: 10_000 }); + item = await readItem(a, item.id); + } + expect(item.extraction).toEqual({ state: "failed", attempts: 4, lastError: "HTTP 500" }); + expect(await pendingJobCount()).toBe(0); +}); + +test("a 4xx is final at once", async () => { + const { a } = await twoUsers(); + const item = await keepNow(a, await link(a, `${FIX}/missing-page.html`), { timeoutMs: 5000 }); + expect(item.extraction).toEqual({ state: "failed", attempts: 1, lastError: "HTTP 404" }); + expect(await pendingJobCount()).toBe(0); +}); diff --git a/apps/web/lib/library/keeping.ts b/apps/web/lib/library/keeping.ts new file mode 100644 index 0000000..6746d4c --- /dev/null +++ b/apps/web/lib/library/keeping.ts @@ -0,0 +1,163 @@ +import { TooLarge, deleteObject, getObject, keys, putObject } from "@pluck/blob"; +import { applyExtraction, enqueue, getItem, nextUnsorted, recordExtractionFailure, setBlobKeys, type ExtractionResult } from "@pluck/db"; +import { extract, extractFile, extractFromHtml, FetchError, guardedFetch, type Extracted } from "@pluck/extract"; +import { MAX_FILE_BYTES, type Item } from "@pluck/shared-schemas"; +import { runQueue } from "../job-runner"; +import { log } from "../log"; + +// Keeping an Item's page: now (a pluck waits up to `timeoutMs`), later (the job queue), from a file, or from an import. +// +// The one retry policy, counted on the Item (`extraction.attempts`): four tries in all, the later three 30 s, 2 min and +// 10 min after the one before failed; the fourth failure is final. A 4xx (other than 408 and 429), a refused address or a +// file over the limit is final at once. The job queue's own retries (job-runner.ts) only re-run a try whose outcome never +// got written, such as when the database is down. +const MAX_TRIES = 4; +const BACKOFF_MS = [30_000, 120_000, 600_000]; +const QUEUED_TIMEOUT_MS = 20_000; + +type Defer = (task: () => Promise) => void; + +function isPermanent(e: unknown): boolean { + if (e instanceof TooLarge) return true; + if (!(e instanceof FetchError)) return false; + if (e.status) return e.status >= 400 && e.status < 500 && e.status !== 408 && e.status !== 429; + return /blocked|unsupported|over \d+ bytes/.test(e.message); +} + +const errorText = (e: unknown) => + e instanceof TooLarge ? "That file is over 50 MB." + : !(e instanceof Error) ? String(e) + : e.name === "TimeoutError" ? "Timed out" + : e.name === "NoSuchKey" ? "The file has not arrived." // the phone's upload, still on its way + : e.message; + +// The one mapping from what packages/extract read to what packages/db keeps. +const toResult = (x: Extracted): ExtractionResult => ({ + kind: x.kind, preview: x.preview, facts: x.facts, title: x.title, siteName: x.siteName, author: x.author, excerpt: x.excerpt, + canonicalUrl: x.canonicalUrl, lang: x.lang, wordCount: x.wordCount, markdown: x.markdown, extractor: x.extractor, +}); + +// A failed try: recorded on the Item and, unless it is final, the next one queued. +async function failed(userId: string, item: Item, e: unknown, retry: { queue: string; payload: Record }, started: number): Promise { + const attempt = item.extraction.attempts + 1; + const final = isPermanent(e) || attempt >= MAX_TRIES; + const updated = (await recordExtractionFailure(userId, item.id, errorText(e), final)) ?? item; + if (!final) await enqueue(retry.queue, retry.payload, { delayMs: BACKOFF_MS[attempt - 1] }); + log("warn", "extraction failed", { userId, itemId: item.id, durationMs: Date.now() - started, attempt, final, error: errorText(e) }); + return updated; +} + +// Pictures only: an SVG can carry script, and a self-host may serve its blob store from its own domain. +const picture = (contentType: string) => /^image\/(jpeg|png|webp|gif|avif)\b/i.test(contentType); + +// Raw page, file and cover to the blob store. Covers go through the same guarded fetch as pages. +async function storeBlobs(userId: string, itemId: string, x: Pick): Promise { + const update: { coverKey?: string; htmlKey?: string } = {}; + if (x.raw && (x.kind !== "image" || picture(x.raw.contentType))) { + const key = x.kind === "pdf" ? keys.file(userId, itemId, "pdf") : x.kind === "image" ? keys.cover(userId, itemId) : keys.html(userId, itemId); + await putObject(key, x.raw.body, x.raw.contentType); + if (x.kind === "image") update.coverKey = key; + else if (x.kind !== "pdf") update.htmlKey = key; + } + if (x.coverUrl && !update.coverKey) { + try { + const img = await guardedFetch(x.coverUrl, { maxBytes: 5_000_000, accept: "image/*", signal: AbortSignal.timeout(8000) }); + if (picture(img.contentType)) { + await putObject(keys.cover(userId, itemId), img.body, img.contentType); + update.coverKey = keys.cover(userId, itemId); + } + } catch (e) { + log("warn", "cover fetch failed", { userId, itemId, error: errorText(e) }); + } + } + if (update.coverKey || update.htmlKey) await setBlobKeys(userId, itemId, update); +} + +// Captions the browser read go after the page text, so search and the AI tool read what was said. +function withTranscript(x: Extracted, transcript?: string): Extracted { + const t = transcript?.trim(); + if (!t) return x; + return { ...x, markdown: `${x.markdown}\n\n## Transcript\n\n${t}`, wordCount: (x.wordCount ?? 0) + t.split(/\s+/).length }; +} + +// One try at a link's page. On success the Item is updated and blobs are stored (deferred when a caller is waiting). +export async function keepNow( + userId: string, + item: Item, + { html, transcript, timeoutMs, defer }: { html?: string | null; transcript?: string; timeoutMs: number; defer?: Defer }, +): Promise { + const started = Date.now(); + try { + // Shown as an article by the person's choice: its body is kept, whatever the link looks like. + const asArticle = item.preview === "reader" && item.previewGuess !== "reader"; + const named = item.titleIsDerived && item.title !== item.url ? item.title : null; // the device's, or an earlier read's + const x = withTranscript(await extract({ url: item.url!, html, title: named }, { signal: AbortSignal.timeout(timeoutMs), asArticle }), transcript); + const updated = (await applyExtraction(userId, item.id, toResult(x))) ?? item; + const blobs = () => + storeBlobs(userId, item.id, x).catch(async (e) => { + log("warn", "blob store failed", { userId, itemId: item.id, error: errorText(e) }); + }); + if (defer) defer(blobs); + else await blobs(); + log("info", "extracted", { userId, itemId: item.id, durationMs: Date.now() - started, extractor: x.extractor }); + return updated; + } catch (e) { + return failed(userId, item, e, { queue: "extract", payload: { userId, itemId: item.id } }, started); + } +} + +// The `extract` job: the next try, if the Item still wants one. +export async function keepQueued(userId: string, itemId: string): Promise { + const item = await getItem(userId, itemId); + if (!item || item.extraction.state !== "pending") return; // done, failed, re-plucked or gone + await keepNow(userId, item, { timeoutMs: QUEUED_TIMEOUT_MS }); +} + +// Many pages are read by the job queue, so a function cut off at 60 s leaves none of them behind; +// `defer` starts on them as soon as the response is out, and the cron tick carries on with the rest. +export async function keepLater(userId: string, items: Item[], { defer }: { defer?: Defer } = {}): Promise { + for (const item of items) await enqueue("extract", { userId, itemId: item.id }); + if (!items.length) return; + const work = async () => void (await runQueue({ extract: (p) => keepQueued(String(p.userId), String(p.itemId)) }, { queue: "extract", budgetMs: 40_000 })); + if (defer) defer(work); + else await work(); +} + +// The `file` job: an Image or PDF from the phone's share sheet, read once its upload is in blob storage. +// An upload that has not arrived yet is a failed try like any other, so the same policy waits for it. +export async function keepFile(userId: string, itemId: string, key: string, contentType: string, name: string): Promise { + const item = await getItem(userId, itemId); + if (!item || item.extraction.state !== "pending") return; + const started = Date.now(); + try { + const x = await extractFile({ body: await getObject(key, { maxBytes: MAX_FILE_BYTES }), contentType, fileName: name }); + if (x.kind === "image") await setBlobKeys(userId, itemId, { coverKey: key }); + await applyExtraction(userId, itemId, toResult({ ...x, canonicalUrl: null })); + } catch (e) { + if (e instanceof TooLarge) await deleteObject(key); + await failed(userId, item, e, { queue: "file", payload: { userId, itemId, key, contentType, name } }, started); + } +} + +// An imported link: a page the old app kept is read offline; without one (or with no text in it) it is fetched later. +export async function keepImported(userId: string, item: Item, { html, title, delayMs }: { html: (() => string) | null; title: string | null; delayMs: number }): Promise { + const page = html ? await extractFromHtml(html(), item.url!).catch(() => null) : null; + if (page && page.wordCount > 0) await applyExtraction(userId, item.id, toResult({ ...page, title: title ?? page.title })); + else await enqueue("extract", { userId, itemId: item.id }, { delayMs }); +} + +// Items read before Previews, sorted again a few at a time from the page kept with them, so the sites see nothing; +// one with no kept page is fetched again. A title or Preview the person chose stays theirs (applyExtraction). +export async function sortBackfill(limit = 10): Promise { + const page = await nextUnsorted(limit); + if (!page) return 0; + for (const u of page) { + try { + const html = u.htmlKey ? (await getObject(u.htmlKey)).toString("utf8") : null; + await applyExtraction(u.userId, u.itemId, toResult(await extract({ url: u.url, html, title: u.title }, { signal: AbortSignal.timeout(10_000) }))); + } catch (e) { + log("warn", "preview backfill skipped an item", { userId: u.userId, itemId: u.itemId, error: errorText(e) }); // it keeps its old look + } + } + return page.length; +} diff --git a/apps/web/lib/library/sharing.ts b/apps/web/lib/library/sharing.ts index c7c8aad..1fd9081 100644 --- a/apps/web/lib/library/sharing.ts +++ b/apps/web/lib/library/sharing.ts @@ -9,7 +9,7 @@ import { sendEmail } from "../email"; import { sharedListEmail } from "../emails"; import { ApiError } from "../errors"; import { log } from "../log"; -import { queueExtraction } from "./items"; +import { keepLater } from "./keeping"; import { ORIGIN } from "../origin"; type Defer = (task: () => Promise) => void; @@ -96,7 +96,7 @@ export async function copyList(userId: string, listId: string, { defer }: { defe } } const list = await createList(userId, { name: page.list.name, purpose: page.list.purpose, createdBy: "user", items: entries }); - await queueExtraction(userId, toRead, { defer }); + await keepLater(userId, toRead, { defer }); return list; } diff --git a/apps/web/lib/schedule.ts b/apps/web/lib/schedule.ts index 1168cef..1fae458 100644 --- a/apps/web/lib/schedule.ts +++ b/apps/web/lib/schedule.ts @@ -1,8 +1,7 @@ import { claimOnce, narrowAiGrants, runDueExpiry } from "@pluck/db"; import { FIRST_PARTY_REDIRECTS } from "./auth"; -import { sortBackfill } from "./capture"; import { announceNewPosts } from "./indexnow"; -import { pollInbound, purgeBinNow, queueDueDigests } from "./library"; +import { pollInbound, purgeBinNow, queueDueDigests, sortBackfill } from "./library"; import { log } from "./log"; // Time-based work, run from every cron tick and the worker loop. Each piece is claimed in the diff --git a/docs/architecture.md b/docs/architecture.md index 5af471d..fa44643 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -26,7 +26,7 @@ One Next.js app (`apps/web`) is the website, the signed-in app, the REST API for | `apps/web/lib/api.ts` | `route()` and `publicRoute()`: auth, scope, rate limits, body parsing and the error shape for every REST route. Most route files are a few lines. | | `apps/web/lib/library/*` | The library: the product rules over `packages/db` (URL normalisation, the hourly pluck budget, the Kind and text rules, unknown tags refused with the nearest ones, kicking extraction). REST routes and the connector both call it. | | `apps/web/lib/connector.ts` | Binds `PluckLibrary` (the interface in `packages/mcp/src/library.ts`) to the library for one person, with every write marked as made by AI. | -| `apps/web/lib/capture.ts`, `jobs.ts`, `schedule.ts`; `apps/web/worker.ts` | Keeping a page, the job handlers, and the time-based work (expiry, the Monday email, emptying the bin, polling email-in). `worker.ts` loops over them for self-host; on Vercel the cron route does. | +| `apps/web/lib/library/keeping.ts`, `jobs.ts`, `job-runner.ts`, `schedule.ts`; `apps/web/worker.ts` | Keeping a page (now, later, from a file or an import, with its one retry policy), the job handlers and the loop that runs them, and the time-based work (expiry, the Monday email, emptying the bin, polling email-in). `worker.ts` loops over them for self-host; on Vercel the cron route does. | | `apps/web/lib/auth.ts`, `signin-code.ts`, `signin-link.ts` | Better Auth: email magic link, the 6-digit code the iPhone app uses, and the OAuth provider for the extension, the iPhone app and AI tools. | | `apps/web/lib/email.ts`, `emails.ts`, `digest-email.ts` | `sendEmail` (SMTP to Mailpit locally, SES in production) and the email templates. | | `packages/db` | Every collection, index and query (`collections.ts` defines them all). Every query is scoped by `userId`; `public.ts` is the only reader for public Lists. | @@ -45,13 +45,13 @@ One Next.js app (`apps/web`) is the website, the signed-in app, the REST API for ## How a pluck travels -**From the web, the extension or the iPhone app.** `POST /api/v1/items` goes through `route()` (`lib/api.ts`), then `pluckAny` (`lib/library/items.ts`), then `pluckLink` in `packages/db`, which upserts on the person and the URL's hash. A Re-pluck finds the existing Item, moves it to the Inbox, takes it out of the bin and touches it; it never makes a second Item. The page is then read inline with a 5-second cap (`extractAttempt` in `lib/capture.ts`); covers and the kept HTML go to blob storage after the response. A block of many links makes the rows at once and queues an `extract` job for each. +**From the web, the extension or the iPhone app.** `POST /api/v1/items` goes through `route()` (`lib/api.ts`), then `pluckAny` (`lib/library/items.ts`), then `pluckLink` in `packages/db`, which upserts on the person and the URL's hash. A Re-pluck finds the existing Item, moves it to the Inbox, takes it out of the bin and touches it; it never makes a second Item. The page is then read inline with a 5-second cap (`keepNow` in `lib/library/keeping.ts`); covers and the kept HTML go to blob storage after the response. A block of many links makes the rows at once and queues an `extract` job for each (`keepLater`). A page gets four tries, counted on the Item: now, then 30 s, 2 min and 10 min after each failure; a 4xx is final at once. **From your AI.** `app/mcp/route.ts` checks the OAuth token or API key, that the tool is still connected and the rate limits, then serves `createPluckServer` (`packages/mcp/src/server.ts`), which lists only the tools the token's scopes allow. Tools call `connectorFor(userId)`, which reaches the same library, with `by: "mcp"` so Tags, notes, Lists and saves show "by AI". **From email.** SES writes the message to the bucket under `inbound/`; the schedule polls it and queues an `email-in` job, which becomes an Item through `ingestInbound` (`lib/library/email-in.ts`). -**Jobs.** The queue is the `jobs` collection (`packages/db/src/jobs.ts`: claim with a lease, retry, fail). The handlers are in `apps/web/lib/jobs.ts`. They run: +**Jobs.** The queue is the `jobs` collection (`packages/db/src/jobs.ts`: claim with a lease, retry, fail). The handlers are in `apps/web/lib/jobs.ts`; the loop that claims and retries them is `lib/job-runner.ts`. They run: - locally: `pnpm --filter @pluck/web worker`; - self-hosted: the `worker` service in `docker-compose.yml`, the same image as `web`; @@ -93,11 +93,6 @@ These are the places where the code is shallower than it needs to be: a rule wri - **Problem:** `tagNames`, `listNames`, `inList`, `searchText` and `searchBody` are "kept in step by the writes that change them", so each writer has to know which helpers to call. `inList` carries a product rule (Items in a List never expire): a writer that forgets `syncListNames` breaks expiry without a failing test. - **Shape:** one `afterItemsChanged(userId, itemIds)` that recomputes every derived field and count, called by writers instead of picking helpers. Test: an invariant test that runs each write and checks the stored fields equal a fresh recompute. -### 3. Keeping a page is spread over four files (M) - -- **Files:** `apps/web/lib/capture.ts` (`extractAttempt`, its own backoff of 30 s, 2 min, 10 min over four attempts), `lib/jobs.ts` (the `extract` handler, and the generic job retry of `30s * 2^n` over five attempts), `lib/library/items.ts` (`pluck`, `retryExtraction`, `queueExtraction`, and a dynamic `import("../jobs")` to dodge an import cycle), `library/files.ts` and `library/account.ts` (each builds an `ExtractionResult` its own way), and `pluckLink` in `packages/db` (decides whether a page needs reading). -- **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. ### 5. One session module for the extension and the iPhone app (M) diff --git a/packages/db/src/jobs.ts b/packages/db/src/jobs.ts index ba75edf..ec8ee66 100644 --- a/packages/db/src/jobs.ts +++ b/packages/db/src/jobs.ts @@ -13,10 +13,10 @@ export async function enqueue(queue: string, payload: Record, { return res.insertedId.toHexString(); } -export async function claimJob(leaseMs = 60_000): Promise { +export async function claimJob(leaseMs = 60_000, queue?: string): Promise { const now = new Date(); const doc: JobDoc | null = await (await jobs()).findOneAndUpdate( - { state: { $in: ["queued", "running"] }, visibleAt: { $lte: now } }, + { state: { $in: ["queued", "running"] }, visibleAt: { $lte: now }, ...(queue && { queue }) }, { $set: { state: "running", visibleAt: new Date(now.getTime() + leaseMs) }, $inc: { attempts: 1 } }, { sort: { visibleAt: 1 }, returnDocument: "after" }, );