feat(statements): OCR intake for scanned utility bills

Staff key 300+ utility statements per company per month by hand. This adds
the ingest -> split -> OCR -> match -> review pipeline that proposes customer
and amount per page instead (RECEIPT_CAPTURE_SPEC §2), posting through the
existing BillingService.createBatch seam with source=OCR and a per-document
captureRef so machine and hand capture share one write path and audit trail.

Everything was designed against 10 real scanned statements (46 pages of CFE,
CESPT and Telnor bills) rather than from the sample-free spec. The scans have
no text layer at all — they are camera images — so OCR is mandatory, and they
arrive bundled one customer per page. Measured on those pages the parser
identifies the provider 46/46 and reads an account reference 43/46; against
the dev database that is 39/46 (85%) exact auto-match, 40/46 identified, with
the rest genuine review cases. That closes the OCR-provider question in favour
of self-hosted Tesseract: it clears the bar for a queue where a human confirms
every row, and OcrProvider keeps a managed API a one-line swap.

The samples corrected three things the spec had wrong or unknown:

- Clave catastral is NOT predial. DATMEX.clave (934 rows) is what CESPT and
  predial bills print; DATMEX.predial, which PROPERTY_TAX.accountNumber holds,
  has 663 distinct values across 1135 rows and appears on no statement. The
  clave now lives on Property.cadastralKey as the matcher's secondary key;
  predial is left untouched. This had been blocking predial matching.
- Gas was recoverable: 160 of 334 DATMEX.gas values are real account numbers
  (the rest are ESTACIONARIO/CILINDRO descriptors), now in GAS.meterNumber.
- Phone is one billed line per property (534/18/1 across phone1/2/3), so the
  new TELEPHONE ServiceKind backfills from phone1 only, not three rows.

Matching is scoped to one column per service kind and never reads the customer
name — a CESPT receipt prints ARNAIZ ROSAS ELSA AURORA for an account this
office holds under CATT, RANDY, because the printed name is the registrant,
not the current owner. Where a provider prints a payment barcode it beats the
printed label (one CFE label OCR'd a digit too many while its barcode was
correct) and the two cross-check, with disagreement forcing review.

Confirming a document whose service had no reference writes it back, so gas
and any other cold start is a one-time cost rather than a permanent queue.

Verified end to end against the live dev API and MinIO: real scans uploaded
over HTTP, matched, confirmed against a check, and the resulting rows checked
in MySQL (negative amounts, captureSource=OCR, concept derived from the batch
kind, captureRef linking back to each page). Re-confirming a posted batch is
refused. Test data was removed afterwards.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

apps/api/src/statements/statements.service.ts

