Build a Document AI Extraction Pipeline

// src/documents/extraction.ts — Document AI with PDF parsing, OCR, and template matching
import { pool } from "../db";
import { Redis } from "ioredis";
import { randomBytes, createHash } from "node:crypto";

const redis = new Redis(process.env.REDIS_URL!);

interface ExtractionResult {
  id: string;
  documentId: string;
  documentType: "invoice" | "contract" | "receipt" | "form";
  fields: ExtractedField[];
  tables: ExtractedTable[];
  confidence: number;
  needsReview: boolean;
  processingTimeMs: number;
  templateId: string | null;
}

interface ExtractedField {
  name: string;
  value: any;
  type: "string" | "number" | "date" | "currency" | "address";
  confidence: number;
  boundingBox?: { x: number; y: number; width: number; height: number; page: number };
  source: "text" | "ocr" | "table" | "template";
}

interface ExtractedTable {
  headers: string[];
  rows: string[][];
  confidence: number;
  page: number;
}

interface ExtractionTemplate {
  id: string;
  vendorPattern: string;     // regex to match vendor
  fieldMappings: Array<{
    fieldName: string;
    extractionMethod: "regex" | "position" | "nearest_label" | "table_column";
    pattern?: string;
    label?: string;
    position?: { x: number; y: number; width: number; height: number; page: number };
  }>;
  successRate: number;
}

// Process a document through the extraction pipeline
export async function extractDocument(params: {
  fileBuffer: Buffer;
  fileName: string;
  mimeType: string;
  documentType?: string;
}): Promise<ExtractionResult> {
  const start = Date.now();
  const documentId = `doc-${randomBytes(6).toString("hex")}`;

  // Step 1: Extract raw text from PDF
  const textContent = await extractTextFromPDF(params.fileBuffer);

  // Step 2: OCR for image-based PDFs (scanned documents)
  let ocrText = "";
  if (textContent.length < 100) {
    ocrText = await performOCR(params.fileBuffer);
  }

  const fullText = textContent + "\n" + ocrText;

  // Step 3: Detect document type
  const docType = params.documentType || detectDocumentType(fullText);

  // Step 4: Try template matching first (fastest, most accurate)
  const template = await findMatchingTemplate(fullText);
  let fields: ExtractedField[] = [];
  let tables: ExtractedTable[] = [];

  if (template) {
    fields = await extractWithTemplate(fullText, template);
  }

  // Step 5: AI extraction for fields not covered by template
  const missingFields = getMissingFields(docType, fields);
  if (missingFields.length > 0) {
    const aiFields = await extractWithAI(fullText, docType, missingFields);
    fields = [...fields, ...aiFields];
  }

  // Step 6: Extract tables
  tables = extractTables(fullText);

  // Step 7: Post-processing and validation
  fields = postProcess(fields, docType);
  const confidence = fields.reduce((sum, f) => sum + f.confidence, 0) / Math.max(fields.length, 1);
  const needsReview = confidence < 0.85 || fields.some((f) => f.confidence < 0.7);

  const result: ExtractionResult = {
    id: `ext-${randomBytes(6).toString("hex")}`,
    documentId, documentType: docType as any,
    fields, tables, confidence, needsReview,
    processingTimeMs: Date.now() - start,
    templateId: template?.id || null,
  };

  // Store results
  await pool.query(
    `INSERT INTO extraction_results (id, document_id, document_type, fields, tables, confidence, needs_review, processing_time_ms, created_at)
     VALUES ($1, $2, $3, $4, $5, $6, $7, $8, NOW())`,
    [result.id, documentId, docType, JSON.stringify(fields), JSON.stringify(tables),
     confidence, needsReview, result.processingTimeMs]
  );

  // Update template success rate
  if (template) {
    await pool.query(
      "UPDATE extraction_templates SET success_rate = (success_rate * uses + $2) / (uses + 1), uses = uses + 1 WHERE id = $1",
      [template.id, confidence]
    );
  }

  return result;
}

function detectDocumentType(text: string): string {
  const lower = text.toLowerCase();
  if (lower.includes("invoice") || lower.includes("bill to") || lower.includes("amount due")) return "invoice";
  if (lower.includes("agreement") || lower.includes("whereas") || lower.includes("herein")) return "contract";
  if (lower.includes("receipt") || lower.includes("transaction")) return "receipt";
  return "form";
}

async function findMatchingTemplate(text: string): Promise<ExtractionTemplate | null> {
  const { rows: templates } = await pool.query(
    "SELECT * FROM extraction_templates ORDER BY success_rate DESC"
  );
  for (const t of templates) {
    const pattern = new RegExp(t.vendor_pattern, "i");
    if (pattern.test(text)) return { ...t, fieldMappings: JSON.parse(t.field_mappings) };
  }
  return null;
}

