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'; import { notifyDraftReady } from './notifyDraft'; import { removeBackgroundsForItem } from './backgroundRemoval'; /** * 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. * * Background removal (#281) follows drafting rather than running on its own * pass. That couples the two: an environment with no ANTHROPIC_API_KEY drafts * nothing, so it cuts out nothing either. That is the intended trade — a * separate pass would re-attempt an unreachable sidecar on every sweep for a * row that is going to sit at 'queued' indefinitely — and the admin's per-photo * control in the review queue is the way to do it by hand meanwhile. */ /** 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; remove_background: boolean; } 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, remove_background 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++; // Deliberately after the draft is committed, and catching for itself. // // This is the sender's tick from the submission page, honoured here so // they never waited for it — and a failure must not mark a draft that was // written correctly as failed. The photo keeps its original in that case, // and the admin's per-photo control is still there to do it by hand. // // Awaited, unlike the notification below, so a sweep that has returned // has finished its work. Nothing is waiting on this: the worker is off // the request path, which is the whole reason drafting lives here. if (row.remove_background) { await removeBackgroundsForItem(row.item_id).catch((err) => console.error(`[drafting] background removal for item ${row.item_id}:`, err) ); } // Fire and forget, and deliberately after the draft is committed. A mail // failure must never mark a draft that was written correctly as failed — // the queue is what the admin actually works from, and the email is a // convenience on top of it. void notifyDraftReady(row.item_id).catch((err) => console.error(`[drafting] notifying for item ${row.item_id}:`, err) ); } 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 }; }