@@ -0,0 +1,457 @@
+import {
+  BadRequestException,
+  Inject,
+  Injectable,
+  Logger,
+  NotFoundException,
+} from "@nestjs/common";
+import {
+  Prisma,
+  type ServiceKind,
+  type StatementDocumentStatus,
+} from "@jorgecuadros/database";
+import { PrismaService } from "../prisma/prisma.service";
+import { StorageService } from "../storage/storage.service";
+import { BillingService } from "../billing/billing.service";
+import type { UploadedFileLike } from "../storage/upload-file";
+import { OCR_PROVIDER, type OcrProvider } from "./ocr/ocr.provider";
+import { parseStatement } from "./parsers/statement-parser";
+import { StatementMatcherService } from "./statement-matcher.service";
+import type { ConfirmBatchDto, ReviewDocumentDto } from "./statement.dto";
+
+/**
+ * Default ledger concept per service kind. The names are the legacy
+ * `TYPE OF TRX` values already in `type_transactions`, resolved by name once
+ * per confirm rather than hard-coded as ids, which differ per environment.
+ */
+const CONCEPT_BY_KIND: Partial<Record<ServiceKind, string>> = {
+  ELECTRIC: "ELECTRIC",
+  WATER: "WATER",
+  TELEPHONE: "TELEPHONE",
+  GAS: "GAS BUTANO",
+  PROPERTY_TAX: "PROPERTY TAXES",
+  FEDERAL_ZONE: "FEDERAL ZONE",
+  CABLE: "CABLE",
+};
+
+/** Statuses a document can still be worked on from. */
+const OPEN: StatementDocumentStatus[] = ["NEEDS_REVIEW", "MATCHED", "CONFIRMED"];
+
+@Injectable()
+export class StatementsService {
+  private readonly logger = new Logger(StatementsService.name);
+
+  constructor(
+    private readonly prisma: PrismaService,
+    private readonly storage: StorageService,
+    private readonly billing: BillingService,
+    private readonly matcher: StatementMatcherService,
+    @Inject(OCR_PROVIDER) private readonly ocr: OcrProvider,
+  ) {}
+
+  ocrAvailable(): Promise<boolean> {
+    return this.ocr.available();
+  }
+
+  // --- ingest ---------------------------------------------------------------
+
+  /**
+   * Accept a batch of scanned PDFs and start processing.
+   *
+   * Processing is kicked off but deliberately not awaited: 300 pages of OCR is
+   * minutes of CPU, far past any sane HTTP timeout. The caller gets the batch
+   * id immediately and polls its status, which is also what lets the review
+   * queue show partial progress.
+   */
+  async createBatch(
+    files: UploadedFileLike[],
+    serviceKind: ServiceKind,
+    uploadedById: string,
+    label?: string,
+  ) {
+    if (!files?.length) throw new BadRequestException("No se recibió ningún archivo.");
+    if (!(await this.ocr.available())) {
+      throw new BadRequestException(
+        "El servidor no tiene OCR instalado; no se pueden procesar recibos.",
+      );
+    }
+
+    const batch = await this.prisma.statementBatch.create({
+      data: { serviceKind, uploadedById, label, fileCount: files.length },
+    });
+
+    // Buffers are held for the background pass; the request's own copies would
+    // otherwise be garbage once the response is sent.
+    const copies = files.map((f) => ({ buffer: f.buffer, name: f.originalname }));
+    void this.process(batch.id, copies, serviceKind).catch(async (err) => {
+      this.logger.error(`Batch ${batch.id} failed: ${(err as Error).message}`);
+      await this.prisma.statementBatch.update({
+        where: { id: batch.id },
+        data: { status: "FAILED", error: (err as Error).message },
+      });
+    });
+
+    return batch;
+  }
+
+  /** Render → OCR → parse → match, one document row per page. */
+  private async process(
+    batchId: string,
+    files: { buffer: Buffer; name?: string }[],
+    serviceKind: ServiceKind,
+  ) {
+    await this.prisma.statementBatch.update({
+      where: { id: batchId },
+      data: { status: "PROCESSING" },
+    });
+
+    let pageNumber = 0;
+    for (const file of files) {
+      // The source PDF is kept as well as the page images: it is the artifact
+      // the office actually received, and the only way to re-run a corrected
+      // parser over the original later.
+      const sourceKey = `statement/${batchId}/source-${pageNumber + 1}.pdf`;
+      await this.storage.put(sourceKey, file.buffer, "application/pdf");
+
+      const pages = await this.ocr.renderPages(file.buffer);
+      for (const image of pages) {
+        pageNumber += 1;
+        const storageKey = `statement/${batchId}/page-${pageNumber}.png`;
+        await this.storage.put(storageKey, image, "image/png");
+
+        try {
+          const ocr = await this.ocr.recognize(image);
+          const parsed = parseStatement(ocr);
+          const match = await this.matcher.match(parsed, serviceKind);
+
+          const notes = [...parsed.notes, match.note].filter(Boolean);
+          // A confident field match is only trusted when nothing contradicts
+          // it: a barcode that disagrees with the printed number means one of
+          // the two was misread, and which one is a judgement call.
+          const trusted = match.confident && parsed.crossChecked !== false;
+
+          await this.prisma.statementDocument.create({
+            data: {
+              batchId,
+              pageNumber,
+              storageKey,
+              status: trusted ? "MATCHED" : "NEEDS_REVIEW",
+              ocrRawText: ocr.text,
+              ocrConfidence: new Prisma.Decimal(ocr.confidence.toFixed(3)),
+              provider: parsed.provider,
+              extractedAccountRef: parsed.accountRef,
+              extractedAmount:
+                parsed.amount != null ? new Prisma.Decimal(parsed.amount) : null,
+              extractedPeriod: parsed.period,
+              extractedDueDate: parsed.dueDate,
+              extractedCadastralKey: parsed.cadastralKey,
+              matchedPropertyServiceId: match.propertyServiceId,
+              matchedCustomerId: match.customerId,
+              matchNote: notes.join("; ").slice(0, 190),
+            },
+          });
+        } catch (err) {
+          // One unreadable page must not abandon the other 299.
+          await this.prisma.statementDocument.create({
+            data: {
+              batchId,
+              pageNumber,
+              storageKey,
+              status: "OCR_FAILED",
+              matchNote: (err as Error).message.slice(0, 190),
+            },
+          });
+        }
+      }
+    }
+
+    await this.prisma.statementBatch.update({
+      where: { id: batchId },
+      data: { status: "READY_FOR_REVIEW" },
+    });
+  }
+
+  // --- reads ----------------------------------------------------------------
+
+  async listBatches(page: number, pageSize: number) {
+    const [total, items] = await this.prisma.$transaction([
+      this.prisma.statementBatch.count(),
+      this.prisma.statementBatch.findMany({
+        orderBy: { createdAt: "desc" },
+        skip: (page - 1) * pageSize,
+        take: pageSize,
+        include: {
+          uploadedBy: { select: { name: true } },
+          _count: { select: { documents: true } },
+        },
+      }),
+    ]);
+    return { items, total, page, pageSize, pageCount: Math.ceil(total / pageSize) };
+  }
+
+  async getBatch(id: string) {
+    const batch = await this.prisma.statementBatch.findUnique({
+      where: { id },
+      include: { uploadedBy: { select: { name: true } } },
+    });
+    if (!batch) throw new NotFoundException("Lote no encontrado.");
+
+    const counts = await this.prisma.statementDocument.groupBy({
+      by: ["status"],
+      where: { batchId: id },
+      _count: { _all: true },
+    });
+    const totals = await this.prisma.statementDocument.aggregate({
+      where: { batchId: id, status: { in: OPEN } },
+      _sum: { extractedAmount: true },
+    });
+
+    return {
+      ...batch,
+      byStatus: Object.fromEntries(counts.map((c) => [c.status, c._count._all])),
+      pendingTotal: totals._sum.extractedAmount?.toFixed(2) ?? "0.00",
+    };
+  }
+
+  async listDocuments(batchId: string, status?: StatementDocumentStatus) {
+    return this.prisma.statementDocument.findMany({
+      where: { batchId, ...(status ? { status } : {}) },
+      orderBy: { pageNumber: "asc" },
+      include: {
+        matchedCustomer: { select: { id: true, name: true } },
+        matchedPropertyService: {
+          select: {
+            id: true,
+            kind: true,
+            accountNumber: true,
+            meterNumber: true,
+            property: { select: { id: true, addressLine1: true } },
+          },
+        },
+      },
+    });
+  }
+
+  /** The rendered page image, so a reviewer can read what the parser read. */
+  async pageImage(documentId: string) {
+    const doc = await this.prisma.statementDocument.findUnique({
+      where: { id: documentId },
+      select: { storageKey: true },
+    });
+    if (!doc) throw new NotFoundException("Documento no encontrado.");
+    return this.storage.getStream(doc.storageKey);
+  }
+
+  // --- review ---------------------------------------------------------------
+
+  /** Staff correction of an extracted field or of the match itself. */
+  async review(id: string, dto: ReviewDocumentDto, reviewedById: string) {
+    const doc = await this.prisma.statementDocument.findUnique({ where: { id } });
+    if (!doc) throw new NotFoundException("Documento no encontrado.");
+    if (doc.status === "POSTED") {
+      throw new BadRequestException("Este documento ya fue registrado.");
+    }
+
+    // Changing the service implies its owner; deriving the customer here rather
+    // than trusting a client-supplied pair is what stops a page being posted to
+    // one customer's ledger against another customer's service.
+    let matchedCustomerId = doc.matchedCustomerId;
+    let matchedPropertyServiceId = dto.matchedPropertyServiceId ?? undefined;
+
+    if (dto.matchedPropertyServiceId) {
+      const svc = await this.prisma.propertyService.findUnique({
+        where: { id: dto.matchedPropertyServiceId },
+        select: { property: { select: { customerId: true } } },
+      });
+      if (!svc) throw new BadRequestException("Servicio no encontrado.");
+      matchedCustomerId = svc.property.customerId;
+    } else if (dto.matchedCustomerId) {
+      matchedCustomerId = dto.matchedCustomerId;
+
+      // A reviewer picks a *customer*, not one of their service rows. Without
+      // a service the posting still works, but the confirmed reference has
+      // nowhere to be written back, so the same account would land in review
+      // again next month — which is exactly the behaviour that is supposed to
+      // make gas (whose numbers the migration never populated) a one-time cost.
+      // So: if the batch's service kind resolves to exactly one of that
+      // customer's services that has no reference yet, attach it. Exactly one
+      // — with two candidates there is no way to tell which meter or line the
+      // bill belongs to, and guessing would write a real number onto the wrong
+      // service.
+      const batch = await this.prisma.statementBatch.findUnique({
+        where: { id: doc.batchId },
+        select: { serviceKind: true },
+      });
+      if (batch) {
+        const field = batch.serviceKind === "GAS" ? "meterNumber" : "accountNumber";
+        const blank = await this.prisma.propertyService.findMany({
+          where: {
+            kind: batch.serviceKind,
+            [field]: null,
+            property: { customerId: matchedCustomerId },
+          },
+          select: { id: true },
+          take: 2,
+        });
+        if (blank.length === 1) matchedPropertyServiceId = blank[0].id;
+      }
+    }
+
+    return this.prisma.statementDocument.update({
+      where: { id },
+      data: {
+        extractedAccountRef: dto.accountRef ?? undefined,
+        extractedAmount:
+          dto.amount != null ? new Prisma.Decimal(dto.amount) : undefined,
+        extractedPeriod: dto.period ?? undefined,
+        extractedDueDate: dto.dueDate ? new Date(dto.dueDate) : undefined,
+        matchedPropertyServiceId,
+        matchedCustomerId,
+        status: dto.status ?? "MATCHED",
+        reviewedById,
+        reviewedAt: new Date(),
+      },
+    });
+  }
+
+  async reject(id: string, reviewedById: string) {
+    const doc = await this.prisma.statementDocument.findUnique({ where: { id } });
+    if (!doc) throw new NotFoundException("Documento no encontrado.");
+    if (doc.status === "POSTED") {
+      throw new BadRequestException("Este documento ya fue registrado.");
+    }
+    return this.prisma.statementDocument.update({
+      where: { id },
+      data: { status: "REJECTED", reviewedById, reviewedAt: new Date() },
+    });
+  }
+
+  // --- posting --------------------------------------------------------------
+
+  /**
+   * Post every confirmable document in a batch to the ledger.
+   *
+   * This goes through `BillingService.createBatch` — the same method the manual
+   * "Editor" screen uses — rather than writing `Transaction` rows directly, so
+   * OCR-sourced and hand-keyed receipts share one write path, one validation
+   * path and one audit trail. `source: "OCR"` and a per-line `captureRef` of
+   * the document id give the duplicate-post guard something to key on, so a
+   * batch confirmed twice cannot double-charge anyone.
+   */
+  async confirmBatch(batchId: string, dto: ConfirmBatchDto, reviewedById: string) {
+    const batch = await this.prisma.statementBatch.findUnique({
+      where: { id: batchId },
+    });
+    if (!batch) throw new NotFoundException("Lote no encontrado.");
+
+    const docs = await this.prisma.statementDocument.findMany({
+      where: {
+        batchId,
+        status: { in: dto.includeReviewed ? ["MATCHED", "CONFIRMED"] : ["MATCHED"] },
+        matchedCustomerId: { not: null },
+      },
+      orderBy: { pageNumber: "asc" },
+    });
+    if (!docs.length) {
+      throw new BadRequestException("No hay documentos listos para registrar.");
+    }
+
+    const missing = docs.filter((d) => d.extractedAmount == null);
+    if (missing.length) {
+      throw new BadRequestException(
+        `Falta el importe en ${missing.length} documento(s): página(s) ` +
+          missing.map((d) => d.pageNumber).join(", "),
+      );
+    }
+
+    const typeId = dto.typeId ?? (await this.conceptFor(batch.serviceKind));
+
+    const result = await this.billing.createBatch(
+      {
+        domain: "UTILITY",
+        transactionDate: dto.transactionDate,
+        checkNumber: dto.checkNumber,
+        currency: dto.currency ?? "MXN",
+        typeId,
+        lines: docs.map((d) => ({
+          customerId: d.matchedCustomerId!,
+          // Charges are negative in this ledger: a negative amount is what the
+          // customer owes. The parser reads the printed (positive) figure, so
+          // the sign is applied here, at the single point where a statement
+          // becomes a ledger row.
+          amount: -Math.abs(Number(d.extractedAmount)),
+          reference: d.extractedAccountRef ?? undefined,
+          period: d.extractedPeriod ?? undefined,
+          outstanding: dto.outstanding ?? false,
+        })),
+      },
+      { source: "OCR", refs: docs.map((d) => d.id) },
+    );
+
+    // `items[i]` is positionally parallel to `lines[i]` (seam guarantee 1), so
+    // the created rows zip straight back onto the documents that produced them.
+    await this.prisma.$transaction(
+      docs.map((d, i) =>
+        this.prisma.statementDocument.update({
+          where: { id: d.id },
+          data: {
+            status: "POSTED",
+            postedTransactionId: result.items[i].id,
+            reviewedById,
+            reviewedAt: new Date(),
+          },
+        }),
+      ),
+    );
+
+    // Teach the matcher. When a document was matched by clave catastral or by
+    // hand because the scoped field was blank, writing the reference back means
+    // next month's statement for the same account matches on its own — this is
+    // what turns gas (whose numbers the migration never populated) from a
+    // permanent review queue into a one-time cost.
+    await this.learnAccountRefs(docs, batch.serviceKind);
+
+    await this.closeIfDone(batchId);
+
+    return { posted: result.count, total: result.total, checkNumber: dto.checkNumber };
+  }
+
+  /** Write a confirmed reference onto a service that had none. */
+  private async learnAccountRefs(
+    docs: { matchedPropertyServiceId: string | null; extractedAccountRef: string | null }[],
+    kind: ServiceKind,
+  ) {
+    const field = kind === "GAS" ? "meterNumber" : "accountNumber";
+    for (const d of docs) {
+      if (!d.matchedPropertyServiceId || !d.extractedAccountRef) continue;
+      await this.prisma.propertyService.updateMany({
+        // Only fills a hole — never overwrites a number already on file, which
+        // would let one misread page rewrite good reference data.
+        where: { id: d.matchedPropertyServiceId, [field]: null },
+        data: { [field]: d.extractedAccountRef },
+      });
+    }
+  }
+
+  private async closeIfDone(batchId: string) {
+    const open = await this.prisma.statementDocument.count({
+      where: { batchId, status: { in: OPEN } },
+    });
+    if (open === 0) {
+      await this.prisma.statementBatch.update({
+        where: { id: batchId },
+        data: { status: "COMPLETED", completedAt: new Date() },
+      });
+    }
+  }
+
+  private async conceptFor(kind: ServiceKind): Promise<string | undefined> {
+    const name = CONCEPT_BY_KIND[kind];
+    if (!name) return undefined;
+    const row = await this.prisma.typeTransaction.findFirst({
+      where: { nameEn: name },
+      select: { id: true },
+    });
+    return row?.id;
+  }
+}