query-orchestration/docs/ai.generated/service.details.md

# Service Processing Details

This document provides detailed information about what each service in the document processing pipeline actually does.

## Document Processing Pipeline Services

### 1. storeEventRunner
**Purpose**: Entry point for document processing, triggered by S3 upload events

**Code Locations**:
- `cmd/storeEventRunner/` - Service entry point and main.go
- `api/storeEventRunner/` - Runner implementation and message handling
- `internal/document/store/` - Core document storage logic
- `internal/serviceconfig/queue/documentinit/` - Queue configuration for next stage

**Processing Logic**:
- Receives S3 event notifications (ObjectCreated:Put, ObjectCreated:CompleteMultipartUpload)
- Validates that the event type is supported
- Parses the S3 object key to extract structured information (client ID, location, timestamps)
- Filters for objects in the "import" location only
- Extracts document metadata (bucket, key, hash/ETag)
- Sends message to DOCINIT queue with document location information

**Key Validations**:
- Only processes supported S3 events (Put and CompleteMultipartUpload)
- Only processes objects with "import" location in their key path
- Validates key can be parsed into expected structure

**Output**: Sends to DOCINIT queue with {bucket, key, hash}

---

### 2. docInitRunner
**Purpose**: Document registration and deduplication

**Code Locations**:
- `cmd/docInitRunner/` - Service entry point and main.go
- `api/docInitRunner/` - Runner implementation and message handling
- `internal/document/init/` - Core document initialization logic
- `internal/serviceconfig/queue/documentsync/` - Queue configuration for next stage

**Processing Logic**:
- Parses the S3 key to extract client ID, batch ID, and other metadata
- Checks if document already exists by looking up hash in database for the client
- If document exists (duplicate):
  - Adds new document entry to existing document record
  - Does NOT trigger downstream processing (deduplication)
- If new document:
  - Creates new document record in database with client ID and hash
  - Associates batch ID if present
  - Adds document entry with S3 location details
  - Sends message to DOCSYNC queue to continue processing

**Key Features**:
- Hash-based deduplication per client
- Maintains multiple entries for same document (different uploads)
- Transaction-based database operations for consistency

**Output**: Sends to DOCSYNC queue with {document_id} (only for new documents)

---

### 3. docSyncRunner
**Purpose**: Client synchronization validation

**Code Locations**:
- `cmd/docSyncRunner/` - Service entry point and main.go
- `api/docSyncRunner/` - Runner implementation and message handling
- `internal/document/sync/` - Core document synchronization logic
- `internal/client/` - Client management utilities
- `internal/document/` - Shared document utilities
- `internal/serviceconfig/queue/documentclean/` - Queue configuration for next stage

**Processing Logic**:
- Retrieves document summary from database
- Fetches client configuration
- Checks if client has "CanSync" flag enabled
- If CanSync is true:
  - Sends document to DOCCLEAN queue for processing
- If CanSync is false:
  - Logs that document won't be synced
  - Processing stops here for this document

**Key Validations**:
- Client must exist in database
- Client must have CanSync=true to proceed

**Output**: Sends to DOCCLEAN queue with {document_id} (only if client can sync)

---

### 4. docCleanRunner
**Purpose**: Document validation and preparation

**Code Locations**:
- `cmd/docCleanRunner/` - Service entry point and main.go
- `api/docCleanRunner/` - Runner implementation and message handling
- `internal/document/clean/` - Core document cleaning and validation logic
- `internal/document/types/` - Document type definitions and MIME type handling
- `internal/serviceconfig/queue/documenttext/` - Queue configuration for next stage

**Processing Logic**:
- Checks if document has already been cleaned (idempotency)
- If not cleaned:
  - Retrieves document entry from database
  - Makes HEAD request to S3 to get content metadata
  - Validates content type/MIME type is acceptable (PDF)
  - Downloads document from S3
  - Validates document integrity (not corrupt)
  - Stores clean record with:
    - Success: bucket, key, hash, mimetype
    - Failure: reason for failure (invalid MIME, corrupt file)
