jorgecuadros-platform/apps/api/src/policy-ocr/policy-ocr.service.ts

import {
  BadRequestException,
  Inject,
  Injectable,
  Logger,
  NotFoundException,
} from "@nestjs/common";
import { Currency, Prisma } from "@jorgecuadros/database";
import { PrismaService } from "../prisma/prisma.service";
import { StorageService } from "../storage/storage.service";
import type { UploadedFileLike } from "../storage/upload-file";
import { OCR_PROVIDER, type OcrPage, type OcrProvider } from "../statements/ocr/ocr.provider";
import { parsePolicy, type ParsedDriver, type ParsedVehicle } from "./parsers/policy-parser";
import { PolicyMatcherService } from "./policy-matcher.service";
import type {
  ConfirmPolicyBatchDto,
  ConfirmPolicyDocumentDto,
  ReviewPolicyDocumentDto,
} from "./policy-ocr.dto";

/**
 * Insurance OCR intake — mirrors the statement pipeline at
 * `apps/api/src/statements/statements.service.ts`. Reuses the OCR seam and
 * Tesseract binding unchanged; the parsers and matcher are policy-specific.
 *
 * Why a parallel pipeline rather than a column on StatementDocument: the
 * matcher keys on `Policy.policyNumber`, the confirm step writes to a
 * different table (`Policy`, not `Transaction`), and the review UI shows
 * different fields. Sharing one queue would either bloat the row with null
 * columns or force the review screen to branch on a discriminator — both
 * worse than a thin second table.
 */
@Injectable()
export class PolicyOcrService {
  private readonly logger = new Logger(PolicyOcrService.name);

  constructor(
    private readonly prisma: PrismaService,
    private readonly storage: StorageService,
    private readonly matcher: PolicyMatcherService,
    @Inject(OCR_PROVIDER) private readonly ocr: OcrProvider,
  ) {}

  ocrAvailable(): Promise<boolean> {
    return this.ocr.available();
  }

  storageAvailable(): boolean {
    return this.storage.available;
  }

  // --- ingest ---------------------------------------------------------------

