feat(intake): draft queued submissions without ever losing one (#223)

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 <noreply@anthropic.com>

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<Photo[]> {
+  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/<name>' 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<string[]> {
+  // 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<void> {
+  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<void> {
+  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<SweepResult> {
+  const { rows } = await pool.query<QueuedRow>(
+    `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 };
+}

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 });
+  });
+});