Data Ingestion — Email & Documents
| Owner | Classification | Version | Effective | Next review | Status |
|---|---|---|---|---|---|
| Platform Engineering | Internal | 0.1 | 2026-08-29 | 2026-11-29 | Draft |
Purpose. Reference for the platform's inbound email and document ingestion pipeline — components, message flow, connector configuration, and the procedure to operate it. The pipeline is shared: every module (cash application, invoice, journal, close hub) ingests through it; only the classification target and downstream extraction differ.
Pipeline
email arrives
→ IMAP mailbox (e.g. Gmail)
→ emailprocessor (apps/emailprocessor — IMAP poller; must be running)
→ processmail (workflow, packages/system/systemwf)
→ documentreceipt (row created, status Queued; attachments → object storage)
→ Mail Processing Rules (procrs_5) → { documenttype, subcategory }
⏱ document-processqueue (Temporal SCHEDULE, ~1 min — the drainer)
→ processqueue (picks up status=Queued receipts) → rundocprocess
→ rundocprocess (preprocessing rules → extract via Gemini → status Extracted)
→ PROMOTE — routed by documenttype:
remittance → processremittance → runremittancetransformations
invoice → processinvoice → runinvoicetransformations
journal → processjournal (dispatched to the journal queue)
→ domain record (cash.remittanceadvice / invoice / journal voucher)
→ transform → validate → assign → approve
→ Smart Link / matching (per module)The link between "email received" and "record created" is a scheduled drainer, not a direct call.
processmailonly creates thedocumentreceiptin statusQueued(orAssignedif unclassified) and stops. A separate Temporal schedule (document-processqueue, ~1 min) runsprocessqueue, which selectsQueuedreceipts and dispatchesrundocprocess→ extraction → promotion for each. So ingestion has two independently-running pieces: the poller (emailprocessor) and the drainer schedule. If either is absent, documents pile up silently — emails un-fetched, or receipts stranded atQueued. See the reproducibility caveat under Operational constraints.
Components
| Component | Location | Responsibility |
|---|---|---|
emailprocessor | apps/emailprocessor/src | Polls each active mailbox over IMAP (ImapFlow), copies attachments to object storage, and starts the mailbox's target workflow. A standalone service, separate from the API and the workflow worker. |
sys.inbox | database table | One row per mailbox connector — IMAP credentials, folder, poll interval, active flag, and the target workflow. |
processmail | packages/system/systemwf/src/workflows/processmail | Converts any inbound email into a documentreceipt, moves attachments, and hands the record to classification. |
| Mail Processing Rules | coresystem.processruleset id procrs_5 (rules in coresystem.processrule) | Classifies the email into { documenttype, subcategory }. |
document-processqueue | Temporal schedule (~1 min) → processqueue | The drainer. A recurring schedule that runs processqueue. Not backed by code or finbase.taskschedule — created directly in Temporal (see reproducibility caveat). Drives all doc types. |
processqueue | packages/system/systemwf/src/workflows/processqueue | Selects documentreceipt rows with status = 'Queued' and dispatches rundocprocess for each. |
rundocprocess | packages/system/systemwf/src/workflows/rundocprocess | Per receipt: runs preprocessing rules, resolves the extraction engine (Gemini) from application settings, extracts (→ status Extracted), then dispatches the promote workflow by documenttype (see below). Requires the receipt in Queued state; an unsupported documenttype errors. |
Promote workflow (routed by documenttype) | packages/system/systemwf (processremittance, processinvoice) · packages/finance/journal/journalwf (processjournal, dispatched by name to the journal queue) | remittance → processremittance, invoice → processinvoice, journal → processjournal. Promotes the extracted receipt into its domain record (cash.remittanceadvice / invoice / journal voucher), applying internal transforms (field coercion / promotion). |
| Post-promote chain (per type) | runremittancetransformations / runinvoicetransformations / journal equivalent | script transforms → validate (→ *validationchecks) → assign (BPD rules) → approve. |
Classification
The procrs_5 ruleset resolves the document type by precedence (sequencenumber):
| Seq | Rule | Signal |
|---|---|---|
| 1–3 | Mailbox name | inbox name contains remittance / invoice / journal |
| 10–13 | Subject or attachment | keywords in the subject line or attachment filename |
| 20–21 | Body | keywords in the email body |
| 99 | Default | catch-all → unclassified Email |
Mailbox-name classification is the most deterministic route, which is why dedicated inboxes exist per document type. An email reaches the correct lane when it either arrives on a purpose-named inbox or carries the matching keywords in its subject, attachment name, or body. The ruleset ships active and covers remittance, invoice, and journal without modification.
Connector configuration
Connectors are rows in sys.inbox:
| Column | Meaning |
|---|---|
id | connector name; its text is matched by the mailbox-name classification rule |
protocol | IMAP |
host / port / secure / starttls | IMAP endpoint (e.g. imap.gmail.com / 993 / t / f) |
username / password | mailbox login; for Gmail, password is an app password (~16–19 chars), not the account password |
folder | mailbox folder, usually INBOX |
syncinterval | poll interval in seconds |
isactive | true to poll the connector |
targetconfiguration | JSON: {"type":"workflow","workflow":{"id":"processmail"}} |
customsearch | optional IMAP search filter (e.g. unread-only) |
Configured connectors (as shipped):
| id | account | active | target |
|---|---|---|---|
cashapp-remittance-in | finaisse.test@gmail.com | ⚪ | processmail |
cashapp-payments-in | placeholder | ⚪ | processmail |
finaisse-in | finaisse.test@gmail.com | 🟢 | processmail |
fininvoicehub-in | fininvoicehub@gmail.com | 🟢 | processmail |
collections-in | fincolllectionhub@gmail.com | 🟢 | processcollectionemail |
closehub-in | placeholder | ⚪ | processmail |
Operating
Start the poller (from blitz/src):
bun run --cwd apps/emailprocessor devActivate the cash-application remittance connector (credentials are already set):
UPDATE sys.inbox SET isactive = true WHERE id = 'cashapp-remittance-in';
-- Prevent double-processing: it shares the Gmail account with finaisse-in.
UPDATE sys.inbox SET isactive = false WHERE id = 'finaisse-in';Send a test email to the mailbox account with a document attachment, then verify:
-- documentreceipt created and classified
SELECT id, documenttype, subcategory, status, filenames
FROM <schema>.documentreceipt ORDER BY id DESC LIMIT 5;
-- remittance extracted
SELECT id, customername FROM cash.remittanceadvice ORDER BY id DESC LIMIT 5;Operational constraints
- The poller (
emailprocessor) is a separate service. When it is not running, no mailbox is polled and no email is ingested — the pipeline is silent, with no error. - Two active connectors on the same mailbox account and folder both fetch the same message, producing duplicate
documentreceiptrecords. Exactly one connector must be active per mailbox account. - A fully-configured connector remains inactive until
isactive = true. - Gmail connectors require an app password and IMAP enabled on the account.
- Classification is name-first: deterministic routing requires the document-type keyword in the inbox name; otherwise the subject, attachment name, or body must carry it.
- Extraction and matching are distinct steps. Re-running extraction does not refresh matching, and vice versa.
- The
document-processqueuedrainer schedule is not reproducible. It exists only as a Temporal schedule created by hand — there is no code that defines or ensures it, and nofinbase.taskschedulerow (that table is for automation schedules, not system plumbing). A fresh environment (new machine, Railway, restored dump) therefore has no drainer → every ingesteddocumentreceiptstalls atQueuedand no domain record is ever created, silently. Provisioning it belongs in a platform deploy/bootstrap step (like a migration); until that exists, the schedule must be (re)created by hand per environment. processqueueonly selectsQueuedreceipts. A receipt that reachesExtractedbut fails to promote is stranded — the drainer will not re-pick it (noExtracted-orphan recovery today).
Deployment (Railway)
The pipeline is service- and data-driven; a database restore alone is insufficient.
| Requirement | Status |
|---|---|
emailprocessor deployed as a running service | ⚪ verify |
sys.inbox rows present and passwords set (may be env-injected on restore) | ⚪ verify |
procrs_5 Mail Processing Rules present (ships with the dump) | ⚪ verify |
document-processqueue Temporal schedule created (not in the dump — must be provisioned per env) | 🔴 gap — no code path yet |
| IMAP egress and Gmail app password reachable from Railway | ⚪ verify |
| Exactly one active connector per mailbox account | ⚪ verify |
Related
- Backend architecture: Backend Services
- Platform operations: Local Development
Revision history
| Version | Date | Author | Change |
|---|---|---|---|
| 0.1 | 2026-08-29 | Platform Engineering | Initial draft — pipeline, components, classification, connector configuration, operating procedure, operational constraints, Railway checklist. |
| 0.2 | 2026-09-04 | Platform Engineering | Corrected the pipeline: the receipt→record link is the scheduled document-processqueue drainer (processqueue → rundocprocess → promote), not a direct call. Added drainer/processqueue/rundocprocess components, the transform→validate→assign→approve tail, the schedule-reproducibility gap, and the Extracted-orphan constraint. |