query-orchestration/queue_flow.md

Queue Pipeline Flow Documentation

Overview

This document describes the SQS queue-based processing pipeline that orchestrates document processing through various stages. The system uses AWS SQS queues to connect microservices (runners) in a sequential processing pipeline.

Visual Pipeline Flow

flowchart TB
    %% S3 Event Source
    S3[(S3 Bucket<br/>Document Upload)] -->|S3 Event Notification| SQ1

    %% Main Pipeline Queues and Runners
    SQ1[store_event<br/>Queue] --> R1{storeEventRunner<br/>:8081}
    R1 -->|Creates doc record| SQ2[document_init<br/>Queue]

    SQ2 --> R2{docInitRunner<br/>:8082}
    R2 -->|Check duplicates<br/>by ETag hash| SQ3[document_sync<br/>Queue]

    SQ3 --> R3{docSyncRunner<br/>:8083}
    R3 -->|Validate client<br/>can_sync flag| SQ4[document_clean<br/>Queue]

    SQ4 --> R4{docCleanRunner<br/>:8084}
    R4 -->|Clean per<br/>collector standards| SQ5[document_text<br/>Queue]

    SQ5 --> R5{docTextRunner<br/>:8085}
    R5 -->|AWS Textract<br/>OCR extraction| DB[(PostgreSQL)]

    %% Client Sync Flow
    API1[Query API<br/>:8080] -->|Client update| SQ8[client_sync<br/>Queue]
    SQ8 --> R8{clientSyncRunner<br/>:8089}
    R8 -->|Batch process<br/>all client docs| SQ3

    %% Styling
    classDef queue fill:#ffd700,stroke:#333,stroke-width:2px,color:#000
    classDef runner fill:#87ceeb,stroke:#333,stroke-width:2px,color:#000
    classDef storage fill:#98fb98,stroke:#333,stroke-width:2px,color:#000
    classDef api fill:#ffa07a,stroke:#333,stroke-width:2px,color:#000

    class SQ1,SQ2,SQ3,SQ4,SQ5,SQ8 queue
    class R1,R2,R3,R4,R5,R8 runner
    class S3,DB storage
    class API1 api

Flow Legend

• 🟨 Yellow boxes: SQS Queues
• 🟦 Blue diamonds: Runner Services (with port numbers)
• 🟩 Green cylinder: S3 Storage / PostgreSQL Database
• 🟧 Orange box: REST API endpoint
• Solid arrows: Message flow direction

Detailed Process Flow

sequenceDiagram
    participant S3
    participant SE as store_event Queue
    participant SER as storeEventRunner
    participant DI as document_init Queue
    participant DIR as docInitRunner
    participant DS as document_sync Queue
    participant DSR as docSyncRunner
    participant DC as document_clean Queue
    participant DCR as docCleanRunner
    participant DT as document_text Queue
    participant DTR as docTextRunner
    participant DB as PostgreSQL

    S3->>SE: S3 Event (ObjectCreated:*)
    SE->>SER: Poll message
    SER->>DB: Check if import path
    SER->>DI: Send doc info {bucket, key, hash}

    DI->>DIR: Poll message
    DIR->>DB: Check duplicate by hash
    DIR->>DB: Create document record
    DIR->>DS: Send {documentID}

    DS->>DSR: Poll message
    DSR->>DB: Check client can_sync
    DSR->>DC: Send {documentID} if eligible

    DC->>DCR: Poll message
    DCR->>DB: Check if already clean
    DCR->>DB: Perform cleaning
    DCR->>DT: Send {documentID}

    DT->>DTR: Poll message
    DTR->>DB: Check if text extracted
    DTR-->>DTR: Call AWS Textract
    DTR->>DB: Store extracted text

S3 Event Trigger Configuration

AWS Service Type

S3 Event Notifications (also known as S3 Bucket Notifications)

Configuration Requirements

The pipeline is initiated by S3 Event Notifications configured to send messages directly to an SQS queue:

{
  "QueueConfigurations": [{
    "QueueArn": "arn:aws:sqs:{region}:{account-id}:store_event",
    "Events": ["s3:ObjectCreated:*"],
    "Filter": {
      "Key": {
        "FilterRules": [
          {"Name": "prefix", "Value": "documents/"}
        ]
      }
    }
  }]
}

Important: Files under the batches/ path should be excluded from triggering the pipeline as they are temporary files. This can be accomplished through:

1. Prefix filters in S3 configuration (specify allowed prefixes)
2. Application-level filtering in storeEventRunner (recommended)

AWS CLI Setup Command

aws s3api put-bucket-notification-configuration \
  --bucket {BUCKET_NAME} \
  --notification-configuration '{
    "QueueConfigurations": [{
      "QueueArn": "arn:aws:sqs:{region}:{account-id}:store_event",
      "Events": ["s3:ObjectCreated:*"]
    }]
  }'

Queue Pipeline Flow Tables

Main Document Processing Pipeline

Order	Runner Service	Reads From Queue	Writes To Queue	Description
1	storeEventRunner	store_event	document_init	Receives S3 event notifications when files are uploaded to the bucket
2	docInitRunner	document_init	document_sync	Creates document records, checks for duplicates based on ETag hash
3	docSyncRunner	document_sync	document_clean	Validates client eligibility for document processing (can_sync flag)
4	docCleanRunner	document_clean	document_text	Cleans document content according to collector standards
5	docTextRunner	document_text	-	Extracts text using AWS Textract OCR service, stores in database

Additional Async Runners

These runners are triggered by API calls or other events outside the main pipeline:

Runner Service	Reads From Queue	Writes To Queue	Trigger
clientSyncRunner	client_sync	document_sync	Triggered by API calls or when client configuration changes

Pipeline Flow Diagrams

1. Document Upload Flow (Main Pipeline)

S3 Upload Event
    ↓
[store_event queue]
    ↓
storeEventRunner
    ↓
[document_init queue]
    ↓
docInitRunner
    ↓
[document_sync queue]
    ↓
docSyncRunner
    ↓
[document_clean queue]
    ↓
docCleanRunner
    ↓
[document_text queue]
    ↓
docTextRunner
    ↓
PostgreSQL (text stored)

2. Client Sync Flow (Batch Processing)

Client Update (API)
    ↓
[client_sync queue]
    ↓
clientSyncRunner
    ↓
[document_sync queue] → (feeds into main pipeline)

Queue Names and Environment Variables

Queue Purpose	Queue Name Variable	Queue URL Variable
S3 Events	QNAME_STORE_EVENT	STORE_EVENT_URL
Document Init	QNAME_DOCUMENT_INIT	DOCUMENT_INIT_URL
Document Sync	QNAME_DOCUMENT_SYNC	DOCUMENT_SYNC_URL
Document Clean	QNAME_DOCUMENT_CLEAN	DOCUMENT_CLEAN_URL
Document Text	QNAME_DOCUMENT_TEXT	DOCUMENT_TEXT_URL
Client Sync	QNAME_CLIENT_SYNC	CLIENT_SYNC_URL

Key Design Principles

1. Single Responsibility: Each runner has one specific task in the pipeline
2. Sequential Processing: Documents flow through defined stages in order
3. Idempotent Operations: Duplicate detection prevents reprocessing
4. Async Reprocessing: Client updates trigger document reprocessing
5. Scalability: Each runner can be scaled independently based on workload

IAM Permissions Required

For the S3-to-SQS integration:

• S3 bucket must have permission to send messages to the SQS queue
• SQS queue policy must allow the S3 service principal to send messages
• Each runner service needs permissions to:
  • Receive and delete messages from its input queue
  • Send messages to its output queue(s)
  • Access S3 buckets for document operations
  • Access AWS Textract (for docTextRunner)

Monitoring and Debugging

Each runner exposes Prometheus metrics on port 8080 and includes:

• Queue processing metrics
• Error rates
• Processing duration
• Health check endpoints

Runner ports for local development:

• storeEventRunner: 8081
• docInitRunner: 8082
• docSyncRunner: 8083
• docCleanRunner: 8084
• docTextRunner: 8085
• clientSyncRunner: 8089