Current Document Processing State

This page documents the current document ingestion paths while the unified document processing pipeline is being rolled out.

The target architecture keeps the original document metadata in documents, the extracted text in document_texts, and the automatic processing state plus parser outputs in document_processing_runs.

Entry points

Manual upload

Owner service: docs-service.

Current flow:

1. POST /docs/upload receives a multipart file.
2. The route requires entity_type and entity_id.
3. The route creates or reuses a documents row.
4. The route uploads the binary to the configured storage provider.
5. The route creates a document_versions row.
6. The route updates the current storage metadata on documents.
7. The route calls extractAndStoreDocumentText.
8. Extracted text is persisted into document_texts.

Relevant files:

• backend/docs-service/routes/docs.js
• backend/docs-service/modules/text/extractAndStoreDocumentText.js
• backend/docs-service/modules/text/documentTextRepository.js

Current canonical writes:

• documents
• document_versions
• document_texts

Email ingestion

Owner service: mail-ingestion-service, with file persistence delegated to docs-service.

Current flow:

1. Mail rules resolve attachments or downloadable links from a message.
2. Actions such as INDEX_DOCUMENT, utility bill parsing, and condominium installment parsing upload the file through docs-service /docs/upload.
3. Because upload goes through docs-service, the file is saved in documents, document_versions, and document_texts.
4. Mail-specific metadata is saved in mail_message_documents.
5. The upload endpoint triggers a document_processing_runs pipeline run.
6. Parsing/classification results are written to document_processing_runs, not materialized directly by mail-ingestion-service.

Relevant files:

• backend/mail-ingestion-service/modules/rules/actions/indexDocument.action.js
• backend/mail-ingestion-service/modules/attachments/utilityBillAttachmentProcessor.js
• backend/mail-ingestion-service/modules/rules/actions/parseCondominiumInstallments.action.js

Current canonical writes:

• documents, through docs-service
• document_versions, through docs-service
• document_texts, through docs-service
• document_processing_runs, through document-indexer-service
• mail_message_documents

Mail-specific state kept outside the pipeline:

• mail_messages.classification_json
• mail_message_documents.classification_json, only for source/link metadata and parser hints such as service_contract_id or target_person
• mail_message_documents.detected_document_type, as a compatibility hint
• mail_message_documents.processing_status, as mail-rule execution status

Cloud sync

Owner service: storage-connector-service, with indexing delegated to document-indexer-service.

Current flow:

1. storage-connector-service detects provider changes.
2. The connector records the provider file metadata and source reference.
3. The connector calls document-indexer-service /index/storage-file.
4. The indexer queues the file as a low-priority storage job.
5. indexStorageFile creates or updates a documents row by source_connection_id and source_file_id.
6. The indexer creates a document processing run and publishes the pipeline event.
7. The text extraction consumer calls docs-service /docs/:id/ensure-text.
8. docs-service retrieves cloud bytes on demand through storage-connector-service /connections/:id/files/content.
9. docs-service persists extracted text in document_texts.
10. Later pipeline steps classify, parse, index, and update document status.

Relevant files:

• backend/storage-connector-service/modules/sync/syncEngine.js
• backend/storage-connector-service/modules/downstream/indexingClient.js
• backend/document-indexer-service/routes/indexingRoutes.js
• backend/document-indexer-service/modules/indexing/indexStorageFile.js

Current canonical writes:

• storage_synced_files
• documents
• document_processing_runs
• document_texts, through docs-service /docs/:id/ensure-text
• vector payloads in Qdrant

Current constraint:

• Cloud-origin files are not copied into rAInty storage during sync. The source file must still be reachable from the provider when the text extraction consumer runs.

Known duplicated automatic state:

• storage_synced_files.status
• storage_synced_files.error_message
• storage_synced_files.metadata_json
• documents.indexing_status
• documents.indexing_error

Existing storage model

documents already supports source tracking:

• source_type
• source_connection_id
• source_file_id

This means cloud-origin documents should not need a parallel business document table. The same documents row can represent both rAInty-stored files and external cloud files, with source fields identifying the origin.

document_texts already supports extracted text layers:

• PDF_TEXT
• OCR_GOOGLE
• LLM_RAG_TEXT
• MANUAL

This table should remain the canonical place for OCR/text extraction output.

Pipeline state target

document_processing_runs becomes the canonical table for automatic processing state and parser outputs:

• pipeline status
• current step
• detected document type
• classification confidence
• classification JSON
• structured extraction JSON
• failure reason
• attempt count

Source-specific tables may keep source tracking data, but should not become the canonical place for document classification or parser results.

Source tracking versus pipeline result

Source tables must remain focused on origin tracking.

storage_synced_files keeps cloud sync metadata:

• connection_id
• folder_id
• provider_file_id
• provider filename/path/checksum/modified timestamp
• sync status
• document_id

It must not store pipeline state such as processing run ids, parser output, or classification JSON.