- Creates versioned clean entry for tracking
- Sends to DOCTEXT queue regardless of success/failure

**Key Validations**:
- MIME type validation (must be acceptable format)
- Document corruption check
- Hash verification against S3 ETag

**Output**: Sends to DOCTEXT queue with {document_id}

---

### 5. docTextRunner
**Purpose**: Text extraction from documents using AWS Textract

**Code Locations**:
- `cmd/docTextRunner/` - Service entry point and main.go
- `api/docTextRunner/` - Runner implementation and message handling
- `internal/document/text/` - Core text extraction logic using AWS Textract
- `internal/document/types/` - Document type utilities for text processing
- `internal/serviceconfig/textract/` - AWS Textract client configuration
- `internal/serviceconfig/queue/querysync/` - Queue configuration for next stage

Tests for this use mocks only. No tests to the actual aws textract service. 

**Processing Logic**:
- Checks if text has already been extracted (idempotency)
- If not extracted:
  - Retrieves clean document record
  - Skips if document failed cleaning
  - Downloads document from S3
  - Determines page count
  - For each page (in parallel using thread pool):
    - Sends page to AWS Textract for OCR
    - Processes base text blocks
    - Identifies required features (tables, forms)
    - Performs additional extraction if needed
    - Merges text with page markers
  - Stores extracted text in S3
  - Records text extraction in database with versioning
- Sends to QUERYSYNC queue for query processing

**Key Features**:
- Parallel page processing for performance
- Supports tables and forms extraction
- Stores full text with page boundaries preserved
- Version tracking for text extractions

**Output**: Sends to QUERYSYNC queue with {document_id}

---

### 6. querySyncRunner
**Purpose**: Query dependency resolution and scheduling

**Code Locations**:
- `cmd/querySyncRunner/` - Service entry point and main.go
- `api/querySyncRunner/` - Runner implementation and message handling
- `internal/query/sync/` - Core query synchronization logic
- `internal/query/result/sync/` - Result synchronization utilities
- `internal/serviceconfig/queue/query/` - Queue configuration for next stage

**Processing Logic**:
- Verifies document has extracted text
- Queries database for unsynced queries that:
  - Are configured for this client (via collector)
  - Have no dependencies OR
  - Have all dependencies already satisfied
- For each eligible query:
  - Sends message to QUERY queue for execution

**Key Features**:
- Dependency graph resolution
- Only schedules queries with satisfied dependencies
- Client-specific query configuration via collectors

**Output**: Sends to QUERY queue with {document_id, query_id} for each eligible query

---

### 7. queryRunner
**Purpose**: Query execution and result storage

**Code Locations**:
- `cmd/queryRunner/` - Service entry point and main.go
- `api/queryRunner/` - Runner implementation and message handling
- `internal/query/result/set/` - Core query result processing
- `internal/query/result/` - Query execution logic and processors
- `internal/query/result/processor/` - Base processor interface and utilities
- `internal/query/types/contextFull/` - CONTEXT_FULL query type implementation
- `internal/query/types/jsonExtractor/` - JSON_EXTRACTOR query type implementation
- `internal/query/` - Query management utilities
- `internal/query/result/sync/` - Result synchronization for triggering dependents

**Processing Logic**:
- Retrieves document text version information
- Fetches query definition and active version
- If query has dependencies:
  - Retrieves results from required queries
  - Validates all dependencies are satisfied
- Executes query based on type:
  - **CONTEXT_FULL**: Returns complete document text
  - **JSON_EXTRACTOR**: Extracts specific JSON paths from dependent results
- Stores result in database with:
  - Query result value
  - Query version used
  - Text version processed
  - Dependencies tracked
- Identifies queries that depend on this result
- Sends dependent queries to QUERY queue (recursive processing)

**Key Features**:
- Supports multiple query types with different processors
- Dependency tracking and validation
- Recursive processing for dependent queries
- Version tracking for reproducibility

