apps/api/src/queue/store.ts

// 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> = T | Promise<T>;

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<void>;
  createJob(optionsSnapshot: object): Awaitable<JobRow>;
  getJob(id: string): Awaitable<JobRow | null>;
  setJobStatus(id: string, status: JobStatus): Awaitable<void>;
  updateJobOptions(id: string, optionsSnapshot: object): Awaitable<void>;
  listResumableJobs(): Awaitable<JobRow[]>;
  recoverInterruptedJob(id: string): Awaitable<void>;
  createItem(input: {
    jobId: string;
    ord: number;
    filename: string;
    uploadPath: string;
    detectedMode: DetectedMode;
  }): Awaitable<ItemRow>;
  getItem(id: string): Awaitable<ItemRow | null>;
  listItemsByJob(jobId: string): Awaitable<ItemRow[]>;
  setItemStatus(
    id: string,
    status: ItemStatus,
    errorPreview?: string,
    meta?: SetItemStatusMeta,
  ): Awaitable<void>;
  setItemPaths(
    id: string,
    paths: Partial<{
      evidencePath: string;
      taskPackagePath: string;
      judgeResultPath: string;
      exportZipPath: string;
    }>,
  ): Awaitable<void>;
  takeNextQueuedItem(jobId: string): Awaitable<ItemRow | null>;
  resetItemForRetry(id: string): Awaitable<void>;
}

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