rag-service/tests/catalog/repository-ocr.test.ts

import assert from "node:assert/strict";
import test from "node:test";
import { CatalogError, CatalogRepository } from "../../src/modules/catalog/repository.js";

const versionRow = {
  version_id: "version-1",
  source_id: "source-1",
  version_number: 2,
  previous_version_id: "version-0",
  state: "indexing",
  tags: [],
  source_content_hash: null,
  processing_fingerprint: "fingerprint",
  metadata_hash: "metadata",
  embedding_provider: "test",
  embedding_model: "test",
  embedding_dimensions: 3,
  expected_document_count: 1,
  expected_point_count: 0,
  verified_point_count: 0,
  qdrant_collection: "rag"
};

const jobRow = {
  ocr_job_id: "job-1",
  ocr_document_id: "document-1",
  ocr_remote_job_id: "remote-1",
  ocr_remote_idempotency_key: "document-hash:ocr-v1:pages-hash",
  ocr_state: "running",
  ocr_requested_pages: [1, 3],
  ocr_completed_pages: 1,
  ocr_config_version: "ocr-v1",
  ocr_attempt_count: 2,
  ocr_heartbeat_at: new Date("2026-09-14T10:00:00Z"),
  ocr_lease_expires_at: new Date("2026-09-14T10:01:00Z"),
  ocr_next_attempt_at: null,
  ocr_error_code: null,
  ocr_error_detail: null
};

function poolFor(handler: (sql: string, params?: unknown[]) => Promise<{ rowCount: number; rows: unknown[] }>) {
  const client = {
    query: async (sql: string, params?: unknown[]) => /^(BEGIN|COMMIT|ROLLBACK|SET CONSTRAINTS)/.test(sql.trim())
      ? { rowCount: 0, rows: [] }
      : handler(sql, params),
    release() {}
  };
  return { query: handler, connect: async () => client };
}

test("pending OCR lookup returns the persisted version and jobs for a duplicate ingestion", async () => {
  const pool = poolFor(async (sql, params) => {
    assert.match(sql, /JOIN rag_version_documents/);
    assert.match(sql, /JOIN rag_ocr_jobs/);
    assert.match(sql, /source_content_hash IS NULL/);
    assert.match(sql, /state IN \('pending', 'indexing', 'review_required'\)/);
    assert.deepEqual(params, ["source-1", "manifest", "fingerprint", "metadata"]);
    return { rowCount: 1, rows: [{ ...versionRow, ...jobRow }] };
  });
  const repository = new CatalogRepository(pool as never);

  const candidate = await repository.findPendingOcrVersion({
    sourceId: "source-1",
    originalManifestHash: "manifest",
    processingFingerprint: "fingerprint",
    metadataHash: "metadata"
  });

  assert.equal(candidate?.version.versionId, "version-1");
  assert.deepEqual(candidate?.jobs.map((job) => [job.jobId, job.remoteJobId, job.remoteIdempotencyKey]), [
    ["job-1", "remote-1", "document-hash:ocr-v1:pages-hash"]
  ]);
});

test("pending OCR lookup returns undefined when only terminal candidates exist", async () => {
  const repository = new CatalogRepository(poolFor(async () => ({ rowCount: 0, rows: [] })) as never);
  const candidate = await repository.findPendingOcrVersion({
    sourceId: "source-1",
    originalManifestHash: "manifest",
    processingFingerprint: "fingerprint",
    metadataHash: "metadata"
  });
  assert.equal(candidate, undefined);
});

test("lease claiming uses a locked queue row and keeps its remote identity", async () => {
  const calls: string[] = [];
  const pool = poolFor(async (sql, params) => {
    calls.push(sql);
    if (sql.includes("FOR UPDATE SKIP LOCKED")) {
      assert.match(sql, /state = 'queued'/);
      return { rowCount: 1, rows: [{ job_id: "job-1" }] };
    }
    assert.match(sql, /attempt_count = attempt_count \+ 1/);
    assert.deepEqual(params, ["job-1", "30000"]);
    return { rowCount: 1, rows: [jobRow] };
  });
  const repository = new CatalogRepository(pool as never);

  const claimed = await repository.claimNextOcrJob(30_000);

  assert.equal(calls.length, 2);
  assert.equal(claimed?.remoteJobId, "remote-1");
  assert.equal(claimed?.remoteIdempotencyKey, "document-hash:ocr-v1:pages-hash");
});

test("lease claiming returns undefined when no queued job is eligible", async () => {
  const repository = new CatalogRepository(poolFor(async () => ({ rowCount: 0, rows: [] })) as never);
  assert.equal(await repository.claimNextOcrJob(30_000), undefined);
});

test("expired leases return to queued without clearing remote recovery identity", async () => {
  const repository = new CatalogRepository(poolFor(async (sql) => {
    assert.match(sql, /state = 'running'/);
    assert.match(sql, /lease_expires_at <= now\(\)/);
    assert.doesNotMatch(sql, /remote_(job_id|idempotency_key)\s*=/);
    return { rowCount: 1, rows: [{ ...jobRow, ocr_state: "queued" }] };
  }) as never);

  const recovered = await repository.recoverExpiredOcrLeases();

  assert.deepEqual(recovered.map((job) => [job.state, job.remoteJobId, job.remoteIdempotencyKey]), [
    ["queued", "remote-1", "document-hash:ocr-v1:pages-hash"]
  ]);
});

