feat(dn): add staged ingestion schema

scripts/pb-add-dn-staged-ingestion.py

@@ -0,0 +1,220 @@
+#!/usr/bin/env python3
+"""Add the staged delivery-note ingestion schema.
+
+The migration is idempotent and is a dry run unless ``--apply`` is supplied.
+It reads existing PocketBase fields and indexes before appending anything, so
+running it against a live instance preserves unrelated schema changes.
+"""
+import sys
+
+from pb_env import PB, http, superuser_token
+
+
+DRY_RUN = "--dry-run" in sys.argv or "--apply" not in sys.argv
+
+
+def get_collection(token, name):
+    code, response = http("GET", f"/api/collections/{name}", token=token)
+    return response if code == 200 else None
+
+
+def require_collection(token, name):
+    collection = get_collection(token, name)
+    if collection is None:
+        sys.exit(f"collection {name} not found")
+    return collection
+
+
+def relation(name, collection_id, required=False):
+    return {
+        "name": name,
+        "type": "relation",
+        "required": required,
+        "collectionId": collection_id,
+        "maxSelect": 1,
+        "cascadeDelete": False,
+    }
+
+
+def field(name, field_type="text", **options):
+    return {"name": name, "type": field_type, "required": False, **options}
+
+
+def add_fields(token, name, wanted):
+    collection = require_collection(token, name)
+    existing = list(collection.get("fields", []))
+    names = {item.get("name") for item in existing}
+    additions = [item for item in wanted if item["name"] not in names]
+    if not additions:
+        print(f"  {name}: fields already present")
+        return
+    print(f"  {name}: add fields {[item['name'] for item in additions]}")
+    if not DRY_RUN:
+        code, response = http(
+            "PATCH", f"/api/collections/{collection['id']}",
+            {"fields": existing + additions}, token,
+        )
+        if code != 200:
+            sys.exit(f"PATCH {name} failed: {code} {response}")
+
+
+def add_indexes(token, name, wanted):
+    collection = require_collection(token, name)
+    existing = list(collection.get("indexes", []) or [])
+    additions = [item for item in wanted if item not in existing]
+    if not additions:
+        print(f"  {name}: indexes already present")
+        return
+    print(f"  {name}: add indexes {additions}")
+    if not DRY_RUN:
+        code, response = http(
+            "PATCH", f"/api/collections/{collection['id']}",
+            {"indexes": existing + additions}, token,
+        )
+        if code != 200:
+            sys.exit(f"PATCH {name} indexes failed: {code} {response}")
+
+
+def create_collection(token, name, fields, indexes):
+    existing = get_collection(token, name)
+    if existing is not None:
+        print(f"  {name}: already exists")
+        add_fields(token, name, fields)
+        add_indexes(token, name, indexes)
+        return existing["id"]
+    print(f"  {name}: create collection")
+    if DRY_RUN:
+        return name
+    code, response = http(
+        "POST", "/api/collections",
+        {"name": name, "type": "base", "fields": fields, "indexes": indexes},
+        token,
+    )
+    if code != 200:
+        sys.exit(f"CREATE {name} failed: {code} {response}")
+    return response["id"]
+
+
+def system_fields():
+    return [
+        field("created", "date", onCreate=True),
+        field("updated", "date", onCreate=True, onUpdate=True),
+    ]
+
+
+def main():
+    try:
+        token = superuser_token()
+    except SystemExit as error:
+        # A dry run is useful in CI and during review even when no PB
+        # credentials are mounted.  It still reports the complete intended
+        # mutation and, by definition, performs no writes.
+        if not DRY_RUN:
+            raise
+        print(f"offline dry run: {error}")
+        print("would create: dn_batch_sources, dn_batch_pages, "
+              "supplier_document_aliases, dn_review_events, dn_pipeline_locks")
+        print("would extend: dn_batches, dn_batch_notes, delivery_notes")
+        print("would add unique indexes: source(batch,file_hash), "
+              "page(source,source_page,region), alias(normalised_key), "
+              "lock(batch), delivery_notes(batch_note)")
+        print("would add partial unique capture-session index on dn_batches")
+        print("\nDRY RUN. No schema changes written.")
+        return 0
+    print(f"target: {PB}   mode: {'DRY RUN' if DRY_RUN else 'APPLY'}")
+
+    batches = require_collection(token, "dn_batches")
+    notes = require_collection(token, "dn_batch_notes")
+    suppliers = require_collection(token, "suppliers")
+    delivery_notes = require_collection(token, "delivery_notes")
+    batch_id = batches["id"]
+    notes_id = notes["id"]
+    supplier_id = suppliers["id"]
+    delivery_notes_id = delivery_notes["id"]
+
+    source_id = create_collection(
+        token, "dn_batch_sources",
+        system_fields() + [
+            relation("batch", batch_id, required=True),
+            field("file", "file", maxSelect=1, maxSize=20971520),
+            field("file_hash"), field("ordinal", "number"), field("mime"),
+            field("original_name"), field("capture_session_id"),
+            field("delivery_session_id"), field("captured_at", "date"),
+            field("latitude", "number"), field("longitude", "number"),
+            field("location_accuracy_m", "number"),
+        ],
+        ["CREATE UNIQUE INDEX `idx_dn_batch_sources_batch_file_hash` "
+         "ON `dn_batch_sources` (`batch`, `file_hash`)"]
+    )
+    create_collection(
+        token, "dn_batch_pages",
+        system_fields() + [
+            relation("batch", batch_id, required=True),
+            relation("source", source_id, required=True),
+            field("source_page", "number"), field("region", "number"),
+            field("image", "file", maxSelect=1, maxSize=20971520),
+            field("image_hash"), field("mime"), field("width", "number"),
+            field("height", "number"), field("rotation", "number"),
+            field("read_status"), field("extraction", "json", maxSize=2000000),
+            field("schema_version"), field("model"), field("prompt_version"),
+            field("read_error"), field("manual_group"),
+        ],
+        ["CREATE UNIQUE INDEX `idx_dn_batch_pages_source_page_region` "
+         "ON `dn_batch_pages` (`source`, `source_page`, `region`)"]
+    )
+    create_collection(
+        token, "supplier_document_aliases",
+        system_fields() + [
+            relation("supplier", supplier_id), field("raw_value"),
+            field("normalised_key"), field("source"), field("approved_by"),
+        ],
+        ["CREATE UNIQUE INDEX `idx_supplier_document_aliases_normalised_key` "
+         "ON `supplier_document_aliases` (`normalised_key`)"]
+    )
+    create_collection(
+        token, "dn_review_events",
+        system_fields() + [
+            relation("batch", batch_id, required=True), relation("note", notes_id),
+            field("action"), field("before", "json", maxSize=2000000),
+            field("after", "json", maxSize=2000000), field("reason"), field("actor"),
+        ], [],
+    )
+    create_collection(
+        token, "dn_pipeline_locks",
+        system_fields() + [relation("batch", batch_id, required=True),
+                           field("expires_at", "date"), field("worker")],
+        ["CREATE UNIQUE INDEX `idx_dn_pipeline_locks_batch` "
+         "ON `dn_pipeline_locks` (`batch`)"]
+    )
+
+    add_fields(token, "dn_batches", [
+        field("stage", "select", maxSelect=1,
+              values=["queued", "preparing", "reading", "split_review", "complete", "failed"]),
+        field("source_count", "number"), field("processing_completed_at", "date"),
+        field("capture_session_id"),
+    ])
+    add_fields(token, "dn_batch_notes", [
+        field("stage", "select", maxSelect=1,
+              values=["matching", "needs_review", "ready_to_post", "posted", "dismissed"]),
+        field("assembly_revision", "number"), field("pages", "json", maxSize=2000000),
+        relation("supplier", supplier_id), field("supplier_name_raw"),
+        field("match_method"), field("match_evidence", "json", maxSize=2000000),
+        field("posted_at", "date"),
+    ])
+    add_fields(token, "delivery_notes", [relation("batch_note", notes_id)])
+
+    add_indexes(token, "delivery_notes", [
+        "CREATE UNIQUE INDEX `idx_delivery_notes_batch_note` "
+        "ON `delivery_notes` (`batch_note`)",
+    ])
+    add_indexes(token, "dn_batches", [
+        "CREATE UNIQUE INDEX `idx_dn_batches_capture_session_id` "
+        "ON `dn_batches` (`capture_session_id`) WHERE `capture_session_id` != ''",
+    ])
+
+    print("\nDRY RUN. No schema changes written." if DRY_RUN else "\nSchema applied.")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/lib/dn-pipeline/stages.test.ts