mail_message_documents keeps mail attachment linkage metadata:

• mail_message_id
• document_id
• provider_attachment_id
• original_file_name
• source URL or similar attachment metadata when needed for deduplication

It must not store parsed document JSON or automatic pipeline output.

Legacy columns such as classification_json, detected_document_type, and processing_status may still exist for backward compatibility and idempotency, but the canonical automatic result is document_processing_runs.

Pipeline statuses:

Status	Meaning
REQUESTED	A processing run was created but no step has started yet.
TEXT_EXTRACTION_PENDING	Text extraction or OCR has to run.
TEXT_READY	Text is available in document_texts.
CLASSIFICATION_PENDING	Document classification has to run.
CLASSIFIED	Document type and classification metadata are available.
PARSING_PENDING	Structured extraction has to run.
PARSED	Structured extraction JSON is available.
INDEXING_PENDING	Vector indexing has to run.
INDEXED	Vector indexing completed.
NEEDS_REVIEW	User review is required.
FAILED	Processing failed and the failure is persisted.
APPROVED	User approved the classified/parsed result.
MATERIALIZED	Business records were created or linked from the approved output.

Redis choreography convention

Redis infrastructure already exists in backend/shared/redisBus.js.

Use Redis Streams for guaranteed pipeline advancement:

• RedisBus.xaddJson
• RedisBus.consumeLoop
• consumer groups
• xAck only after the DB state update succeeds

Use Pub/Sub only for non-critical notifications and UI/admin observability.

The shared naming convention is defined in:

• backend/shared/documentProcessingEvents.js

Command and control API

document-indexer-service exposes command/control endpoints. These endpoints update document_processing_runs and publish Redis stream events. They do not execute the whole pipeline synchronously.

Method	Path	Purpose
POST	/documents/:id/process	Create a new processing run and publish document.processing.requested.
POST	/documents/:id/retry	Retry the latest failed run, or a specific failed run when runId is passed.
POST	/documents/:id/reset	Reset the latest run to REQUESTED, or create one if missing.
GET	/documents/:id/processing-status	Return the latest processing run for a document.
PATCH	/documents/:id/processing-result	Update the latest processing run with classification/parser output.
GET	/processing-runs	List processing runs, optionally filtered by document_id and status.
GET	/processing-runs/:id	Return one processing run.
PATCH	/processing-runs/:id/result	Update one processing run with classification/parser output.

docs-service owns text extraction and OCR through:

Method	Path	Purpose
POST	/docs/:id/ensure-text	Return existing valid document_texts or extract/OCR and persist text.

For cloud-origin files that are not yet copied into rAInty storage, ensure-text retrieves the file through storage-connector-service using the document source reference (source_connection_id and source_file_id). Inline contentBase64 remains supported only for legacy/manual callers; cloud sync does not use it.

Entry point handoff

The three document ingress paths now hand off to the same pipeline state:

• Manual upload: docs-service /docs/upload stores the document, ensures text, then calls document-indexer-service /documents/:id/process.
• Email: mail actions upload through docs-service /docs/upload with source_type=EMAIL; the upload endpoint triggers the pipeline.
• Cloud sync: storage-connector-service sends only source metadata to document-indexer-service /index/storage-file; that worker creates or links the documents row and creates a pipeline run. The text extraction consumer later retrieves the provider file on demand through docs-service.

Guaranteed stream:

<env>:document:processing

Events:

• document.processing.requested
• document.text.extraction.requested
• document.text.ready
• document.classification.requested
• document.classified
• document.parsing.requested
• document.parsed
• document.indexing.requested
• document.indexed
• document.review.required
• document.materialization.requested
• document.materialized
• document.processing.failed
• document.processing.completed

Each choreography step must:

1. read document_processing_runs;
2. verify the expected state;
3. skip idempotently when the state is no longer valid;
4. do its own work;
5. update status, current_step, timestamps, and output JSON;
6. publish the next event through RedisBus.xaddJson;
7. persist FAILED with error_message and attempt_count on error.

Implemented consumers:

Service	Consumer group	Events handled
docs-service	document-processing:text-extraction	document.text.extraction.requested
document-indexer-service	document-processing:pipeline-router	document.processing.requested, document.text.ready, document.classification.requested, document.classified, document.parsing.requested, document.parsed, document.indexing.requested

The current classification worker is deterministic and conservative. It uses the existing documents.document_type metadata when available, otherwise it classifies the document as OTHER. The parser step currently handles UTILITY_BILL and CONDOMINIUM_INSTALLMENTS through llm-gateway, using persisted document_texts as input. Other document types continue directly to indexing or review.

Failure handling:

• the worker updates the run to FAILED;
• failed_step stores the failing step name;
• error_message stores the error;
• attempt_count is incremented;
• document.processing.failed is published for observability.

Redis is only the advancement mechanism. The canonical state remains in document_processing_runs, so events can be skipped idempotently when a run has already advanced or has been reset/retried.