// apps/api/src/queue/store.ts // Storage contract plus the SQLite implementation used for local/single-node // development. Multi-instance deployments use PostgresStore from // ./postgres-store.js via store-factory.ts. import Database from "better-sqlite3"; import { mkdirSync, existsSync } from "node:fs"; import { dirname } from "node:path"; import { randomUUID } from "node:crypto"; export type Awaitable = T | Promise; export type JobStatus = "pending" | "running" | "completed" | "cancelled"; export type ItemStatus = | "queued" | "running" | "judging" | "completed" | "failed" | "stopped"; export type DetectedMode = "api" | "mcp"; export interface JobRow { id: string; status: JobStatus; options_snapshot: string; // JSON, never contains apiKey created_at: number; updated_at: number; } export interface ItemRow { id: string; job_id: string; ord: number; filename: string; upload_path: string; detected_mode: DetectedMode; status: ItemStatus; evidence_path: string | null; task_package_path: string | null; judge_result_path: string | null; export_zip_path: string | null; error_preview: string | null; error_details_path: string | null; attempt_count: number; last_stage: string | null; created_at: number; updated_at: number; } export interface SetItemStatusMeta { lastStage?: string | null; errorDetailsPath?: string | null; } export interface Store { close(): Awaitable; createJob(optionsSnapshot: object): Awaitable; getJob(id: string): Awaitable; setJobStatus(id: string, status: JobStatus): Awaitable; updateJobOptions(id: string, optionsSnapshot: object): Awaitable; listResumableJobs(): Awaitable; recoverInterruptedJob(id: string): Awaitable; createItem(input: { jobId: string; ord: number; filename: string; uploadPath: string; detectedMode: DetectedMode; }): Awaitable; getItem(id: string): Awaitable; listItemsByJob(jobId: string): Awaitable; setItemStatus( id: string, status: ItemStatus, errorPreview?: string, meta?: SetItemStatusMeta, ): Awaitable; setItemPaths( id: string, paths: Partial<{ evidencePath: string; taskPackagePath: string; judgeResultPath: string; exportZipPath: string; }>, ): Awaitable; takeNextQueuedItem(jobId: string): Awaitable; resetItemForRetry(id: string): Awaitable; } const SCHEMA_SQL = ` PRAGMA journal_mode = WAL; PRAGMA foreign_keys = ON; PRAGMA busy_timeout = 5000; CREATE TABLE IF NOT EXISTS jobs ( id TEXT PRIMARY KEY, status TEXT NOT NULL, options_snapshot TEXT NOT NULL, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS items ( id TEXT PRIMARY KEY, job_id TEXT NOT NULL REFERENCES jobs(id) ON DELETE CASCADE, ord INTEGER NOT NULL, filename TEXT NOT NULL, upload_path TEXT NOT NULL, detected_mode TEXT NOT NULL, status TEXT NOT NULL, evidence_path TEXT, task_package_path TEXT, judge_result_path TEXT, export_zip_path TEXT, error_preview TEXT, error_details_path TEXT, attempt_count INTEGER NOT NULL DEFAULT 0, last_stage TEXT, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS idx_items_job_ord ON items(job_id, ord); CREATE INDEX IF NOT EXISTS idx_items_status ON items(status); CREATE INDEX IF NOT EXISTS idx_items_job_status_ord ON items(job_id, status, ord); `; const ERROR_PREVIEW_MAX = 2 * 1024; export function nowMs() { return Date.now(); } export function truncateError(s: string | null | undefined): string | null { if (!s) return null; return s.length > ERROR_PREVIEW_MAX ? s.slice(0, ERROR_PREVIEW_MAX) + "...(truncated)" : s; } function hasColumn(db: Database.Database, table: string, column: string): boolean { const rows = db.prepare(`PRAGMA table_info(${table})`).all() as Array<{ name?: string }>; return rows.some((row) => row.name === column); } function ensureSqliteColumns(db: Database.Database): void { const additions: Array<[string, string]> = [ ["error_details_path", "ALTER TABLE items ADD COLUMN error_details_path TEXT"], ["attempt_count", "ALTER TABLE items ADD COLUMN attempt_count INTEGER NOT NULL DEFAULT 0"], ["last_stage", "ALTER TABLE items ADD COLUMN last_stage TEXT"], ]; additions.forEach(([column, sql]) => { if (!hasColumn(db, "items", column)) db.exec(sql); }); } function normalizeItemRow(row: ItemRow): ItemRow { return { ...row, attempt_count: Number(row.attempt_count || 0), error_details_path: row.error_details_path ?? null, last_stage: row.last_stage ?? null, }; } export class SqliteStore implements Store { private db: Database.Database; constructor(sqlitePath: string) { if (!existsSync(dirname(sqlitePath))) { mkdirSync(dirname(sqlitePath), { recursive: true }); } this.db = new Database(sqlitePath); this.db.exec(SCHEMA_SQL); ensureSqliteColumns(this.db); } close(): void { this.db.close(); } // jobs createJob(optionsSnapshot: object): JobRow { const id = randomUUID(); const t = nowMs(); const job: JobRow = { id, status: "pending", options_snapshot: JSON.stringify(optionsSnapshot), created_at: t, updated_at: t, }; this.db .prepare( "INSERT INTO jobs (id, status, options_snapshot, created_at, updated_at) VALUES (?, ?, ?, ?, ?)", ) .run(job.id, job.status, job.options_snapshot, job.created_at, job.updated_at); return job; } getJob(id: string): JobRow | null { const row = this.db.prepare("SELECT * FROM jobs WHERE id = ?").get(id) as JobRow | undefined; return row ?? null; } setJobStatus(id: string, status: JobStatus): void { this.db .prepare("UPDATE jobs SET status = ?, updated_at = ? WHERE id = ?") .run(status, nowMs(), id); } updateJobOptions(id: string, optionsSnapshot: object): void { this.db .prepare("UPDATE jobs SET options_snapshot = ?, updated_at = ? WHERE id = ?") .run(JSON.stringify(optionsSnapshot), nowMs(), id); } listResumableJobs(): JobRow[] { return this.db .prepare("SELECT * FROM jobs WHERE status = 'running' ORDER BY created_at") .all() as JobRow[]; } recoverInterruptedJob(id: string): void { const t = nowMs(); const tx = this.db.transaction((jobId: string) => { this.db .prepare( `UPDATE items SET status = 'queued', error_preview = NULL, last_stage = 'recovered', updated_at = ? WHERE job_id = ? AND status IN ('running', 'judging')`, ) .run(t, jobId); this.db .prepare("UPDATE jobs SET status = 'running', updated_at = ? WHERE id = ?") .run(t, jobId); }); tx(id); } // items createItem(input: { jobId: string; ord: number; filename: string; uploadPath: string; detectedMode: DetectedMode; }): ItemRow { const id = randomUUID(); const t = nowMs(); const row: ItemRow = { id, job_id: input.jobId, ord: input.ord, filename: input.filename, upload_path: input.uploadPath, detected_mode: input.detectedMode, status: "queued", evidence_path: null, task_package_path: null, judge_result_path: null, export_zip_path: null, error_preview: null, error_details_path: null, attempt_count: 0, last_stage: null, created_at: t, updated_at: t, }; this.db .prepare( `INSERT INTO items (id, job_id, ord, filename, upload_path, detected_mode, status, evidence_path, task_package_path, judge_result_path, export_zip_path, error_preview, error_details_path, attempt_count, last_stage, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, ) .run( row.id, row.job_id, row.ord, row.filename, row.upload_path, row.detected_mode, row.status, row.evidence_path, row.task_package_path, row.judge_result_path, row.export_zip_path, row.error_preview, row.error_details_path, row.attempt_count, row.last_stage, row.created_at, row.updated_at, ); return row; } getItem(id: string): ItemRow | null { const row = this.db.prepare("SELECT * FROM items WHERE id = ?").get(id) as ItemRow | undefined; return row ? normalizeItemRow(row) : null; } listItemsByJob(jobId: string): ItemRow[] { return ( this.db .prepare("SELECT * FROM items WHERE job_id = ? ORDER BY ord ASC") .all(jobId) as ItemRow[] ).map(normalizeItemRow); } setItemStatus( id: string, status: ItemStatus, errorPreview?: string, meta: SetItemStatusMeta = {}, ): void { this.db .prepare( `UPDATE items SET status = ?, error_preview = ?, last_stage = COALESCE(?, last_stage), error_details_path = COALESCE(?, error_details_path), updated_at = ? WHERE id = ?`, ) .run( status, truncateError(errorPreview), meta.lastStage ?? null, meta.errorDetailsPath ?? null, nowMs(), id, ); } setItemPaths( id: string, paths: Partial<{ evidencePath: string; taskPackagePath: string; judgeResultPath: string; exportZipPath: string; }>, ): void { const fields: string[] = []; const values: unknown[] = []; if (paths.evidencePath !== undefined) { fields.push("evidence_path = ?"); values.push(paths.evidencePath); } if (paths.taskPackagePath !== undefined) { fields.push("task_package_path = ?"); values.push(paths.taskPackagePath); } if (paths.judgeResultPath !== undefined) { fields.push("judge_result_path = ?"); values.push(paths.judgeResultPath); } if (paths.exportZipPath !== undefined) { fields.push("export_zip_path = ?"); values.push(paths.exportZipPath); } if (!fields.length) return; fields.push("updated_at = ?"); values.push(nowMs()); values.push(id); this.db.prepare(`UPDATE items SET ${fields.join(", ")} WHERE id = ?`).run(...values); } takeNextQueuedItem(jobId: string): ItemRow | null { const tx = this.db.transaction((id: string) => { const row = this.db .prepare( "SELECT * FROM items WHERE job_id = ? AND status = 'queued' ORDER BY ord ASC LIMIT 1", ) .get(id) as ItemRow | undefined; if (!row) return null; const t = nowMs(); this.db .prepare( `UPDATE items SET status = 'running', attempt_count = attempt_count + 1, error_preview = NULL, last_stage = 'running', updated_at = ? WHERE id = ?`, ) .run(t, row.id); return normalizeItemRow({ ...row, status: "running" as const, attempt_count: Number(row.attempt_count || 0) + 1, last_stage: "running", updated_at: t, }); }); return tx(jobId); } resetItemForRetry(id: string): void { this.db .prepare( `UPDATE items SET status = 'queued', error_preview = NULL, error_details_path = NULL, attempt_count = 0, last_stage = NULL, evidence_path = NULL, task_package_path = NULL, judge_result_path = NULL, export_zip_path = NULL, updated_at = ? WHERE id = ?`, ) .run(nowMs(), id); } }