@@ -0,0 +1,25 @@
+import { describe, expect, it } from "vitest";
+import {
+  displayBatchStage,
+  displayCandidateStage,
+  isCandidateTerminal,
+} from "./stages";
+
+describe("stage compatibility", () => {
+  it("maps legacy states without rewriting live rows", () => {
+    expect(displayBatchStage("analysing", "")).toBe("reading");
+    expect(displayCandidateStage("blocked", "")).toBe("needs_review");
+    expect(displayCandidateStage("promoted", "")).toBe("posted");
+  });
+
+  it("prefers an explicit stage", () => {
+    expect(displayBatchStage("needs_attention", "split_review")).toBe("split_review");
+    expect(displayCandidateStage("blocked", "matching")).toBe("matching");
+  });
+
+  it("recognises terminal candidates", () => {
+    expect(isCandidateTerminal("posted")).toBe(true);
+    expect(isCandidateTerminal("dismissed")).toBe(true);
+    expect(isCandidateTerminal("needs_review")).toBe(false);
+  });
+});

src/lib/dn-pipeline/stages.ts

@@ -0,0 +1,54 @@
+export type BatchStage =
+  | "queued"
+  | "preparing"
+  | "reading"
+  | "split_review"
+  | "complete"
+  | "failed";
+
+export type CandidateStage =
+  | "matching"
+  | "needs_review"
+  | "ready_to_post"
+  | "posted"
+  | "dismissed";
+
+const BATCH_STAGES: readonly BatchStage[] = [
+  "queued",
+  "preparing",
+  "reading",
+  "split_review",
+  "complete",
+  "failed",
+];
+
+const CANDIDATE_STAGES: readonly CandidateStage[] = [
+  "matching",
+  "needs_review",
+  "ready_to_post",
+  "posted",
+  "dismissed",
+];
+
+export function displayBatchStage(legacy: string, explicit: string): BatchStage {
+  if ((BATCH_STAGES as readonly string[]).includes(explicit)) {
+    return explicit as BatchStage;
+  }
+  if (legacy === "analysing") return "reading";
+  if (legacy === "processed") return "complete";
+  if (legacy === "needs_attention") return "split_review";
+  return "queued";
+}
+
+export function displayCandidateStage(legacy: string, explicit: string): CandidateStage {
+  if ((CANDIDATE_STAGES as readonly string[]).includes(explicit)) {
+    return explicit as CandidateStage;
+  }
+  if (legacy === "promoted") return "posted";
+  if (legacy === "dismissed") return "dismissed";
+  if (legacy === "blocked") return "needs_review";
+  return "ready_to_post";
+}
+
+export const isCandidateTerminal = (stage: CandidateStage) =>
+  stage === "posted" || stage === "dismissed";

