Orcha Ingestion: Document Sources

How documents enter the system and the webhook handler design.

Document Sources

Documents enter the system through multiple sources, all converging to a common queue-for-ingestion! function:

See Email Integration for details on OAuth-based email ingestion.

User Upload

Multipart file upload from web interface. Accepts PDF and image files.

curl -X POST -F "file=@invoice.pdf" http://localhost:8888/document/new/upload

Response:

{"ingestion/id": "019234ab-...", "document/id": "019234cd-...", "created?": true}

Email Webhooks (Outlook & Gmail)

OAuth-based email integration receives push notifications when new emails arrive. Both providers follow a similar pattern:

1. Webhook receives notification (provider-specific format)
2. Webhook validates request and queues sync task
3. Background worker claims and processes pending syncs
4. Attachments are extracted and passed to queue-for-ingestion!

See Email Integration for full details.

Webhook Handler Design

All intake endpoints share a common queue-for-ingestion! function located in src/com/getorcha/erp/http/ingestion.clj that:

1. Computes content hash (SHA-256)
2. Upserts document by content hash (using PostgreSQL ON CONFLICT)
3. Detects insert vs update using xmax system column (xmax = 0 means INSERT, xmax > 0 means UPDATE)
4. Uploads to S3 only if document was newly created
5. Creates ingestion record with source information
6. Enqueues ingestion ID to SQS

Document ID Strategy: UUIDv7 + Content Hash

Primary Key: UUIDv7

Documents use UUIDv7 as the primary key instead of serial integers or random UUIDs.

Why UUIDv7?

This gives time-ordered inserts for B-tree friendliness while being globally unique without coordination.

PostgreSQL 18+ provides native uuidv7() function. We generate UUIDv7 in Clojure before S3 upload so the document ID is known for the S3 path, then INSERT with explicit ID.

Deduplication: Content Hash

SHA-256 hash of document content stored as a unique secondary index.

CREATE UNIQUE INDEX idx_document_content_hash
ON document(content_hash)
WHERE content_hash IS NOT NULL;

Why not use content hash as primary key?

Content hash could theoretically serve as the document ID (content-addressed storage), but:

1. S3 path before hashing: We need the document ID for the S3 path before computing the hash (can't hash without reading the content, can't upload without a path)
2. Re-ingestion scenarios: A document should retain its original ID regardless of re-processing
3. Hash algorithm changes: If we ever need to change the hash algorithm, the secondary index approach allows migration without rewriting all IDs

Document vs Ingestion: Source Tracking

Source information is stored on the ingestion (ap_ingestion), not the document:

Why this design:

• Same document can be submitted by different sources
• Each ingestion preserves its source context
• Document remains a pure "content entity"

XOR Constraint

Each ingestion must have exactly one source:

CONSTRAINT ingestion_source_xor CHECK (num_nonnulls(doc_source_id, uploaded_by) = 1)

Duplicate Handling

When a document is ingested, behavior depends on whether a document with the same content hash exists:

Key behaviors:

• Every submission creates a new ingestion
• Duplicate content reuses the same document
• S3 upload only happens for new documents
• Each ingestion is processed independently

Re-ingestion Scenarios

When the same content is submitted again:

1. Document upsert finds existing row (xmax > 0)
2. S3 upload is skipped (file already exists)
3. New ingestion is created with current source info
4. Ingestion ID is queued for processing
5. Worker processes the new ingestion

This means:

• Same document can have multiple ingestion records
• Each ingestion tracks its source separately
• Processing happens independently per ingestion
• History is preserved for auditing

File Extension Mapping

The file extension for S3 storage is derived from Content-Type header:

(def mime-type->extension
  {"application/pdf" "pdf"
   "image/png"       "png"
   "image/jpeg"      "jpg"
   "image/tiff"      "tiff"
   "image/gif"       "gif"
   "image/bmp"       "bmp"
   "image/webp"      "webp"})

S3 path format: documents/{document-id}.{ext}

Source Metadata

Each ingestion can store source-specific metadata in ap_ingestion.source_metadata JSONB column:

This metadata lives on the ingestion (not document) because:

• Different sources provide different metadata
• Same document may come from multiple sources
• Each ingestion preserves its source context

Email Integration (Outlook & Gmail)

Email ingestion allows customers to connect their email inboxes via OAuth. The system automatically extracts invoice attachments from incoming emails.

Supported providers:

• Microsoft Outlook via Microsoft Graph API (primary, implemented first)
• Gmail via Gmail API + Google Cloud Pub/Sub (secondary)

For complete specification, see .prd/email-ingestion.md.

Architecture Overview

┌──────────────┐     ┌──────────────────────┐     ┌─────────────────┐
│ Email        │────▶│ Webhook Handler      │────▶│ pending_sync    │
│ Provider     │     │ POST /webhooks/{prov}│     │ table           │
│ (push notif) │     └──────────────────────┘     └────────┬────────┘
└──────────────┘                                           │
                                                           ▼
