0pt/src/services/document-capture-service.ts

/**
 * src/services/document-capture-service.ts
 * =========================================
 * Orchestrates the capture half of the 0pt.document_ocr capability:
 *   poll a customer-registered mailbox (Microsoft Graph, app-only) →
 *   download PDF attachments → persist the bytes as an artifact →
 *   enqueue extraction (an N0D job placeholder; a worker or the
 *   /extraction API route performs the real Document Intelligence call).
 *
 * Tenant-agnostic: mailbox address + Graph credentials + folder are all
 * caller-supplied per invocation — nothing here is stored keyed only by a
 * hardcoded mailbox. Nothing here assumes invoices, vendors, or Protosell;
 * that logic (vendor-kb matching, dup-hash, DQ scoring, invoice
 * classification) was intentionally left behind in
 * v0-datahub/lib/invoice-inbox — it does not belong in a generic capture
 * pipeline that also has to serve HR forms, contracts, receipts, etc.
 *
 * KNOWN GAP — ST0RE bytes storage: st0re-client.ts registers artifact
 * METADATA only (a pointer to provider/bucket/key); it has no real
 * bytes-upload endpoint yet. Until one exists, this service writes
 * attachment bytes to LocalBlobStore (src/adapters/local-blob-store.ts), a
 * clearly-marked interim local-disk store, then registers that location
 * with ST0RE so the artifact/lineage metadata is still correct. Search this
 * file for "INTERIM" to find the seam to replace.
 *
 * Singleton note: unlike source-service.ts / dataset-service.ts (which are
 * instantiated fresh per API route file because each request is
 * self-contained), DocumentCaptureService holds the only in-memory record
 * of *which bytes belong to which document_id* — document-extraction-service
 * must read the exact same instance to find them. So this module also
 * exports a shared `documentCaptureService` singleton for both API routes
 * to import, instead of each route constructing its own instance.
 */
import crypto from "crypto";
import { GraphMailboxAdapter, type GraphMailboxConfig, type GraphMessage, type GraphAttachment } from "../adapters/graph-mailbox-adapter";
import { LocalBlobStore } from "../adapters/local-blob-store";
import { createSt0reClient, type St0reClient } from "../clients/st0re-client";
import { createN0dClient, type N0dClient } from "../clients/n0d-client";
import { createAud0Client, type Aud0Client } from "../clients/aud0-client";
import type { DocumentRecord, DocumentStatus } from "../types/document";

export interface CaptureMailboxContext {
  tenant_id: string;
  project_id: string;
  actor_id: string;
  correlation_id: string;
}

export interface CaptureMailboxInput {
  mailbox: GraphMailboxConfig;
  since_iso?: string;
  top?: number;
  max_pages?: number;
}

export interface CaptureMailboxResult {
  scanned_messages: number;
  captured_documents: DocumentRecord[];
  errors: string[];
}

interface StoredBytesLocation {
  bucket: string;
  key: string;
}

export class DocumentCaptureService {
  private documents: Map<string, DocumentRecord> = new Map();
  private bytesByDocumentId: Map<string, StoredBytesLocation> = new Map();
  private st0re: St0reClient;
  private n0d: N0dClient;
  private aud0: Aud0Client;
  private blobStore: LocalBlobStore;

  constructor(opts?: { st0reBaseUrl?: string; n0dBaseUrl?: string; aud0BaseUrl?: string; blobStore?: LocalBlobStore }) {
    this.st0re = createSt0reClient(opts?.st0reBaseUrl);
    this.n0d = createN0dClient(opts?.n0dBaseUrl);
    this.aud0 = createAud0Client(opts?.aud0BaseUrl);
    this.blobStore = opts?.blobStore ?? new LocalBlobStore();
  }

  async captureFromMailbox(input: CaptureMailboxInput, ctx: CaptureMailboxContext): Promise<CaptureMailboxResult> {
    if (!ctx.tenant_id || !ctx.project_id || !ctx.actor_id) {
      throw new Error("captureFromMailbox: missing tenant_id/project_id/actor_id");
    }
    if (!GraphMailboxAdapter.isConfigured(input.mailbox)) {
      throw new Error(
        "captureFromMailbox: missing mailbox.tenantId/clientId/clientSecret/mailboxAddress",
      );
    }
    const graph = new GraphMailboxAdapter(input.mailbox);
    const result: CaptureMailboxResult = { scanned_messages: 0, captured_documents: [], errors: [] };

    const messages = await graph.pollMessages({
      sinceISO: input.since_iso,
      top: input.top,
      maxPages: input.max_pages,
    });

    for (const message of messages) {
      result.scanned_messages++;
      if (!message.hasAttachments) continue;
      try {
        const attachments = await graph.listAttachments(message.id);
        for (const attachment of attachments) {
          if (!attachment.isPdf) continue;
          const document = await this.captureAttachment(input.mailbox, message, attachment, ctx);
          result.captured_documents.push(document);
        }
      } catch (e: unknown) {
        const msg = e instanceof Error ? e.message : String(e);
        result.errors.push(`${message.id}: ${msg.slice(0, 200)}`);
      }
    }
    return result;
  }

