# Pipeline Architecture

## Overview

The pipeline system manages asynchronous document processing with persistent job state and coordinated execution across embedded or external workers.

## Architecture Pattern

The pipeline uses a Factory pattern to provide a unified interface for both local and remote job processing:

```mermaid
classDiagram
    class IPipeline {
        <<interface>>
        +start() Promise~void~
        +stop() Promise~void~
        +enqueueScrapeJob() Promise~string~
        +enqueueRefreshJob() Promise~string~
        +getJob() Promise~PipelineJob~
        +cancelJob() Promise~void~
    }
    
    class PipelineFactory {
        <<factory>>
        +createPipeline(docService, options) Promise~IPipeline~
    }
    
    class PipelineManager {
        -docService DocumentManagementService
        -workers PipelineWorker[]
        -queue JobQueue
        +start() Promise~void~
        +enqueueScrapeJob() Promise~string~
    }
    
    class PipelineClient {
        -serverUrl string
        -trpcClient TRPCClient
        +start() Promise~void~
        +enqueueScrapeJob() Promise~string~
    }
    
    class PipelineWorker {
        -job PipelineJob
        -scraper ScraperService
        +execute() Promise~void~
    }
    
    IPipeline <|.. PipelineManager : implements
    IPipeline <|.. PipelineClient : implements
    PipelineFactory ..> IPipeline : creates
    PipelineFactory ..> PipelineManager : creates (local)
    PipelineFactory ..> PipelineClient : creates (remote)
    PipelineManager o-- PipelineWorker : manages
```

**Code Reference:** 
- `src/pipeline/PipelineFactory.ts` - Factory implementation
- `src/pipeline/trpc/interfaces.ts` lines 20-35 - IPipeline interface
- `src/pipeline/PipelineManager.ts` - Local implementation
- `src/pipeline/PipelineClient.ts` - Remote implementation

## Core Components

### PipelineFactory

Central factory that selects pipeline implementation based on configuration:

```typescript
interface PipelineOptions {
  recoverJobs?: boolean; // Enable job recovery from database
  serverUrl?: string; // External worker URL
  concurrency?: number; // Worker concurrency limit
}
```

**Selection Logic:**

- `serverUrl` specified → PipelineClient (external worker)
- `recoverJobs: true` → PipelineManager with recovery
- `recoverJobs: false` → PipelineManager without recovery

### PipelineManager

Manages job queue and worker coordination for embedded processing:

**Responsibilities:**

- Job queue management with concurrency limits
- Worker lifecycle management
- Progress tracking and status updates
- Database state synchronization
- Job recovery after restart

**Job Recovery:**

- Loads QUEUED and RUNNING jobs from database on startup
- Resets RUNNING jobs to QUEUED for re-execution
- Maintains job configuration for reproducible processing

### PipelineClient

Type-safe tRPC client providing identical interface to PipelineManager for external worker communication:

**Features:**

- tRPC client for remote job operations over HTTP
- Identical method signatures to PipelineManager
- Error handling and connection management
- Connectivity check via ping

### PipelineWorker

Executes individual jobs with progress reporting:

**Execution Flow:**

1. Fetch job configuration from queue
2. Initialize scraper with job parameters
3. Process content through scraper pipeline
4. Update progress via callbacks
5. Store results and mark completion

## Job Lifecycle

### Job States

```
QUEUED → RUNNING → COMPLETED
              ↓
            FAILED
              ↓
           CANCELLED
```

### State Transitions

- **QUEUED**: Job created, waiting for worker
- **RUNNING**: Worker processing job
- **COMPLETED**: Successful completion
- **FAILED**: Error during processing
- **CANCELLED**: Manual cancellation

### Progress Tracking

Jobs report progress through callback mechanism:

- Pages discovered and processed
- Current processing status
- Error messages and warnings
- Estimated completion time

## Write-Through Architecture

### Single Source of Truth

PipelineJob objects contain both runtime state and database fields, ensuring consistency between memory and persistent storage.

### Immediate Persistence

All state changes immediately write to database:

- Status transitions
- Progress updates
- Error information
- Configuration changes

### Recovery Mechanism

Database state enables automatic recovery:

1. Load pending jobs on startup
2. Reset RUNNING jobs to QUEUED
3. Resume processing with original configuration
4. Maintain progress history

## Concurrency Management

### Worker Pool

PipelineManager maintains configurable worker pool:

- Default concurrency: 3 workers
- Configurable via environment or CLI
- Workers process jobs independently
- Queue coordination prevents conflicts

### Job Distribution

Jobs are distributed to available workers using:

- FIFO queue ordering
- Worker availability checking
- Load balancing across workers
- Graceful worker shutdown handling

## External Worker RPC

### Procedures (tRPC)

- `ping` - Connectivity check
- `enqueueJob` - Create new job
- `getJobs` - List jobs with optional filtering
- `getJob` - Get job details
- `cancelJob` - Cancel a job
- `clearCompletedJobs` - Remove completed/cancelled/failed jobs

### Data Contracts

Requests and responses use shared TypeScript types through tRPC, ensuring end-to-end type safety.

### Error Handling

Errors propagate as structured tRPC errors with messages suitable for user feedback and logs.

## Configuration Persistence

### Job Configuration

Each job stores complete scraper configuration:

- Source URL and scraping parameters
- Library name and version information
- Processing options and filters
- Retry and timeout settings

### Reproducible Processing

Stored configuration enables:

- Exact re-indexing with same parameters
- Configuration auditing and debugging
- Version-specific processing rules
- Consistent results across runs

## Monitoring and Observability

### Progress Reporting

Real-time progress updates through:

- Callback-based progress notifications
- Database persistence of progress state
- Web UI polling for status display
- Log-based progress tracking

### Error Tracking

Comprehensive error information:

- Exception stack traces
- Processing context at failure
- Retry attempt logging
- User-friendly error messages

### Performance Metrics

Job execution metrics:

- Processing duration
- Pages processed per minute
- Memory and resource usage
- Queue depth and throughput

## Scaling Patterns

### Vertical Scaling

Increase worker concurrency within single process:

- Higher concurrency limits
- More memory allocation
- Faster storage backend

### Horizontal Scaling

Distribute workers across processes:

- External worker deployment
- Load balancer coordination
- Independent worker scaling
- Database connection pooling

### Hybrid Deployment

Combine embedded and external workers:

- Coordinator with embedded workers
- Additional external workers for peak load
- Flexible resource allocation
- Cost-optimized scaling