┌──────────────┐     ┌──────────────────────┐     ┌─────────────────┐
│ S3           │◀────│ Sync Worker          │◀────│ Claim pending   │
│ + SQS        │     │ (background)         │     │ FOR UPDATE      │
└──────────────┘     └──────────────────────┘     │ SKIP LOCKED     │
       │                                           └─────────────────┘
       ▼
┌──────────────────────────────────────────────────────────────────┐
│            Existing Ingestion Pipeline (SQS → Workers)           │
└──────────────────────────────────────────────────────────────────┘

Provider Abstraction

Both providers implement a common protocol:

(defprotocol EmailProvider
  (create-subscription! [this doc-source-id])
  (renew-subscription! [this doc-source-id])
  (sync-since! [this doc-source-id last-sync-token])
  (fetch-message [this message-id])
  (refresh-token! [this doc-source-id]))

OAuth Flow

1. User clicks "Connect Outlook" or "Connect Gmail"
2. Redirect to provider OAuth consent screen
3. User grants read-only mail access
4. Callback exchanges code for tokens
5. Refresh token stored in AWS SSM (KMS-encrypted)
6. Webhook subscription created with provider

Webhook Processing

Webhooks are processed asynchronously to ensure fast response times:

1. Validate - Check clientState (Outlook) or verification token (Gmail)
2. Lookup - Find doc_source_id by subscription ID or email address
3. Queue - Upsert into ap_doc_source_email_pending_sync table
4. Respond - Return 200 OK immediately

The ap_doc_source_email_pending_sync table uses doc_source_id as primary key, ensuring multiple notifications coalesce into a single sync operation.

Sync Worker

Background worker polls for pending syncs:

;; Claim atomically with SELECT FOR UPDATE SKIP LOCKED
(db/execute!
  "UPDATE ap_doc_source_email_pending_sync
   SET status = 'processing'
   WHERE doc_source_id = (
     SELECT doc_source_id FROM ap_doc_source_email_pending_sync
     WHERE status = 'pending'
     LIMIT 1 FOR UPDATE SKIP LOCKED
   ) RETURNING doc_source_id")

For each claimed sync:

1. Refresh OAuth access token from SSM
2. Call provider delta/history API with stored watermark
3. For each new message, apply heuristic filter
4. Extract attachments from accepted emails
5. Call queue-for-ingestion! for each attachment
6. Update watermark in ap_doc_source_email.last_sync_token

Two-Stage Classification

Emails pass through two relevancy filters:

Stage 1 (Heuristic):

• Accept if: Has PDF/image attachments OR body matches invoice keywords
• Reject if: No attachments AND no invoice signals
• Errs toward accepting (false positives → Stage 2 handles them)

Stage 2 (LLM):

• Runs at fetch-files-from-s3! stage for email-sourced ingestions
• LLM examines email metadata + attachment preview
• Can mark ingestion as skipped with reason (e.g., "marketing_material")

Database Schema

ap_doc_source_email extensions:

connection_status         email_connection_status  -- active, token_refresh_failed, etc.
subscription_expires_at   TIMESTAMPTZ              -- for proactive renewal
provider_subscription_id  TEXT                     -- Outlook subscription ID
last_sync_token          TEXT                     -- deltaLink or historyId
last_successful_sync_at  TIMESTAMPTZ              -- health monitoring

ap_doc_source_email_pending_sync table:

doc_source_id  UUID PRIMARY KEY  -- coalesces notifications
queued_at      TIMESTAMPTZ
attempts       INTEGER           -- retry tracking
last_error     TEXT
status         TEXT              -- 'pending' or 'processing'

ap_ingestion extensions:

status = 'skipped'  -- new enum value
skip_reason TEXT    -- e.g., "marketing_material", "signature_image"

Source Metadata for Email

Email ingestions store metadata in ap_ingestion.source_metadata:

{
  "provider": "outlook",
  "message_id": "<abc123@mail.example.com>",
  "from": "supplier@example.com",
  "subject": "Invoice #12345 - January 2026",
  "received_at": "2026-01-02T10:30:00Z",
  "attachment_filename": "invoice-12345.pdf"
}

This enables:

• Filtering documents by email sender
• Viewing original email context when reviewing
• Linking multiple attachments from same email via message_id

Deduplication

Three-layer deduplication prevents duplicate processing:

Connection Health

Connection status is tracked for monitoring:

CREATE TYPE email_connection_status AS ENUM (
    'active',              -- working normally
    'token_refresh_failed', -- needs re-authorization
    'subscription_expired', -- webhook subscription lapsed
    'provider_error'        -- API returning errors
);

• Proactive subscription renewal at day 5 of 7
• Token refresh failures trigger re-auth flow after 3 attempts
• Health check flags connections with no sync in >24 hours