**Output**: Sends to QUERY queue with {document_id, query_id} for each dependent query

---

## Additional Synchronization Services

### 8. clientSyncRunner
**Purpose**: Client-level document synchronization

**Code Locations**:
- `cmd/clientSyncRunner/` - Service entry point and main.go
- `api/clientSyncRunner/` - Runner implementation and message handling
- `internal/client/sync/` - Core client synchronization logic
- `internal/serviceconfig/queue/documentsync/` - Queue configuration for document sync

**Processing Logic**:
- Processes CLIENTSYNC queue messages
- Retrieves all documents for a specific client
- Sends each document to DOCSYNC queue for processing
- Enables bulk reprocessing of client documents

**Output**: Sends to DOCSYNC queue with {document_id} for each client document

---

### 9. queryVersionSyncRunner  
**Purpose**: Query version change propagation

**Code Locations**:
- `cmd/queryVersionSyncRunner/` - Service entry point and main.go
- `api/queryVersionSyncRunner/` - Runner implementation and message handling
- `internal/query/versionsync/` - Core query version synchronization logic
- `internal/serviceconfig/queue/clientsync/` - Queue configuration for client sync

**Processing Logic**:
- Processes QUERYVERSIONSYNC queue messages triggered by query updates
- Identifies all clients affected by query version changes
- Sends affected clients to CLIENTSYNC queue for reprocessing
- Enables cascading updates when query logic changes

**Output**: Sends to CLIENTSYNC queue with {client_id} for each affected client

---

## Data Flow Summary

### Primary Document Processing Pipeline
```
S3 Upload Event
    ↓
storeEventRunner (validates event, extracts metadata)
    ↓
docInitRunner (deduplication, document registration)
    ↓
docSyncRunner (client sync validation)
    ↓
docCleanRunner (document validation, MIME check)
    ↓
docTextRunner (OCR via AWS Textract)
    ↓
querySyncRunner (find executable queries)
    ↓
queryRunner (execute queries, trigger dependents)
    ↓
queryRunner (recursive for dependent queries)
```

### Synchronization Flows
```
Query Version Update
    ↓
queryVersionSyncRunner (find affected clients)
    ↓
clientSyncRunner (reprocess all client documents)
    ↓
docSyncRunner → ... (normal processing pipeline)
```

## Key Design Patterns

### Idempotency
- Each service checks if its work has already been done
- Prevents duplicate processing on retries
- Version tracking ensures consistency

### Dependency Management
- Queries can depend on other query results
- Dependency graph is resolved before execution
- Results are chained through recursive processing

### Version Tracking
- Document clean versions
- Text extraction versions
- Query versions
- Enables reprocessing when configurations change

### Error Handling
- Failed documents are marked but continue through pipeline
- Allows partial processing and debugging
- Failure reasons are stored for analysis

### Client Isolation
- All processing is scoped to client ID
- Deduplication is per-client
- Query configurations are per-client via collectors

---

## Shared Code Directories

These directories contain utilities and configurations used across multiple services:

### Core Infrastructure
- `internal/database/` - Database queries, migrations, and repository patterns
- `internal/serviceconfig/` - Configuration management for all services
- `internal/serviceconfig/queue/` - SQS queue client configurations
- `internal/serviceconfig/objectstore/` - S3 client configurations
- `internal/serviceconfig/aws/` - AWS service configurations
- `internal/server/runner/` - Base runner framework for queue-based services

### Business Logic Libraries
- `internal/document/` - General document utilities (shared across document services)
- `internal/client/` - Client management utilities
- `internal/query/` - Query management and utilities (shared across query services)

### Testing & Utilities
- `internal/test/` - Test utilities and helpers
- `internal/validation/` - Input validation utilities
- `mocks/` - Auto-generated mock interfaces for testing

### API Layer
- `api/queryAPI/` - REST API handlers and controllers (not a queue-based runner)
- `pkg/queryAPI/` - Generated API client libraries