test("candidate lifecycle persistence requires exact native, OCR, and metrics evidence", async () => {
  const calls: Array<{ sql: string; params?: unknown[] }> = [];
  const repository = new CatalogRepository(poolFor(async (sql, params) => {
    calls.push({ sql, params });
    if (sql.includes("SELECT version_id, document_id")) return { rowCount: 1, rows: [{ version_id: "version-1", document_id: "document-1" }] };
    if (sql.includes("SELECT DISTINCT")) return { rowCount: 1, rows: [{ version_id: "version-1" }] };
    if (sql.includes("SELECT 1 FROM rag_ocr_jobs")) return { rowCount: 0, rows: [] };
    if (sql.includes("FROM rag_ocr_jobs")) return { rowCount: 1, rows: [{ ...jobRow, ocr_state: "succeeded" }] };
    return { rowCount: 1, rows: [] };
  }) as never);
  const jobs = await repository.listOcrJobs("version-1");
  assert.equal(await repository.completeOcrJob("job-1", { pages: [{ page: 1, text: "OCR", metrics: { inkCoverage: 0.4 } }] } as never), true);
  assert.doesNotMatch(calls[2]!.sql, /candidate_text_hash/);
  assert.deepEqual(await repository.listOcrVersionsAwaitingCandidate(), ["version-1"]);
  await repository.persistOcrCandidate("version-1", [{
    documentId: "document-1", page: 1, method: "ocr", nativeTextSha256: "a".repeat(64),
    ocrTextSha256: "b".repeat(64), candidateTextSha256: "b".repeat(64), metrics: { inkCoverage: 0.4 }, risks: ["CBGO4a"]
  }]);

  assert.deepEqual(jobs.map(({ jobId, state }) => [jobId, state]), [["job-1", "succeeded"]]);
  assert.match(calls[6]!.sql, /native_text_hash = \$8.*ocr_text_hash IS NOT DISTINCT FROM \$9.*metrics = \$10::jsonb/s);
  assert.deepEqual(calls[6]!.params?.slice(-3), ["a".repeat(64), "b".repeat(64), JSON.stringify({ inkCoverage: 0.4 })]);
});

test("review transitions enforce indexing, approval, and rejection state guards", async () => {
  const successfulSql: string[] = [];
  const success = new CatalogRepository(poolFor(async (sql) => {
    successfulSql.push(sql);
    return { rowCount: 1, rows: [] };
  }) as never);
  await success.markReviewRequired("version-1");
  await success.markOcrReviewIndexing("version-1", "reviewer-1");
  await success.rejectOcrVersion("version-1", "reviewer-1", "Unreadable code");
  assert.match(successfulSql[0], /state = 'indexing'.*NOT EXISTS/s);
  assert.match(successfulSql[0], /FROM rag_version_documents/);
  assert.match(successfulSql[0], /rag_ocr_jobs WHERE version_id = \$1 AND state <> 'succeeded'/);
  assert.match(successfulSql[1], /state = 'review_required'/);
  assert.match(successfulSql[2], /state = 'review_required'/);

  const blocked = new CatalogRepository(poolFor(async () => ({ rowCount: 0, rows: [] })) as never);
  await assert.rejects(
    blocked.markReviewRequired("version-2"),
    (error) => error instanceof CatalogError && error.code === "INVALID_VERSION_STATE"
  );
});

test("review context reconstructs exact lifecycle identity and rejects incomplete page evidence", async () => {
  const rows = [{ document_id: "document-1", page_number: 1, native_text_hash: "a".repeat(64), ocr_text_hash: "b".repeat(64),
    candidate_text_hash: "c".repeat(64), metrics: { inkCoverage: 0.4 }, risk_tokens: ["CBGO4a"] }];
  const repository = new CatalogRepository(poolFor(async (sql) => sql.includes("JOIN rag_sources")
    ? { rowCount: 1, rows: [{ version_id: "version-1", source_id: "source-1", state: "review_required", base_active_version_id: null,
      current_active_version_id: null, activate_requested: false, processing_fingerprint: "fingerprint", metadata_hash: "metadata" }] }
    : { rowCount: 1, rows }) as never);

  const context = await repository.loadOcrReviewContext("version-1");
  assert.deepEqual(context?.pages[0], { documentId: "document-1", page: 1, nativeTextSha256: "a".repeat(64),
    ocrTextSha256: "b".repeat(64), candidateTextSha256: "c".repeat(64), metrics: { inkCoverage: 0.4 }, risks: ["CBGO4a"] });
  rows[0]!.candidate_text_hash = null as never;
  await assert.rejects(repository.loadOcrReviewContext("version-1"), (error) => error instanceof CatalogError && error.code === "OCR_ARTIFACT_INTEGRITY_FAILED");
});