  private async captureAttachment(
    mailbox: GraphMailboxConfig,
    message: GraphMessage,
    attachment: GraphAttachment,
    ctx: CaptureMailboxContext,
  ): Promise<DocumentRecord> {
    const documentId = `doc_${crypto.randomUUID()}`;
    const now = new Date().toISOString();
    const bucket = "0pt-documents";
    const safeName = (attachment.name || "attachment.pdf").replace(/[^A-Za-z0-9._-]/g, "_");
    const key = `${ctx.tenant_id}/${ctx.project_id}/${message.id}/${attachment.id}/${safeName}`;

    // INTERIM — see file header. Real bytes persistence until ST0RE exposes an upload endpoint.
    const written = await this.blobStore.write(bucket, key, attachment.contentBytes);
    const checksum = crypto.createHash("sha256").update(attachment.contentBytes).digest("hex");

    let record: DocumentRecord = {
      document_id: documentId,
      tenant_id: ctx.tenant_id,
      project_id: ctx.project_id,
      mailbox_address: mailbox.mailboxAddress,
      graph_message_id: message.id,
      graph_attachment_id: attachment.id,
      filename: attachment.name ?? undefined,
      content_type: attachment.contentType ?? undefined,
      size_bytes: attachment.sizeBytes ?? attachment.contentBytes.length,
      received_at: message.receivedAt ?? undefined,
      sender_email: message.senderEmail ?? undefined,
      sender_name: message.senderName ?? undefined,
      subject: message.subject ?? undefined,
      status: "captured",
      correlation_id: ctx.correlation_id,
      created_at: now,
    };
    this.documents.set(documentId, record);
    this.bytesByDocumentId.set(documentId, { bucket: written.bucket, key: written.key });

    // Register the artifact pointer with ST0RE (metadata only — see file header gap note).
    const artifact = await this.st0re.createArtifact({
      tenant_id: ctx.tenant_id,
      project_id: ctx.project_id,
      artifact_type: "document",
      app_id: "0pt",
      produced_by: "0pt.capture_document",
      object_location: { provider: written.provider, bucket: written.bucket, key: written.key, region: "local" },
      content_type: attachment.contentType || "application/pdf",
      size_bytes: written.sizeBytes,
      checksum_sha256: checksum,
      pii_level: "high",
      retention_policy: "document_capture_standard",
      data_residency: "eu",
      risk_level: "L3",
      correlation_id: ctx.correlation_id,
    });

    // Enqueue extraction. A worker (out of scope here) or a caller hitting the
    // /documents/[documentId]/extraction route performs the real DI call.
    const job = await this.n0d.createJob({
      tenant_id: ctx.tenant_id,
      project_id: ctx.project_id,
      job_type: "0pt.document_extraction_job",
      capability_id: "0pt.extract_document",
      requested_by_app_id: "0pt",
      requested_by_actor_id: ctx.actor_id,
      input_artifact_ids: [artifact.artifact_id],
      output_artifact_type: "document_extraction",
      requires_backbone: true,
      requires_gpu: false,
      required_node_roles: ["compute_general"],
      resource_profile: "cpu_io_medium",
      risk_level: "L3",
      correlation_id: ctx.correlation_id,
    });

    record = {
      ...record,
      st0re_artifact_id: artifact.artifact_id,
      n0d_job_id: job.job_id,
      status: "extraction_requested",
      updated_at: new Date().toISOString(),
    };
    this.documents.set(documentId, record);

    await this.aud0.ingestEvent({
      event_id: `evt_0pt_document_captured_${documentId}`,
      event_type: "0pt.document_captured",
      tenant_id: ctx.tenant_id,
      project_id: ctx.project_id,
      actor_id: ctx.actor_id,
      app_id: "0pt",
      capability_id: "0pt.capture_document",
      resource_type: "document",
      resource_id: documentId,
      risk_level: "L3",
      payload: {
        mailbox_address: mailbox.mailboxAddress,
        graph_message_id: message.id,
        filename: attachment.name,
        size_bytes: written.sizeBytes,
        st0re_artifact_id: artifact.artifact_id,
        n0d_job_id: job.job_id,
      },
      created_at: new Date().toISOString(),
      correlation_id: ctx.correlation_id,
    });

    await this.aud0.ingestLineageEvent({
      lineage_event_id: `lineage_0pt_document_captured_${documentId}`,
      tenant_id: ctx.tenant_id,
      project_id: ctx.project_id,
      app_id: "0pt",
      capability_id: "0pt.capture_document",
      source_type: "mailbox",
      source_id: mailbox.mailboxAddress,
      target_type: "document",
      target_id: documentId,
      lineage_type: "produced",
      metadata: { graph_message_id: message.id, st0re_artifact_id: artifact.artifact_id },
      created_at: new Date().toISOString(),
      correlation_id: ctx.correlation_id,
    });

    return record;
  }

  async getDocument(documentId: string): Promise<DocumentRecord> {
    const record = this.documents.get(documentId);
    if (!record) throw new Error(`Document ${documentId} not found`);
    return record;
  }

  async listDocuments(filter?: { tenant_id?: string; project_id?: string; status?: DocumentStatus }): Promise<DocumentRecord[]> {
    let results = Array.from(this.documents.values());
    if (filter?.tenant_id) results = results.filter((r) => r.tenant_id === filter.tenant_id);
    if (filter?.project_id) results = results.filter((r) => r.project_id === filter.project_id);
    if (filter?.status) results = results.filter((r) => r.status === filter.status);
    return results;
  }

  /** Read back the raw bytes captured for a document. INTERIM local store — see file header. */
  async getDocumentBytes(documentId: string): Promise<Buffer> {
    const loc = this.bytesByDocumentId.get(documentId);
    if (!loc) throw new Error(`Document ${documentId} has no stored bytes`);
    return this.blobStore.read(loc.bucket, loc.key);
  }

  setStatus(documentId: string, status: DocumentStatus): void {
    const record = this.documents.get(documentId);
    if (!record) throw new Error(`Document ${documentId} not found`);
    record.status = status;
    record.updated_at = new Date().toISOString();
  }
}

/** Shared singleton — see the "Singleton note" in the file header for why. */
export const documentCaptureService = new DocumentCaptureService();