Story 523 + 524: the document extraction worker

Why this exists (the catastrophe it prevents). A document the firm uploaded and that passed the malware gate is just opaque bytes in S3. Until something reads those bytes into per-page text, the matter intake engine has nothing to build a chronology from, nothing to search, nothing to review — the upload is a dead letter. This worker is the step that turns stored bytes into structured, searchable text. Get it wrong in the obvious ways and you get the two failures that actually hurt: (1) paying a cloud OCR bill on every born-digital PDF that already has a perfectly good text layer, and (2) OCR-ing bytes that were never virus-scanned. This worker is native-first (cheap, deterministic) and scan-gated (it only ever touches documents that already reached stored).

🛰️ Satellite: the whole flow, end to end

upload-complete (#522)            THIS WORKER (#523/#524)
 ─────────────────────            ─────────────────────────────────────────────
 status: stored          SQS      ┌─ native: LiteParse (pypdfium2) ─ text? ─▶ document_pages (engine=pdfium)
 + checksum + mime  ──▶ ingestion │                                              │
 + outbox row           queue ──▶ │  empty (scanned) ─▶ OCR: OcrPort/Textract ─▶ document_pages (engine=textract)
                                  └─────────────────────────────────────────────┘
                                              │
                                              ▼  documents.search_vector = to_tsvector(all page text)
                                              ▼  status ─▶ indexed   (low-confidence page ─▶ needs_review;  error ─▶ failed)
                                              ▼  document_status_events += each transition (continuous trace_id)
                                              ▼  audit_log += document.extracted

The producer side already existed: the FN-OUTBOX relay drains ops.document_ingest_outbox to the ingestion queue with a pointer envelope ({document_id, firm_id, matter_id, object_key, content_type} — never bytes), carrying the W3C trace context. This worker is the consumer that was missing — and, just as importantly, the SQS → Lambda event-source-mapping that connects them (until now there were zero mappings; the merged code was never actually wired to run).

✈️ Airplane: the files

🔬 Microscopic: the decisions that matter

1. Native-first, OCR-fallback — one branch, titled by cost

LiteParse runs first. It asks pypdfium2 (the same PDFium engine browsers render PDFs with, shipped as a self-contained manylinux wheel) for each page's text layer. The branch is one number:

native = LiteParse.extract(bytes, content_type)
if native.stripped_text_length >= NATIVE_TEXT_MIN_CHARS:   # born-digital PDF
    persist_native_result(...)            # engine = pdfium, no OCR bill
else:                                     # scanned PDF / image: ~0 chars
    ocr = OcrPort.recognize(bytes, ...)   # the merged Textract adapter (#1500)
    persist_ocr_result(...)               # engine = textract

A born-digital page returns hundreds–thousands of characters; a scanned page returns ~zero (no text layer). That single threshold is what keeps us from paying Textract for documents that never needed it. The OCR path calls the merged OcrPort adapter from Story-1500 — we do not write a second Textract integration.

2. document_pages is the single home for ALL page text

Both engines write the same table. That is why migration 0115 widens the engine CHECK from ('textract','paddle') to ('pdfium','textract','paddle'): native text has no OCR engine, but it still lives in document_pages. Native rows carry text with no block geometry or confidence (those are OCR concepts); OCR rows carry the neutral 0–1 blocks. Writes are upserts on (firm_id, document_id, page_number) — re-extraction overwrites in place, never duplicates.

3. The state machine lives on documents.processing_status

#522 renamed the legacy 2-value status column to processing_status and widened it to the full ADR-126 machine. The worker advances it directly: stored → extracting → indexed, with needs_review when any OCR page is below its engine's confidence threshold and failed on an unrecoverable, document-specific error. There is no second status axis — one column, one machine, so the document's state can never self-contradict.

4. The hard scan gate (V2): only ever touch stored bytes

GuardDuty malware scanning is asynchronous; #522's gate only enqueues extraction once a document reaches stored. The worker enforces this itself as a hard precondition — a document in any other non-terminal state (still pending/uploaded, not yet scanned) is a no-op the worker re-defers. The worker never extracts or OCRs unscanned bytes.

5. Idempotent & resumable — and no double Textract bill

One record is processed in one database transaction. Two consequences fall out for free:

• At-least-once redelivery of a document already in a terminal state (indexed/needs_review/failed) is acknowledged and skipped — Textract is never re-paid for an already-extracted document.
• A crash mid-extraction commits nothing: the document stays stored, the next delivery re-runs cleanly. There is no partial state to resume and no orphaned page rows, because the page writes, the status transitions, the search vector, and the audit row all commit together or not at all.

6. Oversize is a reject, not a retry

Before fetching a single byte, the worker checks the document size against DOCUMENT_EXTRACTION_MAX_BYTES (default 100 MiB). An oversized document transitions straight to failed with a reason and is acknowledged — it must never be pulled into the worker's bounded memory, and it must never be retried forever.

7. Trace continuity: one trace_id from upload to indexed

The outbox envelope carries the W3C traceparent in its body (not SQS message attributes), so the worker rehydrates it and parents the per-record span to it via context_from_trace_context. Parenting — not linking — is what keeps trace_id continuous: the worker span becomes a child of the API/outbox span, so every document_status_events row the worker writes stamps the same trace_id the upload ran under. An operator can pull one upload's entire journey — request → outbox → relay → SQS → worker → each status transition — under a single id.

API span ──traceparent in outbox body──▶ relay ──SQS──▶ worker
                                                          └ manual_span(DOCUMENT_EXTRACTION_PROCESS_RECORD, context=parent)
                                                              └ get_current_trace_id()  ==  the upload's trace_id
                                                                  └ document_status_events.trace_id

Observability follows ADR-124: two new registry-approved span names (functions.document_extraction.batch CONSUMER, functions.document_extraction.process_record CONSUMER) via manual_span only — no raw tracer calls. The DB and S3 work is covered by T1 auto-instrumentation.

8. RLS for a system actor

The worker runs as the app DB role with no human bound, so audit rows resolve to actor_kind='system'. It sets current_firm_id for firm isolation and current_firm_role='firm_admin' — because ADR-099 record-level RLS gates documents through can_read_matter, which grants firm-wide access only to firm_admin/firm_editor. A system worker must reach every stored document in the firm regardless of any user's matter assignments, so it takes the firm-wide role.

Wiring it for real (the missing link)

Merged code that nothing invokes is not running. The deploy workflow mirrors deploy-ingest-api.yml (ADOT X-Ray wrapper, OTEL env, deps + OTel layers) but adds what a DB- and queue-touching worker needs: in-VPC config for RDS, the runtime DB URL from SSM, a least-privilege IAM role (S3 raw read / derived write, SQS consume, Textract, KMS, logs), and — the actual missing link — an idempotent create-event-source-mapping from pramaan-documents-ingestion-staging to the worker, with ReportBatchItemFailures so a single bad record retries without dragging its batch down.

Verification (system boundary, not component boundary)

The integration tests run the worker against a real RLS-scoped DB on a freshly seeded firm (zero pre-existing rows — the real first run): native PDF → indexed with a pdfium page row and OCR never called; a scanned input → OCR fallback → indexed with a textract row; low-confidence OCR → needs_review (page still succeeded); oversize → failed with bytes never fetched; redelivery of a terminal document → no-op. Each asserts the document_status_events ledger and the document.extracted audit row, and the migration round-trips (downgrade/upgrade) with a clean alembic check and a single head.

One sentence to remember: the worker turns scanned-and-stored bytes into per-page text — native-first to avoid the OCR bill, OCR only as a fallback — advancing one documents.processing_status machine, in one transaction, under one continuous trace_id.