From 9336bb7a5e298b67d2b9a91260873d643b467987 Mon Sep 17 00:00:00 2001 From: Thom Lamb Date: Mon, 31 Aug 2026 19:47:34 -0500 Subject: [PATCH] feat(intake): draft queued submissions without ever losing one (#223) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The governing rule is that a submission is the only irreplaceable thing in this pipeline. The photos are often the only copy of an item no longer in the sender's hands, so a missing key, an unreadable file, a failed call and three exhausted retries all end the same way: the item keeps its photos, stays pending, and waits. Nothing in this file deletes anything. An absent key returns early and spends no attempt. Counting it as a failure would mean a fortnight without a key exhausted the retries and marked every waiting submission failed, with nothing wrong with any of them. A failure leaves the row queued while tries remain, so the sweeper picks it up again, and failed once they are spent, so a dead submission stops costing money and waits for a person instead of retrying forever. Photos are read once and passed down rather than loaded again inside the drafting call — the first read already has to happen to check there is at least one. Co-Authored-By: Claude Opus 5 --- backend/src/intake/draftingWorker.ts | 156 ++++++++++++++++++ .../integration/drafting.integration.test.ts | 93 +++++++++++ 2 files changed, 249 insertions(+) create mode 100644 backend/src/intake/draftingWorker.ts diff --git a/backend/src/intake/draftingWorker.ts b/backend/src/intake/draftingWorker.ts new file mode 100644 index 0000000..3c2effe --- /dev/null +++ b/backend/src/intake/draftingWorker.ts @@ -0,0 +1,156 @@ +import type Anthropic from '@anthropic-ai/sdk'; +import { promises as fs } from 'fs'; +import path from 'path'; +import { pool } from '../db'; +import { typeForExtension } from '../uploadTypes'; +import { getAnthropicClient } from './anthropicClient'; +import { draftListing } from './draftListing'; +import { applyDraft } from './applyDraft'; + +/** + * Turns queued submissions into drafts. + * + * Driven from two places: a call at the end of a successful submission, so a + * draft is usually waiting by the time anybody looks, and a cron sweeper, so a + * restart mid-draft is recoverable rather than a permanently stalled row. + * + * The governing rule is that a submission is the only irreplaceable thing here. + * The photos are often the only copy of an item no longer in the sender's + * hands, so every failure below leaves the row and its images intact and merely + * undrafted. Nothing in this file deletes anything. + */ + +/** Three tries, then it waits for a person rather than burning money on a loop. */ +export const MAX_ATTEMPTS = 3; + +/** Small, because each one is an API call and the sweeper comes round again. */ +const DEFAULT_BATCH = 5; + +type Photo = { mediaType: string; base64: string }; + +interface QueuedRow { + item_id: number; + submitter_note: string | null; +} + +export interface SweepResult { + drafted: number; + failed: number; + skipped: number; +} + +async function readPhotos(itemId: number): Promise { + const { rows } = await pool.query<{ image_path: string }>( + `SELECT image_path FROM item_images WHERE item_id = $1 ORDER BY sort_order`, + [itemId] + ); + + const photos: Photo[] = []; + for (const row of rows) { + // basename only: image_path is stored as '/uploads/' and the + // directory it lives in is a server constant. Same rule as handleRow in + // backfillImageReencode, but using the shared typeForExtension rather than + // that file's private copy of the map. + const file = path.join(process.env.UPLOADS_DIR ?? '', path.basename(row.image_path)); + const mediaType = typeForExtension(path.extname(file)); + // Null for anything the app would refuse to serve. Sending it to the model + // would be paying to have it rejected. + if (mediaType === null) continue; + photos.push({ mediaType, base64: (await fs.readFile(file)).toString('base64') }); + } + return photos; +} + +async function namesOf(table: 'categories' | 'tags'): Promise { + // The table name is a closed union, never caller input — there is nothing + // here to interpolate from a request. + const { rows } = await pool.query<{ name: string }>(`SELECT name FROM ${table} ORDER BY name`); + return rows.map((row) => row.name); +} + +/** + * Records a failure without ever losing the submission. + * + * The row stays reachable either way: 'queued' while tries remain, so the + * sweeper picks it up again, and 'failed' once they are spent, so it stops + * costing money and waits for a person. The item and its photos are untouched + * in both cases. + */ +async function recordFailure(itemId: number, message: string): Promise { + await pool.query( + `UPDATE item_drafts + SET attempts = attempts + 1, + ai_error = $2, + state = CASE WHEN attempts + 1 >= $3 THEN 'failed' ELSE 'queued' END + WHERE item_id = $1`, + [itemId, message.slice(0, 500), MAX_ATTEMPTS] + ); +} + +/** Photos are passed in rather than re-read: the caller has already loaded them + * to check there is at least one, and reading every file off disk twice per + * submission is a cost for nothing. */ +async function draftOne( + client: Anthropic, + itemId: number, + note: string | null, + photos: Photo[] +): Promise { + const outcome = await draftListing(client, { + photos, + note, + categories: await namesOf('categories'), + tags: await namesOf('tags') + }); + + const db = await pool.connect(); + try { + await db.query('BEGIN'); + await applyDraft(db, itemId, outcome); + await db.query('COMMIT'); + } catch (err) { + await db.query('ROLLBACK'); + throw err; + } finally { + db.release(); + } +} + +export async function draftQueued(limit = DEFAULT_BATCH): Promise { + const { rows } = await pool.query( + `SELECT item_id, submitter_note FROM item_drafts + WHERE state = 'queued' AND attempts < $2 + ORDER BY created_at + LIMIT $1`, + [limit, MAX_ATTEMPTS] + ); + + // Unconfigured is not a failure and must not spend an attempt. A fortnight + // without a key would otherwise exhaust the retries and mark every waiting + // submission failed, with nothing wrong with any of them. + const client = getAnthropicClient(); + if (client === null) { + return { drafted: 0, failed: 0, skipped: rows.length }; + } + + let drafted = 0; + let failed = 0; + + for (const row of rows) { + try { + const photos = await readPhotos(row.item_id); + if (photos.length === 0) { + throw new Error('no readable photos'); + } + await draftOne(client, row.item_id, row.submitter_note, photos); + drafted++; + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + console.error(`[drafting] item ${row.item_id}: ${message}`); + await recordFailure(row.item_id, message); + failed++; + } + } + + return { drafted, failed, skipped: 0 }; +} diff --git a/backend/tests/integration/drafting.integration.test.ts b/backend/tests/integration/drafting.integration.test.ts index fe0ae03..39285f4 100644 --- a/backend/tests/integration/drafting.integration.test.ts +++ b/backend/tests/integration/drafting.integration.test.ts @@ -2,6 +2,8 @@ import { pool } from '../../src/db'; import { applyDraft } from '../../src/intake/applyDraft'; import { DraftOutcome } from '../../src/intake/draftListing'; import { resetDb, closeDb } from './setup/testDb'; +import { draftQueued, MAX_ATTEMPTS } from '../../src/intake/draftingWorker'; +import { resetAnthropicClient } from '../../src/intake/anthropicClient'; beforeEach(async () => { await resetDb(); @@ -149,3 +151,94 @@ describe('applying a draft', () => { expect(rows[0]?.ai_error).toBeNull(); }); }); + +describe('the drafting worker', () => { + beforeEach(() => { + delete process.env.ANTHROPIC_API_KEY; + resetAnthropicClient(); + }); + + afterAll(() => { + resetAnthropicClient(); + }); + + // The case that must never lose a submission. An unconfigured environment is + // a working one: the item keeps its photos and waits. + it('leaves submissions queued when there is no key', async () => { + const itemId = await seedSubmission(); + + const result = await draftQueued(); + + expect(result).toEqual({ drafted: 0, failed: 0, skipped: 1 }); + + const { rows } = await pool.query( + `SELECT state, attempts FROM item_drafts WHERE item_id = $1`, + [itemId] + ); + expect(rows[0]?.state).toBe('queued'); + // Skipping is not an attempt. Otherwise a fortnight without a key would + // exhaust the retries and mark everything failed with nothing wrong with it. + expect(rows[0]?.attempts).toBe(0); + }); + + it('does not touch a row that has already spent its attempts', async () => { + const itemId = await seedSubmission(); + await pool.query(`UPDATE item_drafts SET attempts = $2 WHERE item_id = $1`, [ + itemId, + MAX_ATTEMPTS + ]); + process.env.ANTHROPIC_API_KEY = 'sk-ant-not-used'; + resetAnthropicClient(); + + const result = await draftQueued(); + + expect(result.drafted).toBe(0); + expect(result.failed).toBe(0); + }); + + // A submission whose files cannot be read fails without an API call, and + // without the item or its row going anywhere. + it('fails a submission with no readable photos, keeping the item', async () => { + const itemId = await seedSubmission(); + process.env.ANTHROPIC_API_KEY = 'sk-ant-not-used'; + resetAnthropicClient(); + + const result = await draftQueued(); + + expect(result.failed).toBe(1); + + const draft = await pool.query( + `SELECT state, attempts, ai_error FROM item_drafts WHERE item_id = $1`, + [itemId] + ); + expect(draft.rows[0]?.attempts).toBe(1); + // One try spent, two left, so it stays reachable for the sweeper. + expect(draft.rows[0]?.state).toBe('queued'); + expect(draft.rows[0]?.ai_error).toContain('no readable photos'); + + const item = await pool.query(`SELECT status FROM items WHERE id = $1`, [itemId]); + expect(item.rows[0]?.status).toBe('pending'); + }); + + // Three tries, then it stops costing money and waits for a person. The item + // and its photos survive that too. + it('gives up after MAX_ATTEMPTS rather than retrying forever', async () => { + const itemId = await seedSubmission(); + process.env.ANTHROPIC_API_KEY = 'sk-ant-not-used'; + resetAnthropicClient(); + + for (let i = 0; i < MAX_ATTEMPTS; i++) { + await draftQueued(); + } + + const draft = await pool.query(`SELECT state, attempts FROM item_drafts WHERE item_id = $1`, [itemId]); + expect(draft.rows[0]?.attempts).toBe(MAX_ATTEMPTS); + expect(draft.rows[0]?.state).toBe('failed'); + + const item = await pool.query(`SELECT id, status FROM items WHERE id = $1`, [itemId]); + expect(item.rows[0]?.status).toBe('pending'); + + // And it is not picked up again, so a dead submission stops spending money. + expect(await draftQueued()).toEqual({ drafted: 0, failed: 0, skipped: 0 }); + }); +});