  async createBatch(
    files: UploadedFileLike[],
    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 leer PDFs de pólizas.",
      );
    }
    if (!this.storage.available) {
      throw new BadRequestException(
        "El almacenamiento de documentos no está configurado; no se pueden " +
          "guardar los PDFs escaneados.",
      );
    }

    // The provider is not asked of the uploader and not assumed: `process`
    // sets it from what the parsers actually claimed, so the batch label can
    // never contradict its own documents. Until then it says so.
    const batch = await this.prisma.policyOcrBatch.create({
      data: { provider: "por detectar", uploadedById, label, fileCount: files.length },
    });

    const copies = files.map((f) => ({ buffer: f.buffer, name: f.originalname }));
    void this.process(batch.id, copies).catch(async (err) => {
      this.logger.error(`Policy OCR batch ${batch.id} failed: ${(err as Error).message}`);
      await this.prisma.policyOcrBatch.update({
        where: { id: batch.id },
        data: { status: "FAILED", error: (err as Error).message },
      });
    });

    return batch;
  }

  /**
   * Render → text → parse → match, **one PolicyOcrDocument row per uploaded
   * file**. The GMX certificate is a 2-page PDF where page 1 carries the
   * contract header and page 2 carries the per-coverage table — both pages
   * describe the SAME policy, so the parser concatenates them and the
   * matcher runs once. `pageNumber` on the row is repurposed as the file
   * ordinal within the batch (1, 2, 3…) — the unique constraint
   * `(batchId, pageNumber)` still holds and lets a single batch carry many
   * policies.
   *
   * The doc's `storageKey` is the SOURCE PDF (`policy-ocr/{batchId}/source-N.pdf`)
   * rather than a rendered page image, so the review screen can embed the
   * exact artifact the office received. The rendered page PNGs are still
   * stored under `policy-ocr/{batchId}/page-M.png` for any future re-OCR or
   * image-based audit, but they aren't used as `storageKey` for the document.
   */
  private async process(
    batchId: string,
    files: { buffer: Buffer; name?: string }[],
  ) {
    await this.prisma.policyOcrBatch.update({
      where: { id: batchId },
      data: { status: "PROCESSING" },
    });

    let fileOrdinal = 0;
    let globalPageOrdinal = 0;
    const providersSeen = new Set<string>();
    for (const file of files) {
      fileOrdinal += 1;
      const sourceKey = `policy-ocr/${batchId}/source-${fileOrdinal}.pdf`;
      await this.storage.put(sourceKey, file.buffer, "application/pdf");

      const pages = await this.ocr.renderPages(file.buffer);
      const textLayer = await this.ocr.textPages(file.buffer).catch(() => []);

      // One OcrPage per rendered page: text-layer wins when present (cheap,
      // exact), OCR the rendered image when it isn't. Same precedence rule
      // as the statement OCR pipeline.
      const perPageOcr: OcrPage[] = [];
      for (const [index, image] of pages.entries()) {
        globalPageOrdinal += 1;
        const pageStorageKey = `policy-ocr/${batchId}/page-${globalPageOrdinal}.png`;
        await this.storage.put(pageStorageKey, image, "image/png");

        const embedded = textLayer[index] ?? null;
        const pageOcr = embedded ?? (await this.ocr.recognize(image));
        perPageOcr.push(pageOcr);
      }

      // Concatenate every page's text with a blank line between pages so the
      // parser's anchored regexes (^From$, ^Currency\s+...) still work
      // across page boundaries — pdftotext -bbox-layout produces newline-
      // separated text per page already, the `\n\n` just preserves a clear
      // boundary in ocrRawText for debugging.
      const mergedText = perPageOcr.map((p) => p.text).join("\n\n");
      const avgConfidence =
        perPageOcr.length === 0
          ? 0
          : perPageOcr.reduce((s, p) => s + p.confidence, 0) / perPageOcr.length;
      const synthetic: OcrPage = {
        text: mergedText,
        words: [],
        confidence: avgConfidence,
      };

      try {
        const parsed = parsePolicy(synthetic);
        if (parsed.provider === "") {
          throw new Error("no se reconoció el proveedor");
        }
        providersSeen.add(parsed.provider);
        const match = await this.matcher.match(parsed);
        const notes = [...parsed.notes, match.note].filter(Boolean);
        // Confident when exactly one Policy carries the printed number —
        // the only unambiguous hit we trust. A new policy (no match) still
        // needs a customer pick, so it stays in review.
        const trusted = match.confident && parsed.policyNumber != null;

        await this.prisma.policyOcrDocument.create({
          data: {
            batchId,
            pageNumber: fileOrdinal,
            storageKey: sourceKey,
            status: trusted ? "MATCHED" : "NEEDS_REVIEW",
            ocrRawText: mergedText,
            ocrConfidence: new Prisma.Decimal(avgConfidence.toFixed(3)),
            provider: parsed.provider,
            extractedPolicyNumber: parsed.policyNumber,
            extractedInsuredName: parsed.insuredName,
            extractedAdditionalInsured: parsed.additionalInsured,
            extractedAgentName: parsed.agentName,
            extractedLegalAddress: parsed.legalAddress,
            extractedZip: parsed.zip,
            extractedPolicyFrom: parsed.policyFrom,
            extractedPolicyTo: parsed.policyTo,
            extractedPolicyDate: parsed.policyDate,
            extractedCurrency: parsed.currency,
            extractedNetPremium:
              parsed.netPremium != null ? new Prisma.Decimal(parsed.netPremium) : null,
            extractedPolicyFee:
              parsed.policyFee != null ? new Prisma.Decimal(parsed.policyFee) : null,
            extractedBrokerFee:
              parsed.brokerFee != null ? new Prisma.Decimal(parsed.brokerFee) : null,
            extractedTotal:
              parsed.total != null ? new Prisma.Decimal(parsed.total) : null,
            extractedCoveragesJson: parsed.coverages.length
              ? (parsed.coverages as unknown as Prisma.InputJsonValue)
              : Prisma.DbNull,
            extractedPremiumPayment: parsed.premiumPayment,
            extractedCoveragePeriodDays: parsed.coveragePeriodDays,
            extractedVehiclesJson: parsed.vehicles.length
              ? (parsed.vehicles as unknown as Prisma.InputJsonValue)
              : Prisma.DbNull,
            extractedDriversJson: parsed.drivers.length
              ? (parsed.drivers as unknown as Prisma.InputJsonValue)
              : Prisma.DbNull,
            extractedPolicyTypeName: parsed.policyTypeName,
            matchedPolicyId: match.policyId,
            matchedCustomerId: match.customerId,
            matchCandidates: match.candidates.length
              ? (match.candidates as unknown as Prisma.InputJsonValue)
              : Prisma.DbNull,
            matchNote: notes.join("; "),
          },
        });
      } catch (err) {
        // The file as a whole failed to parse (no provider, parse exception).
        // One OCR_FAILED row per file is the right granularity — the page
        // images are still on disk for a re-run after a parser fix.
        await this.prisma.policyOcrDocument.create({
          data: {
            batchId,
            pageNumber: fileOrdinal,
            storageKey: sourceKey,
            status: "OCR_FAILED",
            matchNote: (err as Error).message,
          },
        });
      }
    }

    await this.prisma.policyOcrBatch.update({
      where: { id: batchId },
      data: {
        status: "READY_FOR_REVIEW",
        // Whatever the parsers claimed. A mixed upload is labelled as mixed
        // rather than as whichever provider happened to come first — the
        // review header is the only place staff see what they dropped in.
        provider: [...providersSeen].sort().join(" + ") || "desconocido",
      },
    });
  }

  // --- reads ----------------------------------------------------------------

  async listBatches(page: number, pageSize: number) {
    const [total, items] = await this.prisma.$transaction([
      this.prisma.policyOcrBatch.count(),
      this.prisma.policyOcrBatch.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.policyOcrBatch.findUnique({
      where: { id },
      include: { uploadedBy: { select: { name: true } } },
    });
    if (!batch) throw new NotFoundException("Lote no encontrado.");

    const counts = await this.prisma.policyOcrDocument.groupBy({
      by: ["status"],
      where: { batchId: id },
      _count: { _all: true },
    });
    return {
      ...batch,
      byStatus: Object.fromEntries(counts.map((c) => [c.status, c._count._all])),
    };
  }

  async listDocuments(batchId: string) {
    return this.prisma.policyOcrDocument.findMany({
      where: { batchId },
      orderBy: { pageNumber: "asc" },
      include: {
        matchedCustomer: { select: { id: true, name: true } },
        matchedPolicy: {
          select: {
            id: true,
            policyNumber: true,
            customerId: true,
            customer: { select: { name: true } },
          },
        },
      },
    });
  }

  /**
   * The source PDF for the document, so the review screen can show the
   * exact artifact the office uploaded (the browser's PDF viewer handles
   * scrolling, zoom, and selection natively). The rendered page PNGs
   * remain on disk under `policy-ocr/{batchId}/page-N.png` for any
   * future re-OCR, but the doc row points here at the source.
   */
  async pageImage(documentId: string) {
    const doc = await this.prisma.policyOcrDocument.findUnique({
      where: { id: documentId },
      select: { storageKey: true },
    });
    if (!doc) throw new NotFoundException("Documento no encontrado.");
    return this.storage.getStream(doc.storageKey);
  }

  // --- review ---------------------------------------------------------------

  async review(id: string, dto: ReviewPolicyDocumentDto, reviewedById: string) {
    const doc = await this.prisma.policyOcrDocument.findUnique({ where: { id } });
    if (!doc) throw new NotFoundException("Documento no encontrado.");
    if (doc.status === "POSTED") {
      throw new BadRequestException("Este documento ya fue aplicado.");
    }

    // Trusting a customer-supplied pair (policyId, customerId) without
    // cross-check is how a document lands on the wrong customer's ledger;
    // pin them here from the DB.
    let matchedPolicyId = dto.matchedPolicyId ?? doc.matchedPolicyId;
    let matchedCustomerId = doc.matchedCustomerId;

    if (matchedPolicyId) {
      const p = await this.prisma.policy.findUnique({
        where: { id: matchedPolicyId },
        select: { customerId: true },
      });
      if (!p) throw new BadRequestException("Póliza no encontrada.");
      matchedCustomerId = p.customerId;
    } else if (dto.matchedCustomerId) {
      const c = await this.prisma.customer.findUnique({
        where: { id: dto.matchedCustomerId },
        select: { id: true },
      });
      if (!c) throw new BadRequestException("Cliente no encontrado.");
      matchedCustomerId = c.id;
    }

    return this.prisma.policyOcrDocument.update({
      where: { id },
      data: {
        extractedPolicyNumber: dto.policyNumber ?? undefined,
        extractedInsuredName: dto.insuredName ?? undefined,
        extractedAdditionalInsured: dto.additionalInsured ?? undefined,
        extractedAgentName: dto.agentName ?? undefined,
        extractedLegalAddress: dto.legalAddress ?? undefined,
        extractedZip: dto.zip ?? undefined,
        extractedPolicyFrom: dto.policyFrom ? new Date(dto.policyFrom) : undefined,
        extractedPolicyTo: dto.policyTo ? new Date(dto.policyTo) : undefined,
        extractedPolicyDate: dto.policyDate ? new Date(dto.policyDate) : undefined,
        extractedCurrency: dto.currency ?? undefined,
        extractedNetPremium:
          dto.netPremium != null ? new Prisma.Decimal(dto.netPremium) : undefined,
        extractedPolicyFee:
          dto.policyFee != null ? new Prisma.Decimal(dto.policyFee) : undefined,
        extractedBrokerFee:
          dto.brokerFee != null ? new Prisma.Decimal(dto.brokerFee) : undefined,
        extractedTotal:
          dto.total != null ? new Prisma.Decimal(dto.total) : undefined,
        extractedCoveragesJson: dto.coveragesJson
          ? (dto.coveragesJson as Prisma.InputJsonValue)
          : undefined,
        extractedPremiumPayment: dto.premiumPayment ?? undefined,
        extractedCoveragePeriodDays: dto.coveragePeriodDays ?? undefined,
        matchedPolicyId,
        matchedCustomerId,
        status: dto.forceConfirm ? "CONFIRMED" : "MATCHED",
        reviewedById,
        reviewedAt: new Date(),
      },
    });
  }

  async reject(id: string, reviewedById: string) {
    const doc = await this.prisma.policyOcrDocument.findUnique({ where: { id } });
    if (!doc) throw new NotFoundException("Documento no encontrado.");
    if (doc.status === "POSTED") {
      throw new BadRequestException("Este documento ya fue aplicado.");
    }
    const updated = await this.prisma.policyOcrDocument.update({
      where: { id },
      data: { status: "REJECTED", reviewedById, reviewedAt: new Date() },
    });
    // Rejecting the last open page settles the batch just as confirming it
    // would — without this, a fully-rejected batch sat in READY_FOR_REVIEW
    // forever because only confirmBatch() ever closed one.
    await this.closeIfDone(doc.batchId);
    return updated;
  }

  /**
   * Throw away a whole batch that is pending review: every page that has not
   * been applied is marked REJECTED and the batch itself becomes DISCARDED.
   *
   * Refuses once any page is POSTED — a partly-applied batch has already
   * written Policy (and possibly Transaction) rows, and hiding the paperwork
   * behind a "discarded" label would leave those rows unexplained. Reject the
   * remaining pages individually instead.
   */
  async discardBatch(batchId: string, reviewedById: string) {
    const batch = await this.prisma.policyOcrBatch.findUnique({
      where: { id: batchId },
    });
    if (!batch) throw new NotFoundException("Lote no encontrado.");
    if (batch.status === "DISCARDED") {
      throw new BadRequestException("Este lote ya fue descartado.");
    }

    const posted = await this.prisma.policyOcrDocument.count({
      where: { batchId, status: "POSTED" },
    });
    if (posted > 0) {
      throw new BadRequestException(
        `No se puede descartar: ${posted} página(s) ya se aplicaron a una póliza.`,
      );
    }

    const { count } = await this.prisma.policyOcrDocument.updateMany({
      where: { batchId, status: { notIn: ["POSTED", "REJECTED"] } },
      data: { status: "REJECTED", reviewedById, reviewedAt: new Date() },
    });

    await this.prisma.policyOcrBatch.update({
      where: { id: batchId },
      data: { status: "DISCARDED", completedAt: new Date() },
    });

    return { batchId, rejected: count };
  }

  // --- confirm --------------------------------------------------------------

  /**
   * Apply every confirmed document: create or update the Policy, attach the
   * source PDF as a PolicyDocument, and (when staff asked + premium parses)
   * write a Transaction row. Each step is guarded by status checks so a
   * double-confirm cannot re-apply a document.
   */
  async confirmBatch(batchId: string, dto: ConfirmPolicyBatchDto, reviewedById: string) {
    const batch = await this.prisma.policyOcrBatch.findUnique({ where: { id: batchId } });
    if (!batch) throw new NotFoundException("Lote no encontrado.");

    const results: { documentId: string; policyId: string; postedTransactionId: string | null }[] = [];

    for (const item of dto.documents) {
      const doc = await this.prisma.policyOcrDocument.findUnique({
        where: { id: item.documentId },
      });
      if (!doc) {
        throw new BadRequestException(`Documento ${item.documentId} no encontrado.`);
      }
      if (doc.status === "POSTED") {
        throw new BadRequestException(
          `El documento página ${doc.pageNumber} ya fue aplicado.`,
        );
      }
      if (!item.policyId && !item.customerId) {
        throw new BadRequestException(
          `Documento página ${doc.pageNumber}: falta póliza destino o cliente.`,
        );
      }

      // 1. Resolve the lookup rows the parser can only name. The reviewer's
      // explicit pick always wins; the parsed name is the fallback.
      const lookups = await this.resolveLookups(item, doc);

      // 2. Resolve target Policy (create or update). Field selection: every
      // non-null `extracted*` on the doc (post-review) is written. Null is
      // preserved — never overwrite an existing Policy's `netPremium` with
      // null because the certificate page didn't carry one.
      let policyId = item.policyId ?? null;

      if (policyId) {
        const updateData = buildPolicyUpdateFromDoc(item, doc, lookups);
        await this.prisma.policy.update({
          where: { id: policyId },
          data: updateData,
        });
      } else {
        // Create under the picked customer. `policyNumber` is the only field
        // that must be present.
        if (!item.policyNumber && !doc.extractedPolicyNumber) {
          throw new BadRequestException(
            `Documento página ${doc.pageNumber}: falta número de póliza.`,
          );
        }
        const createData = buildPolicyCreateFromDoc(item, doc, item.customerId!, lookups);
        const created = await this.prisma.policy.create({
          data: createData,
        });
        policyId = created.id;
      }

      // 3. Vehicles and named drivers, for the providers whose face carries
      // them (ANA's automobile and driver's policies; never GMX Hogar).
      await this.applyVehiclesAndDrivers(doc, policyId);

      // 4. Attach the source PDF as a PolicyDocument. `doc.storageKey`
      // already points at the exact upload (`policy-ocr/{batchId}/source-N.pdf`)
      // so the attach is just a stream copy into the policy's namespace —
      // the previous per-page "which file did this page come from" walk is
      // gone because one PDF = one doc now.
      await this.attachSourcePdf(doc.storageKey, policyId, doc.provider);

      // 5. Optionally post the premium to the ledger. Only when staff
      // explicitly asked (`postPremium` true) and netPremium parses — without
      // that gate a missing premium would silently book $0.
      let postedTransactionId: string | null = null;
      const premium =
        item.netPremium != null
          ? item.netPremium
          : doc.extractedNetPremium != null
          ? Number(doc.extractedNetPremium)
          : null;
      if (item.postPremium && premium && premium > 0) {
        const tx = await this.prisma.transaction.create({
          data: {
            customerId: (await this.policyCustomerId(policyId))!,
            domain: "INSURANCE",
            amount: new Prisma.Decimal(-Math.abs(premium)),
            transactionDate: doc.extractedPolicyDate ?? doc.extractedPolicyFrom ?? new Date(),
            currency: (item.currency ??
              doc.extractedCurrency ??
              "MXN") as Currency,
            reference: item.policyNumber ?? doc.extractedPolicyNumber ?? null,
            period: null,
            captureSource: "OCR",
            captureRef: doc.id,
            message: `Prima de póliza ${item.policyNumber ?? doc.extractedPolicyNumber ?? ""}`,
          },
        });
        postedTransactionId = tx.id;
      }

      await this.prisma.policyOcrDocument.update({
        where: { id: doc.id },
        data: {
          status: "POSTED",
          matchedPolicyId: policyId,
          reviewedById,
          reviewedAt: new Date(),
          createdPolicyId: item.policyId ? null : policyId,
          postedTransactionId,
        },
      });

      results.push({
        documentId: doc.id,
        policyId,
        postedTransactionId,
      });
    }

    await this.closeIfDone(batchId);

    return {
      applied: results.length,
      policies: results.map((r) => r.policyId),
      postedTransactions: results.filter((r) => r.postedTransactionId).length,
    };
  }

  /**
   * Turn the two things the parser can only NAME into foreign keys.
   *
   * The parser is a pure function over text and never touches the database,
   * so it emits `policyTypeName` ("AUTO") and `provider` ("ANA"). Resolving
   * them here keeps that boundary and means a renamed lookup row is a data
   * change rather than a parser change.
   *
   * **Resolve, never create.** A missing `policy_types` row is a signal that
   * a human deleted it (that is exactly how M_EMPR disappeared), and silently
   * recreating it would undo that decision with no record. The field stays
   * null and the reviewer can add the row through the lookups screen.
   *
   * An explicit pick from the reviewer always beats the parsed name.
   */
  private async resolveLookups(
    item: ConfirmPolicyDocumentDto,
    doc: { extractedPolicyTypeName: string | null; provider: string | null },
  ): Promise<{ policyTypeId?: string; insuranceProviderId?: string }> {
    const out: { policyTypeId?: string; insuranceProviderId?: string } = {};

    if (item.policyTypeId) {
      out.policyTypeId = item.policyTypeId;
    } else if (doc.extractedPolicyTypeName) {
      const row = await this.prisma.policyType.findUnique({
        where: { name: doc.extractedPolicyTypeName },
        select: { id: true },
      });
      if (row) out.policyTypeId = row.id;
    }

    if (item.insuranceProviderId) {
      out.insuranceProviderId = item.insuranceProviderId;
    } else if (doc.provider) {
      const name = PROVIDER_ROW_NAME[doc.provider] ?? doc.provider;
      const row = await this.prisma.insuranceProvider.findFirst({
        where: { name },
        select: { id: true },
      });
      if (row) out.insuranceProviderId = row.id;
    }

    return out;
  }

  /**
   * Write the parsed `Vehicle` and `InsuredDriver` rows onto the policy.
   *
   * Both inserts are skipped when an equivalent row is already on the policy.
   * The reason is `confirmBatch` applying to an EXISTING policy: the office
   * uploads a renewal for a car already on file, and a blind insert would
   * leave the customer with the same VIN listed twice with no way to tell
   * which row the renewal belongs to. Matching is on the identifier the
   * document actually prints — the VIN for a vehicle (falling back to the
   * plate, since ANA's TRAILER/TOWING slots have no VIN), the licence number
   * for a driver (falling back to the name).
   *
   * Nothing is ever updated or deleted here. A vehicle whose plate changed
   * lands as a second row for a human to reconcile, which is the safe half
   * of the mistake: an over-write would destroy the only record of what was
   * insured last term.
   */
  private async applyVehiclesAndDrivers(
    doc: { extractedVehiclesJson: Prisma.JsonValue | null; extractedDriversJson: Prisma.JsonValue | null },
    policyId: string,
  ): Promise<void> {
    const vehicles = asArray<ParsedVehicle>(doc.extractedVehiclesJson);
    const drivers = asArray<ParsedDriver>(doc.extractedDriversJson);
    if (vehicles.length === 0 && drivers.length === 0) return;

    const policy = await this.prisma.policy.findUnique({
      where: { id: policyId },
      select: { customerId: true },
    });
    if (!policy) return;

    if (vehicles.length) {
      const existing = await this.prisma.vehicle.findMany({
        where: { policyId },
        select: { vinNumber: true, licensePlate: true },
      });
      const seen = new Set(
        existing.flatMap((v) =>
          [v.vinNumber, v.licensePlate].filter((k): k is string => !!k).map(norm),
        ),
      );
      for (const v of vehicles) {
        const key = norm(v.vinNumber ?? v.licensePlate ?? "");
        if (!key || seen.has(key)) continue;
        seen.add(key);
        await this.prisma.vehicle.create({
          data: {
            policyId,
            customerId: policy.customerId,
            make: v.make,
            // ANA prints one BODY cell, not separate model/body columns, so
            // it lands on `bodyType`; `model` stays null rather than being
            // guessed out of the same string.
            bodyType: v.bodyType,
            modelYear: v.modelYear,
            vinNumber: v.vinNumber,
            licensePlate: v.licensePlate,
            // "VEHICLE" / "TRAILER" / "TOWING" — the printed slot, which is
            // the difference between the insured car and the trailer behind
            // it and has no column of its own.
            notes: v.item && v.item !== "VEHICLE" ? v.item : null,
          },
        });
      }
    }

    if (drivers.length) {
      const existing = await this.prisma.insuredDriver.findMany({
        where: { policyId },
        select: { licenseNumber: true, fullName: true },
      });
      const seen = new Set(
        existing.flatMap((d) =>
          [d.licenseNumber, d.fullName].filter((k): k is string => !!k).map(norm),
        ),
      );
      for (const d of drivers) {
        const key = norm(d.licenseNumber ?? d.fullName ?? "");
        if (!key || seen.has(key)) continue;
        seen.add(key);
        await this.prisma.insuredDriver.create({
          data: { policyId, fullName: d.fullName, licenseNumber: d.licenseNumber },
        });
      }
    }
  }

  /**
   * Stream the source PDF (`sourceKey`, set by `process` on the doc row)
   * into the policy's storage namespace and create a `PolicyDocument`
   * pointer. Trivial now that the doc row holds the exact source key —
   * the old per-page "which file did this page come from" walk is gone.
   */
  private async attachSourcePdf(
    sourceKey: string,
    policyId: string,
    provider: string | null,
  ): Promise<void> {
    const got = await this.storage.getStream(sourceKey);
    const chunks: Buffer[] = [];
    for await (const c of got.stream) chunks.push(c as Buffer);
    const buf = Buffer.concat(chunks);

    const newKey = `policy/${policyId}/${Date.now()}-${crypto.randomUUID()}.pdf`;
    await this.storage.put(newKey, buf, "application/pdf");
    await this.prisma.policyDocument.create({
      data: {
        policyId,
        // Named after whichever parser claimed the page. Was hardcoded
        // `GMX_POLICY`, which mislabelled every ANA upload as a GMX
        // document in the policy's file list.
        documentType: `${provider ?? "OCR"}_POLICY`,
        storageKey: newKey,
      },
    });
  }

  private async policyCustomerId(policyId: string): Promise<string | null> {
    const p = await this.prisma.policy.findUnique({
      where: { id: policyId },
      select: { customerId: true },
    });
    return p?.customerId ?? null;
  }

  private async closeIfDone(batchId: string) {
    const open = await this.prisma.policyOcrDocument.count({
      where: {
        batchId,
        status: { in: ["PENDING_OCR", "NEEDS_REVIEW", "MATCHED", "CONFIRMED"] },
      },
    });
    if (open === 0) {
      await this.prisma.policyOcrBatch.updateMany({
        // `updateMany` + a status filter so a discarded batch is never quietly
        // relabelled COMPLETED by a late reject on one of its pages.
        where: { id: batchId, status: { not: "DISCARDED" } },
        data: { status: "COMPLETED", completedAt: new Date() },
      });
    }
  }
}

/**
 * The parser's provider code is not the carrier's row name in
 * `insurance_providers`, and the two namespaces are allowed to differ.
 *
 * ANA is the case that forces this: the office's book is filed under
 * "ANA SEGUROS" (738 policies). A bare "ANA" row also existed with 1 policy
 * and is merged away by `20260815160000_policy_type_repair`, so an exact-name
 * lookup on the parser's "ANA" would find nothing at all after that migration.
 *
 * Anything not listed resolves by its own name.
 */
const PROVIDER_ROW_NAME: Record<string, string> = {
  ANA: "ANA SEGUROS",
};

/** The lookup FKs resolved for one document, absent when unresolvable. */
interface ResolvedLookups {
  policyTypeId?: string;
  insuranceProviderId?: string;
}

/** Map a (post-review) doc + final confirmed fields onto a `Policy.update`
 *  payload. Every field that is null in both inputs is omitted so we never
 *  write null over a value the Policy already carries (the GMX certificate
 *  has no premium — we must not blank the existing Policy.netPremium). */
function buildPolicyUpdateFromDoc(
  item: ConfirmPolicyDocumentDto,
  doc: {
    extractedPolicyNumber: string | null;
    extractedInsuredName: string | null;
    extractedAdditionalInsured: string | null;
    extractedAgentName: string | null;
    extractedLegalAddress: string | null;
    extractedZip: string | null;
    extractedPolicyFrom: Date | null;
    extractedPolicyTo: Date | null;
    extractedPolicyDate: Date | null;
    extractedCurrency: string | null;
    extractedNetPremium: Prisma.Decimal | null;
    extractedPolicyFee: Prisma.Decimal | null;
    extractedBrokerFee: Prisma.Decimal | null;
    extractedTotal: Prisma.Decimal | null;
    extractedCoveragesJson: Prisma.JsonValue | null;
    extractedPremiumPayment: string | null;
    extractedCoveragePeriodDays: number | null;
  },
  lookups: ResolvedLookups,
): Prisma.PolicyUpdateInput {
  const numOrUndef = (a: number | undefined, b: Prisma.Decimal | null): Prisma.Decimal | undefined => {
    if (a != null) return new Prisma.Decimal(a);
    if (b != null) return b;
    return undefined;
  };
  const dateOrUndef = (a: string | undefined, b: Date | null): Date | undefined => {
    if (a) return new Date(a);
    if (b) return b;
    return undefined;
  };
  const strOrUndef = (a: string | undefined, b: string | null): string | undefined => {
    if (a != null && a !== "") return a;
    if (b != null && b !== "") return b;
    return undefined;
  };

  return {
    policyNumber: strOrUndef(item.policyNumber, doc.extractedPolicyNumber),
    // `connect` rather than a raw id: this is the CHECKED update input. Left
    // undefined when unresolved, so an existing Policy never loses a type or
    // carrier it already had because this document could not name one.
    policyType: lookups.policyTypeId ? { connect: { id: lookups.policyTypeId } } : undefined,
    insuranceProvider: lookups.insuranceProviderId
      ? { connect: { id: lookups.insuranceProviderId } }
      : undefined,
    agentName: strOrUndef(item.agentName, doc.extractedAgentName),
    policyFrom: dateOrUndef(item.policyFrom, doc.extractedPolicyFrom),
    policyTo: dateOrUndef(item.policyTo, doc.extractedPolicyTo),
    policyDate: dateOrUndef(item.policyDate, doc.extractedPolicyDate),
    // Left undefined when the document didn't print a term, so the schema
    // default (365) stands for GMX. ANA's by-the-day policies DO print one,
    // and the default would otherwise turn a 4-day tourist policy into an
    // annual one on the renewals screen.
    coveragePeriodDays:
      item.coveragePeriodDays ?? doc.extractedCoveragePeriodDays ?? undefined,
    currency: strOrUndef(item.currency, doc.extractedCurrency) as Currency | undefined,
    netPremium: numOrUndef(item.netPremium, doc.extractedNetPremium),
    policyFee: numOrUndef(item.policyFee, doc.extractedPolicyFee),
    brokerFee: numOrUndef(item.brokerFee, doc.extractedBrokerFee),
    total: numOrUndef(item.total, doc.extractedTotal),
    // coveragesJson / observations: freeform, keep the GMX data when present.
    coveragesJson:
      item.coveragesJson !== undefined
        ? (item.coveragesJson as Prisma.InputJsonValue)
        : doc.extractedCoveragesJson != null
        ? (doc.extractedCoveragesJson as Prisma.InputJsonValue)
        : undefined,
    // Premium payment cadence ("CONTADO") and insured-name fields land in
    // `observations` so the PolicyForm's edits stay the source of truth for
    // structured fields. The reviewer can move them by hand if needed.
    observations: joinObservations(
      doc.extractedInsuredName,
      doc.extractedAdditionalInsured,
      doc.extractedLegalAddress,
      doc.extractedZip,
      doc.extractedPremiumPayment,
      item,
    ),
  };
}

/** Same shape as `buildPolicyUpdateFromDoc`, but for `Policy.create`. The
 *  `customerId` is supplied separately and `policyNumber` is required (a
 *  Policy without a number can't be re-matched by the OCR pipeline). */
function buildPolicyCreateFromDoc(
  item: ConfirmPolicyDocumentDto,
  doc: {
    extractedPolicyNumber: string | null;
    extractedInsuredName: string | null;
    extractedAdditionalInsured: string | null;
    extractedAgentName: string | null;
    extractedLegalAddress: string | null;
    extractedZip: string | null;
    extractedPolicyFrom: Date | null;
    extractedPolicyTo: Date | null;
    extractedPolicyDate: Date | null;
    extractedCurrency: string | null;
    extractedNetPremium: Prisma.Decimal | null;
    extractedPolicyFee: Prisma.Decimal | null;
    extractedBrokerFee: Prisma.Decimal | null;
    extractedTotal: Prisma.Decimal | null;
    extractedCoveragesJson: Prisma.JsonValue | null;
    extractedPremiumPayment: string | null;
    extractedCoveragePeriodDays: number | null;
  },
  customerId: string,
  lookups: ResolvedLookups,
): Prisma.PolicyUncheckedCreateInput {
  const numOrUndef = (a: number | undefined, b: Prisma.Decimal | null): Prisma.Decimal | undefined => {
    if (a != null) return new Prisma.Decimal(a);
    if (b != null) return b;
    return undefined;
  };
  const dateOrUndef = (a: string | undefined, b: Date | null): Date | undefined => {
    if (a) return new Date(a);
    if (b) return b;
    return undefined;
  };
  const strOrUndef = (a: string | undefined, b: string | null): string | undefined => {
    if (a != null && a !== "") return a;
    if (b != null && b !== "") return b;
    return undefined;
  };

  const policyNumber =
    strOrUndef(item.policyNumber, doc.extractedPolicyNumber);
  if (!policyNumber) {
    // Caller already guards this; the throw is a type-narrowing aid.
    throw new Error("policyNumber required for create");
  }

  return {
    policyNumber,
    customerId,
    policyTypeId: lookups.policyTypeId,
    insuranceProviderId: lookups.insuranceProviderId,
    agentName: strOrUndef(item.agentName, doc.extractedAgentName),
    policyFrom: dateOrUndef(item.policyFrom, doc.extractedPolicyFrom),
    policyTo: dateOrUndef(item.policyTo, doc.extractedPolicyTo),
    policyDate: dateOrUndef(item.policyDate, doc.extractedPolicyDate),
    // Left undefined when the document didn't print a term, so the schema
    // default (365) stands for GMX. ANA's by-the-day policies DO print one,
    // and the default would otherwise turn a 4-day tourist policy into an
    // annual one on the renewals screen.
    coveragePeriodDays:
      item.coveragePeriodDays ?? doc.extractedCoveragePeriodDays ?? undefined,
    currency: strOrUndef(item.currency, doc.extractedCurrency) as Currency | undefined,
    netPremium: numOrUndef(item.netPremium, doc.extractedNetPremium),
    policyFee: numOrUndef(item.policyFee, doc.extractedPolicyFee),
    brokerFee: numOrUndef(item.brokerFee, doc.extractedBrokerFee),
    total: numOrUndef(item.total, doc.extractedTotal),
    coveragesJson:
      item.coveragesJson !== undefined
        ? (item.coveragesJson as Prisma.InputJsonValue)
        : doc.extractedCoveragesJson != null
        ? (doc.extractedCoveragesJson as Prisma.InputJsonValue)
        : undefined,
    observations: joinObservations(
      doc.extractedInsuredName,
      doc.extractedAdditionalInsured,
      doc.extractedLegalAddress,
      doc.extractedZip,
      doc.extractedPremiumPayment,
      item,
    ),
  };
}

function joinObservations(
  insured: string | null,
  additional: string | null,
  address: string | null,
  zip: string | null,
  premiumPayment: string | null,
  item: ConfirmPolicyDocumentDto,
): string | undefined {
  const lines: string[] = [];
  const insuredName = strOrUndefDb(item.insuredName, insured);
  if (insuredName) lines.push(`Asegurado: ${insuredName}`);
  const additionalInsured = strOrUndefDb(item.additionalInsured, additional);
  if (additionalInsured) lines.push(`Asegurado adicional: ${additionalInsured}`);
  const legalAddress = strOrUndefDb(item.legalAddress, address);
  if (legalAddress) lines.push(`Dirección: ${legalAddress}`);
  const zipVal = strOrUndefDb(item.zip, zip);
  if (zipVal) lines.push(`C.P.: ${zipVal}`);
  const cadence = strOrUndefDb(item.premiumPayment, premiumPayment);
  if (cadence) lines.push(`Pago de prima: ${cadence}`);
  return lines.length ? lines.join("\n") : undefined;
}

function strOrUndefDb(a: string | undefined, b: string | null): string | undefined {
  if (a != null && a !== "") return a;
  if (b != null && b !== "") return b;
  return undefined;
}

/** A JSON column the parser wrote as an array, read back as one. Anything
 *  else (null, DbNull, a legacy object shape) is an empty list rather than a
 *  crash — these columns are only ever populated by the parser, so a
 *  surprise shape means old data, not a caller to reject. */
function asArray<T>(value: Prisma.JsonValue | null): T[] {
  return Array.isArray(value) ? (value as unknown as T[]) : [];
}

/** Compare identifiers the way a person would: case- and space-insensitive.
 *  VINs and plates are printed inconsistently ("8BPX206" vs "8BPX 206"). */
function norm(s: string): string {
  return s.replace(/\s+/g, "").toUpperCase();
}