import { v4 as uuidv4 } from 'uuid';
import { Queue } from 'bullmq';

interface IngestedDocument {
  correlationId: string;
  storagePath: string;
  originalName: string;
  mimeType: string;
  fileSizeBytes: number;
  status: 'queued' | 'processing' | 'completed' | 'failed';
}

const documentQueue = new Queue('invoice-pipeline', {
  connection: { host: 'redis', port: 6379 },
  defaultJobOptions: { attempts: 3, backoff: { type: 'exponential', delay: 900000 } }
});

export async function registerDocument(file: Express.Multer.File): Promise<IngestedDocument> {
  const allowedTypes = ['application/pdf', 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', 'image/png', 'image/jpeg'];
  if (!allowedTypes.includes(file.mimetype)) {
    throw new Error(`Unsupported MIME type: ${file.mimetype}`);
  }
  if (file.size > 20 * 1024 * 1024) {
    throw new Error('File exceeds 20MB ingestion limit');
  }

  const jobPayload: IngestedDocument = {
    correlationId: uuidv4(),
    storagePath: file.path,
    originalName: file.originalname,
    mimeType: file.mimetype,
    fileSizeBytes: file.size,
    status: 'queued'
  };

  await documentQueue.add('process-invoice', jobPayload, { jobId: jobPayload.correlationId });
  return jobPayload;
}

2. Extraction Stage (ParseFlow Integration)

import FormData from 'form-data';
import fs from 'fs';

interface ExtractionRequest {
  correlationId: string;
  storagePath: string;
  originalName: string;
}

interface ExtractionResult {
  invoice_number: string;
  invoice_date: string;
  due_date: string;
  vendor_name: string;
  vendor_address: string;
  vendor_tax_id: string;
  line_items: Array<{ description: string; quantity: number; unit_price: number; total: number }>;
  subtotal: number;
  tax_amount: number;
  total_amount: number;
  currency: string;
  payment_terms: string;
}

export async function extractDocumentData(payload: ExtractionRequest): Promise<ExtractionResult> {
  const form = new FormData();
  form.append('file', fs.createReadStream(payload.storagePath), payload.originalName);
  form.append('fields', JSON.stringify([
    'invoice_number', 'invoice_date', 'due_date',
    'vendor_name', 'vendor_address', 'vendor_tax_id',
    'line_items', 'subtotal', 'tax_amount', 'total_amount',
    'currency', 'payment_terms'
  ]));

  const response = await fetch('https://parseflow.dev/api/extract', {
    method: 'POST',
    headers: {
      'Authorization': `Bearer ${process.env.PARSEFLOW_API_KEY}`,
      ...form.getHeaders()
    },
    body: form
  });

  if (!response.ok) {
    const errorBody = await response.json().catch(() => ({}));
    throw new Error(`Extraction API failed: ${response.status} - ${errorBody.message || 'Unknown error'}`);
  }

  return response.json();
}

3. Validation & Mathematical Bounds

interface ValidationReport {
  isValid: boolean;
  violations: string[];
}

export function validateExtractedData(raw: Partial<ExtractionResult>): ValidationReport {
  const violations: string[] = [];
  const tolerance = 0.02;

  // Schema requirements
  if (!raw.invoice_number) violations.push('Missing invoice identifier');
  if (!raw.vendor_name) violations.push('Missing vendor designation');
  if (!raw.total_amount) violations.push('Missing total payable amount');

  // Line item reconciliation
  if (raw.line_items && raw.line_items.length > 0) {
    const calculatedSubtotal = raw.line_items.reduce((acc, item) => acc + (item.total || 0), 0);
    if (Math.abs(calculatedSubtotal - (raw.subtotal || 0)) > tolerance) {
      violations.push(`Line item sum (${calculatedSubtotal.toFixed(2)}) diverges from subtotal (${raw.subtotal?.toFixed(2)})`);
    }
  }

  // Tax & total reconciliation
  if (raw.subtotal !== undefined && raw.tax_amount !== undefined && raw.total_amount !== undefined) {
    const expectedTotal = raw.subtotal + raw.tax_amount;
    if (Math.abs(expectedTotal - raw.total_amount) > tolerance) {
      violations.push(`Subtotal + tax (${expectedTotal.toFixed(2)}) does not match total (${raw.total_amount.toFixed(2)})`);
    }
  }

  return { isValid: violations.length === 0, violations };
}

4. Enrichment & Vendor Matching

interface SupplierRecord {
  id: string;
  legal_name: string;
  gl_account: string;
  cost_center: string;
  approver_email: string;
  payment_method: 'ACH' | 'WIRE' | 'CHECK';
}

export async function enrichWithSupplierData(extracted: ExtractionResult): Promise<ExtractionResult & { requires_review: boolean; review_reason?: string }> {
  // Simulated fuzzy match against internal supplier directory
  const matchedSupplier = await db.suppliers.findBestMatch(extracted.vendor_name);

  if (matchedSupplier) {
    return {
      ...extracted,
      supplier_id: matchedSupplier.id,
      gl_account: matchedSupplier.gl_account,
      cost_center: matchedSupplier.cost_center,
      approver_email: matchedSupplier.approver_email,
      payment_method: matchedSupplier.payment_method,
      requires_review: false
    };
  }

  return {
    ...extracted,
    requires_review: true,
    review_reason: 'Vendor not found in approved supplier registry'
  };
}

5. Pipeline Orchestrator & Error Routing

import { Job } from 'bullmq';

export async function executePipeline(job: Job<IngestedDocument>) {
  const doc = job.data;
  try {
    job.updateProgress(20);
    const extracted = await extractDocumentData(doc);
    
    job.updateProgress(50);
    const validation = validateExtractedData(extracted);
    if (!validation.isValid) {
      await job.moveToFailed(new Error(`Validation failed: ${validation.violations.join('; ')}`), true);
      return;
    }

    job.updateProgress(75);
    const enriched = await enrichWithSupplierData(extracted);
    await db.invoices.insert({ ...enriched, correlation_id: doc.correlationId });

    job.updateProgress(90);
    if (enriched.total_amount > 5000 || enriched.requires_review) {
      await notificationService.sendApprovalRequest({
        recipient: enriched.approver_email,
        subject: `AP Approval Required: ${enriched.invoice_number}`,
        payload: enriched
      });
    }

    job.updateProgress(100);
    await job.moveToCompleted({ status: 'routed' }, true);
  } catch (err) {
    const attempts = job.attemptsMade;
    if (attempts >= 3) {
      await job.moveToFailed(err as Error, true);
      await opsAlerting.dispatch({ type: 'dead_letter', correlationId: doc.correlationId, error: (err as Error).message });
    } else {
      // BullMQ handles exponential backoff automatically via job options
      throw err;
    }
  }
}

Pitfall Guide

1. Blind Trust in Extraction Probabilities

Explanation: Modern extraction APIs return confidence scores, but they are probabilistic. Relying solely on API output without deterministic validation introduces financial risk. Fix: Always run mathematical reconciliation (line items → subtotal → tax → total) and schema validation before persisting data. Treat extraction as a draft, not a final record.

2. Ignoring Currency & Locale Variance

Explanation: Invoices from international vendors use different decimal separators, currency codes, and date formats. Naive parsing breaks on 1.234,56 vs 1,234.56. Fix: Normalize all monetary values to a base currency using a live exchange rate service. Parse dates using ISO 8601 standards and validate against expected fiscal periods.

3. Synchronous Processing Bottlenecks

Explanation: Running extraction and validation in a single HTTP request blocks the ingestion endpoint. API latency spikes or rate limits cause timeouts and lost documents. Fix: Decouple ingestion from processing using a message queue. Return a correlation ID immediately, then stream progress via WebSockets or polling endpoints.

4. Missing Idempotency Guarantees

Explanation: Email forwards, SFTP syncs, and user retries frequently submit the same document multiple times. Without deduplication, your ledger receives duplicate liabilities. Fix: Generate a deterministic hash of the file content or use the extraction API's document fingerprint. Check against a processed_hashes table before queuing.

5. Over-Aggressive Fuzzy Matching

Explanation: Vendor name matching using simple Levenshtein distance can incorrectly map Acme Corp to Acme Corporation LLC, attaching wrong GL accounts or payment terms. Fix: Implement a two-tier matching strategy: exact match first, then fuzzy match with a confidence threshold (e.g., ≥0.85). Route low-confidence matches to manual review.

6. Inadequate Dead-Letter Queue Handling

Explanation: Documents that fail after max retries often disappear into logs. Without structured DLQ routing, ops teams cannot triage or reprocess them efficiently. Fix: Persist failed jobs with full context (original file, extraction output, error stack, attempt count). Build a DLQ dashboard with one-click reprocessing and manual override capabilities.

7. Skipping Mathematical Tolerance Bounds

Explanation: Rounding differences between vendor invoices and internal calculations frequently trigger false validation failures. Strict equality checks (===) break on floating-point arithmetic. Fix: Use a tolerance threshold (e.g., 0.02) for all monetary comparisons. Document the tolerance policy in your validation layer and log deviations for audit trails.

Production Bundle

Action Checklist

• Define ingestion limits: enforce file size (20MB), MIME type allowlists, and virus scanning before queueing.
• Implement idempotency: hash incoming files and check against a processed_documents ledger to prevent duplicates.
• Configure extraction API: set up ParseFlow credentials, define required field schemas, and implement circuit breakers for rate limits.
• Build validation rules: enforce mathematical reconciliation, required field checks, and date/future-due validation.
• Map vendor enrichment: connect supplier directory with fuzzy matching thresholds and fallback review routing.
• Establish retry logic: configure exponential backoff (15-minute intervals), max attempts (3), and dead-letter queue persistence.
• Deploy observability: add structured logging, extraction confidence tracking, and pipeline latency metrics to your monitoring stack.

Decision Matrix

Scenario	Recommended Approach	Why	Cost Impact
High-volume, standardized vendors	Full straight-through processing	Predictable layouts yield >96% accuracy; validation catches edge cases	Reduces per-doc cost by ~80%
International vendors, mixed currencies	Extraction + currency normalization + manual review	Exchange rate volatility and locale formatting require human verification	Adds ~$0.05/doc for FX service + review time
New/unknown suppliers	Enrichment fallback + approval routing	Fuzzy matching fails without historical data; routing prevents ledger corruption	Minimal infrastructure cost, shifts labor to AP team
Legacy scanned PDFs (low DPI)	Pre-processing enhancement + extraction	OCR degrades on poor scans; image upscaling improves field detection accuracy	Adds ~$0.02/doc for image preprocessing

Configuration Template

pipeline:
  ingestion:
    max_file_size_mb: 20
    allowed_mime_types:
      - application/pdf
      - application/vnd.openxmlformats-officedocument.spreadsheetml.sheet
      - image/png
      - image/jpeg
    virus_scan: true
  extraction:
    provider: parseflow
    api_endpoint: https://parseflow.dev/api/extract
    timeout_ms: 30000
    retry:
      max_attempts: 3
      backoff_strategy: exponential
      base_delay_ms: 900000
  validation:
    monetary_tolerance: 0.02
    required_fields:
      - invoice_number
      - vendor_name
      - total_amount
    future_due_date_allowed: true
  enrichment:
    fuzzy_match_threshold: 0.85
    unknown_vendor_action: route_to_review
  routing:
    auto_approve_threshold: 5000
    notify_on_review: true
    email_template: ap-approval-request

Quick Start Guide

1. Initialize the queue infrastructure: Deploy a Redis instance and configure BullMQ with the pipeline job options. Set environment variables for PARSEFLOW_API_KEY and database connection strings.
2. Deploy the ingestion endpoint: Spin up the Express/Fastify route with Multer middleware. Apply MIME type filtering and size limits. Return a correlationId immediately upon successful queueing.
3. Run the worker process: Start the BullMQ worker that listens for process-invoice jobs. It will execute the extraction, validation, enrichment, and routing stages sequentially.
4. Verify with a test document: Upload a sample PDF invoice. Monitor the worker logs for extraction confidence scores, validation pass/fail status, and enrichment matches. Check the database for the persisted record and email delivery for threshold-based approvals.
5. Configure observability: Attach structured logging to each stage. Track pipeline_latency_ms, extraction_accuracy_rate, and validation_failure_reasons in your metrics dashboard. Set alerts for dead-letter queue growth or extraction API timeouts.