Document Ingestion Pipeline

This flowchart illustrates the complete document processing pipeline in BlueRobin, from initial upload through OCR, analysis, embedding, and indexing.

Pipeline Overview

flowchart TB
    subgraph Upload["📤 Upload Phase"]
        A[User uploads document] --> B{Validate file}
        B -->|Invalid| B1[Return error]
        B -->|Valid| C[Calculate fingerprint
SHA256]
        C --> D{Duplicate check}
        D -->|Exists| D1[Return existing doc]
        D -->|New| E[Store in MinIO
bucket: env-userId]
        E --> F[Create Document entity
Status: Pending]
        F --> G[Publish: documents.uploaded]
    end

    subgraph OCR["🔍 OCR Phase"]
        G --> H[OcrEventConsumer
subscribes]
        H --> I[Send to Docling OCR
service]
        I --> J{Extraction
successful?}
        J -->|No| J1[Retry with backoff
max 3 attempts]
        J1 -->|Failed| J2[Mark: Failed
Publish: documents.failed]
        J -->|Yes| K[Store text in MinIO
processed/docId/content.md]
        K --> L[Update document
Status: Processing]
        L --> M[Publish: documents.ocr.completed]
    end

    subgraph Analysis["🧠 Content Analysis Phase"]
        M --> N[AnalysisEventConsumer
subscribes]
        N --> O[Generate via Ollama LLM]
        
        subgraph AnalysisTasks["Parallel Analysis"]
            O --> O1[Summary
extraction]
            O --> O2[Keywords
extraction]
            O --> O3[Friendly name
generation]
        end
        
        O1 & O2 & O3 --> P[Update DocumentAnalysis
value object]
        P --> Q[Publish: documents.analysis.completed]
    end

    subgraph Chunking["✂️ Chunking Phase"]
        Q --> R[DocumentChunkingWorker
subscribes]
        R --> S[Semantic chunking
max 512 tokens]
        S --> T[For each chunk:
Publish chunk.created]
    end

    subgraph Embedding["🔢 Embedding Phase"]
        T --> U[EmbeddingFanoutWorker
subscribes]
        U --> V[Fan out to 8 models]
        
        subgraph Models["Parallel Embedding Models"]
            V --> V1[nomic-embed-text]
            V --> V2[mxbai-embed-large]
            V --> V3[snowflake-arctic]
            V --> V4[bge-large-en]
            V --> V5[granite-embedding]
            V --> V6[bge-m3]
            V --> V7[all-minilm]
            V --> V8[paraphrase-multilingual]
        end
        
        V1 & V2 & V3 & V4 & V5 & V6 & V7 & V8 --> W[ChunkEmbeddingWorker
per model]
        W --> X[Generate vectors
via Ollama]
        X --> Y[Store in Qdrant
collection per model]
        Y --> Z[Publish: embeddings.model.completed]
    end

    subgraph Aggregation["📊 Aggregation Phase"]
        Z --> AA[EmbeddingAggregatorWorker
subscribes]
        AA --> AB{All 8 models
completed?}
        AB -->|No| AC[Wait for remaining]
        AB -->|Yes| AD[Publish: documents.embeddings.completed]
    end

    subgraph EntityExtraction["🏷️ Entity Extraction Phase"]
        AD --> AE[EntityExtractionConsumer
subscribes]
        AE --> AF[Send to Spacy NER
service]
        AF --> AG[Extract entities:
PERSON, ORG, DATE, LOC]
        AG --> AH[Store in PostgreSQL
canonical_entities table]
        AH --> AI[Publish: documents.entities.extracted]
    end

    subgraph GraphSync["🕸️ Graph Sync Phase"]
        AI --> AJ[GraphSyncConsumer
subscribes]
        AJ --> AK[Sync to FalkorDB
knowledge graph]
        AK --> AL[Create relationships
between entities]
        AL --> AM[Publish: documents.graph.synced]
    end

    subgraph Classification["📁 Classification Phase"]
        AM --> AN[DocumentClassificationWorker
subscribes]
        AN --> AO[Classify via LLM
into categories]
        AO --> AP[Update DocumentClassification
value object]
        AP --> AQ[Mark: Completed
Publish: documents.classified]
    end

    subgraph Notification["🔔 Notification Phase"]
        AQ --> AR[NatsDocumentEventListener
in Blazor Web]
        AR --> AS[Update UI in real-time
via component state]
        AS --> AT[✅ Document ready
for search]
    end

    style Upload fill:#eee9f5
    style OCR fill:#fdf8ea
    style Analysis fill:#ddd4ed
    style Chunking fill:#edf5f6
    style Embedding fill:#f8eded
    style Aggregation fill:#d5eef0
    style EntityExtraction fill:#faf2d0
    style GraphSync fill:#edf5f6
    style Classification fill:#eee9f5
    style Notification fill:#d5eef0

Processing States

stateDiagram-v2
    [*] --> Pending: Document Created
    Pending --> Processing: OCR Started
    Processing --> Processing: Analysis/Embedding
    Processing --> Completed: All Steps Done
    Processing --> Failed: Error Occurred
    Failed --> Processing: Retry
    Completed --> [*]

NATS Event Flow

Error Handling

• Retry Policy: Exponential backoff with max 3 retries
• Dead Letter Queue: Failed messages after retries
• Idempotency: Each worker checks if step already completed
• Compensation: Failed status allows manual retry via UI