src/lib/pb-types.ts

@@ -8,6 +8,7 @@
  * RATE_PER_LITRE fallback when it is absent.
  */
 import type { RecordModel } from "./pb";
+import type { BatchStage, CandidateStage } from "./dn-pipeline/stages";
 
 export interface SiteRecord extends RecordModel {
   collectionName: "sites";
@@ -302,7 +303,12 @@ export interface DeliveryNoteRecord extends RecordModel {
   ocr_completed_at?: string;
   received_by_name?: string;
   document_type?: "delivery_note" | "invoice" | "collection_note" | "unknown";
-  expand?: { site?: SiteRecord; purchase_order?: PurchaseOrderRecord };
+  batch_note?: string;
+  expand?: {
+    site?: SiteRecord;
+    purchase_order?: PurchaseOrderRecord;
+    batch_note?: DnBatchNoteRecord;
+  };
 }
 
 export type DeliveryNoteProcessingStatus =
@@ -385,6 +391,10 @@ export interface DnBatchRecord extends RecordModel {
   /** Set explicitly at upload — PocketBase's `created` is unusable here. */
   uploaded_at: string;
   open_pages?: unknown;
+  stage?: BatchStage;
+  source_count?: number;
+  processing_completed_at?: string;
+  capture_session_id?: string;
   expand?: { site?: SiteRecord };
 }
 
@@ -436,6 +446,84 @@ export interface DnBatchNoteRecord extends RecordModel {
   status: DnBatchNoteStatus;
   /** Back-reference to the DELIVERY_NOTES row this note was promoted into. Empty while a candidate or blocked. */
   delivery_note: string;
+  stage?: CandidateStage;
+  assembly_revision?: number;
+  pages?: unknown;
+  supplier?: string;
+  supplier_name_raw?: string;
+  match_method?: string;
+  match_evidence?: unknown;
+  posted_at?: string;
+  expand?: { batch?: DnBatchRecord; supplier?: SupplierRecord; delivery_note?: DeliveryNoteRecord };
+}
+
+export interface DnBatchSourceRecord extends RecordModel {
+  collectionName: "dn_batch_sources";
+  batch: string;
+  file: string;
+  file_hash: string;
+  ordinal: number;
+  mime: string;
+  original_name: string;
+  capture_session_id?: string;
+  delivery_session_id?: string;
+  captured_at?: string;
+  latitude?: number;
+  longitude?: number;
+  location_accuracy_m?: number;
+  expand?: { batch?: DnBatchRecord };
+}
+
+export interface DnBatchPageRecord extends RecordModel {
+  collectionName: "dn_batch_pages";
+  batch: string;
+  source: string;
+  source_page: number;
+  region: number;
+  image: string;
+  image_hash: string;
+  mime: string;
+  width: number;
+  height: number;
+  rotation: number;
+  read_status: string;
+  extraction: unknown;
+  schema_version: string;
+  model: string;
+  prompt_version: string;
+  read_error: string;
+  manual_group: string;
+  expand?: { batch?: DnBatchRecord; source?: DnBatchSourceRecord };
+}
+
+export interface SupplierDocumentAliasRecord extends RecordModel {
+  collectionName: "supplier_document_aliases";
+  supplier: string;
+  raw_value: string;
+  normalised_key: string;
+  source: string;
+  approved_by: string;
+  expand?: { supplier?: SupplierRecord };
+}
+
+export interface DnReviewEventRecord extends RecordModel {
+  collectionName: "dn_review_events";
+  batch: string;
+  note: string;
+  action: string;
+  before: unknown;
+  after: unknown;
+  reason: string;
+  actor: string;
+  expand?: { batch?: DnBatchRecord; note?: DnBatchNoteRecord };
+}
+
+export interface DnPipelineLockRecord extends RecordModel {
+  collectionName: "dn_pipeline_locks";
+  batch: string;
+  expires_at: string;
+  worker: string;
+  expand?: { batch?: DnBatchRecord };
 }
 
 export type DocumentStatus =