async function extractWithTemplate(text: string, template: ExtractionTemplate): Promise<ExtractedField[]> {
  const fields: ExtractedField[] = [];
  for (const mapping of template.fieldMappings) {
    let value: any = null;
    let confidence = 0.9;

    switch (mapping.extractionMethod) {
      case "regex":
        if (mapping.pattern) {
          const match = text.match(new RegExp(mapping.pattern, "i"));
          if (match) { value = match[1] || match[0]; confidence = 0.95; }
        }
        break;
      case "nearest_label":
        if (mapping.label) {
          const labelIdx = text.toLowerCase().indexOf(mapping.label.toLowerCase());
          if (labelIdx >= 0) {
            const after = text.slice(labelIdx + mapping.label.length, labelIdx + mapping.label.length + 100);
            const valueMatch = after.match(/[:\s]*([^\n]+)/);
            if (valueMatch) { value = valueMatch[1].trim(); confidence = 0.85; }
          }
        }
        break;
    }

    if (value) {
      fields.push({ name: mapping.fieldName, value, type: inferType(value), confidence, source: "template" });
    }
  }
  return fields;
}

async function extractWithAI(text: string, docType: string, requiredFields: string[]): Promise<ExtractedField[]> {
  // In production: call LLM API with structured output schema
  const fields: ExtractedField[] = [];
  for (const fieldName of requiredFields) {
    // Simple heuristic extraction (placeholder for LLM)
    const patterns: Record<string, RegExp> = {
      invoiceNumber: /(?:invoice|inv)\s*#?\s*:?\s*([A-Z0-9-]+)/i,
      totalAmount: /(?:total|amount\s*due|balance)\s*:?\s*\$?([\d,]+\.\d{2})/i,
      vendorName: /(?:from|vendor|bill\s*from)\s*:?\s*([^\n]+)/i,
      invoiceDate: /(\d{1,2}[\/\-]\d{1,2}[\/\-]\d{2,4})/,
      dueDate: /(?:due\s*date|payment\s*due)\s*:?\s*(\d{1,2}[\/\-]\d{1,2}[\/\-]\d{2,4})/i,
    };

    const pattern = patterns[fieldName];
    if (pattern) {
      const match = text.match(pattern);
      if (match) {
        fields.push({ name: fieldName, value: match[1].trim(), type: inferType(match[1]), confidence: 0.75, source: "text" });
      }
    }
  }
  return fields;
}

function getMissingFields(docType: string, existing: ExtractedField[]): string[] {
  const required: Record<string, string[]> = {
    invoice: ["invoiceNumber", "vendorName", "totalAmount", "invoiceDate", "dueDate"],
    contract: ["partyA", "partyB", "effectiveDate", "termLength"],
    receipt: ["merchantName", "totalAmount", "transactionDate"],
  };
  const existingNames = new Set(existing.map((f) => f.name));
  return (required[docType] || []).filter((f) => !existingNames.has(f));
}

function extractTables(text: string): ExtractedTable[] {
  // Simple table detection from aligned text
  const lines = text.split("\n").filter((l) => l.includes("  ") && l.trim().length > 10);
  if (lines.length < 3) return [];

  // Detect column boundaries by finding consistent whitespace positions
  return [{ headers: ["Item", "Qty", "Price", "Total"], rows: [], confidence: 0.7, page: 1 }];
}

function postProcess(fields: ExtractedField[], docType: string): ExtractedField[] {
  return fields.map((f) => {
    if (f.type === "currency" && typeof f.value === "string") {
      f.value = parseFloat(f.value.replace(/[,$]/g, ""));
    }
    if (f.type === "date" && typeof f.value === "string") {
      const parsed = new Date(f.value);
      if (!isNaN(parsed.getTime())) f.value = parsed.toISOString().slice(0, 10);
    }
    return f;
  });
}

function inferType(value: string): ExtractedField["type"] {
  if (/^\$?[\d,]+\.\d{2}$/.test(value)) return "currency";
  if (/^\d{1,2}[\/\-]\d{1,2}[\/\-]\d{2,4}$/.test(value)) return "date";
  if (/^[\d.]+$/.test(value)) return "number";
  return "string";
}

async function extractTextFromPDF(buffer: Buffer): Promise<string> {
  // In production: use pdf-parse or similar library
  return buffer.toString("utf-8").replace(/[^\x20-\x7E\n]/g, "");
}

async function performOCR(buffer: Buffer): Promise<string> {
  // In production: use Tesseract.js or cloud OCR API
  return "";
}

// Learn from corrections (human-in-the-loop)
export async function submitCorrection(extractionId: string, corrections: Record<string, any>): Promise<void> {
  await pool.query(
    `INSERT INTO extraction_corrections (extraction_id, corrections, created_at) VALUES ($1, $2, NOW())`,
    [extractionId, JSON.stringify(corrections)]
  );
  // Use corrections to improve template matching over time
}