CQRS Pattern Implementation

This diagram shows how BlueRobin implements the CQRS (Command Query Responsibility Segregation) pattern, separating write operations from read-optimized queries.

CQRS Overview

flowchart TB
    subgraph Client["Client Layer"]
        User["User Interface
(Blazor)"]
    end

    subgraph Commands["Command Side (Write)"]
        direction TB
        CmdAPI["Archives API
Write Endpoints"]
        CmdHandler["Command Handlers
(FastEndpoints)"]
        Domain["Domain Model
(Aggregates)"]
        WriteDB[(PostgreSQL
Write Model)]
        Events{{"NATS JetStream
Domain Events"}}
    end

    subgraph Queries["Query Side (Read)"]
        direction TB
        QryAPI["Archives API
Read Endpoints"]
        QryHandler["Query Handlers
(FastEndpoints)"]
        ReadDB[(PostgreSQL
Read Views)]
        VectorDB[(Qdrant
Semantic Search)]
        GraphDB[(FalkorDB
Relationships)]
    end

    subgraph Sync["Synchronization"]
        Projections["Event Projections
(Workers)"]
    end

    User -->|Commands| CmdAPI
    User -->|Queries| QryAPI
    
    CmdAPI --> CmdHandler
    CmdHandler --> Domain
    Domain --> WriteDB
    Domain -.->|Publish| Events

    QryAPI --> QryHandler
    QryHandler --> ReadDB
    QryHandler --> VectorDB
    QryHandler --> GraphDB

    Events -.->|Subscribe| Projections
    Projections --> ReadDB
    Projections --> VectorDB
    Projections --> GraphDB

    style Commands fill:#f0dbd8
    style Queries fill:#d5eef0
    style Sync fill:#faf2d0

Command Flow Detail

sequenceDiagram
    autonumber
    participant UI as Blazor UI
    participant API as Archives API
    participant Endpoint as UploadDocumentEndpoint
    participant Service as DocumentService
    participant Domain as Document Aggregate
    participant DB as PostgreSQL
    participant NATS as NATS JetStream

    UI->>API: POST /api/documents
{file, metadata}
    API->>Endpoint: Route to handler
    
    rect rgb(248, 237, 237)
        Note over Endpoint,Domain: Command Processing
        Endpoint->>Service: UploadAsync(request)
        Service->>Domain: Document.Create(...)
        Domain->>Domain: Validate invariants
        Domain->>Domain: RaiseDomainEvent(DocumentAdded)
    end

    rect rgb(237, 245, 246)
        Note over Service,NATS: Persistence & Event Publishing
        Service->>DB: SaveChangesAsync()
        DB-->>Service: Committed
        Service->>NATS: Publish: documents.uploaded
    end

    Service-->>Endpoint: DocumentId
    Endpoint-->>API: 201 Created
    API-->>UI: Response with ID

Query Flow Detail

sequenceDiagram
    autonumber
    participant UI as Blazor UI
    participant API as Archives API
    participant Endpoint as SearchDocumentsEndpoint
    participant Service as RagService
    participant Qdrant as Qdrant
(Read Model)
    participant PG as PostgreSQL
(Metadata)
    participant MinIO as MinIO
(Content)

    UI->>API: POST /api/rag/search
{query, filters}
    API->>Endpoint: Route to handler
    
    rect rgb(237, 245, 246)
        Note over Endpoint,MinIO: Query Processing (No DB Writes)
        Endpoint->>Service: SearchAsync(query)
        Service->>Service: Generate query embedding
        Service->>Qdrant: Vector similarity search
        Qdrant-->>Service: Matching chunk IDs + scores
        Service->>PG: Get document metadata
(read-only view)
        PG-->>Service: Document details
        Service->>MinIO: Fetch content snippets
        MinIO-->>Service: Text content
    end

    Service-->>Endpoint: SearchResults
    Endpoint-->>API: 200 OK
    API-->>UI: Results with snippets

Event Projection Pipeline

flowchart LR
    subgraph Source["Event Source"]
        E1["DocumentAdded"]
        E2["AnalysisCompleted"]
        E3["EmbeddingsCompleted"]
        E4["EntitiesExtracted"]
    end

    subgraph Workers["Projection Workers"]
        P1["DbUpdateEventConsumer
Updates PostgreSQL views"]
        P2["EmbeddingAggregator
Updates Qdrant"]
        P3["GraphSyncConsumer
Updates FalkorDB"]
    end

    subgraph ReadModels["Read Models"]
        RM1[(PostgreSQL
Document views)]
        RM2[(Qdrant
Vector index)]
        RM3[(FalkorDB
Entity graph)]
    end

    E1 & E2 --> P1
    E3 --> P2
    E4 --> P3

    P1 --> RM1
    P2 --> RM2
    P3 --> RM3

    style Source fill:#f0dbd8
    style Workers fill:#faf2d0
    style ReadModels fill:#d5eef0

Read Model Optimization

flowchart TB
    subgraph Queries["Query Types"]
        Q1["List documents
(paginated)"]
        Q2["Full-text search
(keywords)"]
        Q3["Semantic search
(meaning)"]
        Q4["Entity lookup
(relationships)"]
        Q5["RAG query
(Q&A)"]
    end

    subgraph Optimized["Optimized Read Stores"]
        DB1[(PostgreSQL
B-tree indexes
GIN for arrays)]
        DB2[(PostgreSQL
tsvector FTS)]
        DB3[(Qdrant
HNSW index)]
        DB4[(FalkorDB
Graph traversal)]
    end

    Q1 --> DB1
    Q2 --> DB2
    Q3 --> DB3
    Q4 --> DB4
    Q5 --> DB3
    Q5 --> DB1

    style Queries fill:#eee9f5
    style Optimized fill:#edf5f6

CQRS Benefits in BlueRobin

Commands vs Queries

mindmap
  root((CQRS))
    Commands
      POST /api/documents
        Upload file
        Create metadata
        Publish event
      PUT /api/documents/{id}
        Update analysis
        Update status
      DELETE /api/documents/{id}
        Mark deleted
        Publish event
    Queries
      GET /api/documents
        List with filters
        Paginated results
      POST /api/rag/search
        Vector similarity
        Score ranking
      POST /api/rag/ask
        Context retrieval
        LLM generation
      GET /api/graph/entities
        Graph traversal
        Relationship lookup

Consistency Model