jorgecuadros-platform/apps/api/src/statements/statements.service.ts

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, scopedRefField } 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();
  }

  /** Scans are stored as blobs, so no object storage means no intake. */
  storageAvailable(): boolean {
    return this.storage.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.",
      );
    }
    // Checked here rather than at the first `put`, which would only surface as
    // a FAILED batch minutes later.
    if (!this.storage.available) {
      throw new BadRequestException(
        "El almacenamiento de documentos no está configurado; no se pueden " +
          "guardar los recibos escaneados.",
      );
    }

    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);
      // Page images are still rendered and stored for every file, text layer or
      // not: the review screen shows the reviewer the page, and "what the
      // parser read" is only checkable against a picture of the paper.
      const textLayer = await this.ocr.textPages(file.buffer).catch(() => []);

      for (const [index, image] of pages.entries()) {
        pageNumber += 1;
        const storageKey = `statement/${batchId}/page-${pageNumber}.png`;
        await this.storage.put(storageKey, image, "image/png");

        try {
          const embedded = textLayer[index] ?? null;
          const ocr = embedded ?? (await this.ocr.recognize(image));
          const parsed = parseStatement(ocr);
          if (embedded) {
            parsed.notes.unshift("texto leído del PDF original, sin 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 },
      });
      const field = batch && scopedRefField(batch.serviceKind);
      if (batch && field) {
        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.");
    }
    const updated = await this.prisma.statementDocument.update({
      where: { id },
      data: { status: "REJECTED", reviewedById, reviewedAt: new Date() },
    });
    // Rejecting the last open page settles the batch just as posting 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 posted is marked REJECTED and the batch itself becomes DISCARDED.
   *
   * Refuses once any page is POSTED — those pages already wrote ledger rows
   * against a check, and a "discarded" label on the batch would leave those
   * charges unexplained. Reject the remaining pages individually instead.
   */
  async discardBatch(batchId: string, reviewedById: string) {
    const batch = await this.prisma.statementBatch.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.statementDocument.count({
      where: { batchId, status: "POSTED" },
    });
    if (posted > 0) {
      throw new BadRequestException(
        `No se puede descartar: ${posted} página(s) ya se registraron en el estado de cuenta.`,
      );
    }

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

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

    return { batchId, rejected: count };
  }

  // --- 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 = scopedRefField(kind);
    if (!field) return;
    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.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() },
      });
    }
  }

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