Background Workers

The background-tasks service runs asyncio-based workers as coroutines within the main FastAPI event loop. There is no external task queue or separate worker process -- all workers share a single Docker container on EC2, using the same database connection pool.

Architecture

FastAPI Application (app/main.py)
  |
  Lifespan Context Manager
  |
  startup:
    1. Initialize asyncpg connection pool
    2. Start Match Score Generator
    3. Start Inbound Email Worker (if INBOUND_EMAIL_ENABLED)
    4. Start Inbound CV Worker (if INBOUND_CV_ENABLED)
    5. Start Document Worker (if AGENT_DOCUMENT_WORKER_ENABLED)
    6. Start Bulk CV Import Worker (if BULK_CV_IMPORT_ENABLED)
  |
  shutdown (reverse order):
    1. Signal all workers to stop (set running=False)
    2. Wait for current operations to complete
    3. Close database connection pool

All workers share these characteristics:

They are async def coroutines started via asyncio.create_task() during app startup
Each has a self.running flag checked on every loop iteration
They use asyncio.sleep() between cycles, yielding control to other coroutines
They catch and log all exceptions to prevent a single failure from crashing the worker
They participate in graceful shutdown via the FastAPI lifespan context

Match Score Generator

File: app/services/match_score_generator.py | Cycle: 5 seconds | Batch size: 100 pairs

The primary background worker. Calculates user-job match scores for all unprocessed pairs using the unified scoring system (skills 55%, experience 15%, salary 10%, location 8%, languages 5%, job_type 4%, availability 3%).

Processing Loop

Wake up every 5 seconds
  --> Query get_unscored_pairs(limit=100):
      CROSS JOIN all users (with active CV) × all non-draft jobs
      WHERE NOT EXISTS in UserJobMatchScore
  --> For each missing pair, calculate match score via unified scoring
  --> Save scores to UserJobMatchScore table
  --> Sleep until next cycle

The get_unscored_pairs() query finds gaps across all eligible users and all active jobs in a single SQL query (CROSS JOIN + NOT EXISTS). Newest users and jobs are prioritized via ORDER BY. Over successive cycles, every user-job combination is eventually scored.

On repeated failures, the circuit breaker activates and the worker enters exponential backoff (starting at 60s, up to 30 minutes max).

Control Endpoints

Endpoint	Method	Description
`/api/v1/background-tasks/match-score-generator/status`	GET	Current running state and stats
`/api/v1/background-tasks/match-score-generator/start`	POST	Start the generator
`/api/v1/background-tasks/match-score-generator/stop`	POST	Stop the generator
`/api/v1/background-tasks/match-score-generator/metrics`	GET	Processing metrics

Metrics Response

{
  "total_processed": 15000,
  "processing_rate": 2.5,
  "error_rate": 0.01,
  "last_cycle_duration_ms": 2340,
  "circuit_breaker_state": "closed"
}

info

Match scoring is excluded from billing. It runs as a background batch process, not user-initiated, to avoid unpredictable charges.

Inbound Email Worker

File: app/services/inbound_email/email_worker.py | Cycle: 30 seconds | Condition: INBOUND_EMAIL_ENABLED=true

Polls an IMAP mailbox for incoming job postings sent to jobs+company-slug@aiqlick.com.

Processing Flow

Poll IMAP mailbox (every 30s)
  --> Fetch new emails since last UID
  --> Parse company from plus-address (e.g., jobs+acme-corp@aiqlick.com)
  --> Resolve company from database by slug
  --> Store email + attachments in S3
  --> Trigger job parsing pipeline --> Creates Job record

Key Patterns

Plus-addressing: Company routing via jobs+company-slug@aiqlick.com. The slug after + is normalized and matched against company records.
Attachment selection: Prefers PDF attachments over inline images for job description extraction.
Exponential backoff: On failure: base_delay * 2^(attempt-1), capped at 1 hour.
Circuit breaker: Protects the job parsing service from cascading failures when the LLM is unavailable.

Configuration

Variable	Default	Description
`INBOUND_EMAIL_ENABLED`	`false`	Enable the worker
`EMAIL_IMAP_HOST`	--	IMAP server hostname
`EMAIL_IMAP_USER`	--	IMAP username
`EMAIL_IMAP_PASSWORD`	--	IMAP password
`EMAIL_IMAP_POLL_INTERVAL`	`30`	Poll interval (seconds)

Inbound CV Worker

File: app/services/inbound_cv/cv_email_worker.py | Cycle: 30 seconds | Condition: INBOUND_CV_ENABLED=true

Polls an IMAP mailbox for incoming CVs sent to cv+company-slug@aiqlick.com. Unlike the email worker which creates Job records, this worker creates User, Talent, and CV records.

Processing Flow

Poll IMAP mailbox (every 30s)
  --> Fetch new emails since last UID
  --> Parse company from plus-address (e.g., cv+acme-corp@aiqlick.com)
  --> Resolve company from database
  --> Store email + attachments in S3 (emailType=CV)
  --> Run CV extraction (LangChain structured output via Bedrock)
  --> Create User record (isOnboarding=false, isActive=false)
  --> Create Talent record in company talent pool (source=CV_UPLOAD)
  --> Create CV record linked to User and Talent

warning

The CV worker creates Talent records in the company pool only -- it does NOT create Candidate records. Talents must be manually added to jobs through the pipeline UI.

Configuration

Variable	Default	Description
`INBOUND_CV_ENABLED`	`false`	Enable the worker
`CV_IMAP_HOST`	--	IMAP server hostname
`CV_IMAP_USER`	--	IMAP username (cv@aiqlick.com)
`CV_IMAP_PASSWORD`	--	IMAP password
`CV_IMAP_POLL_INTERVAL`	`30`	Poll interval (seconds)
`CV_IMAP_MAX_MESSAGES`	`25`	Max messages per poll cycle
`INBOUND_CV_RETRY_LIMIT`	`3`	Max retry attempts
`INBOUND_CV_RETRY_DELAY_SECONDS`	`300`	Base retry delay (seconds)

Document Worker

File: app/services/agent/document_worker.py | Cycle: 30 seconds | Condition: AGENT_DOCUMENT_WORKER_ENABLED=true (default)

Processes RAG documents for the AI Agent system. Extracts text, generates vector embeddings, and stores indexed chunks for semantic retrieval.

Processing Flow

Poll for documents with status=PENDING (every 30s, batch of 5)
  --> Download document from S3
  --> Extract text content (PDF, DOCX, images)
  --> Split text into chunks (LangChain text splitters)
     - Strategies: recursive (default), character, token
     - Configurable chunkSize (default 300) and chunkOverlap (default 50)
  --> Generate embeddings via Amazon Titan Embed v2 (1024 dimensions, batches of 10)
  --> Store chunks with embeddings in AgentDocumentChunk table (pgvector)
  --> Update document status: PENDING --> PROCESSING --> CHUNKING --> EMBEDDING --> COMPLETED

Configuration

Variable	Default	Description
`AGENT_DOCUMENT_WORKER_ENABLED`	`true`	Enable the worker
`AGENT_DOCUMENT_POLL_INTERVAL`	`30`	Poll interval (seconds)
`AGENT_DOCUMENT_BATCH_SIZE`	`5`	Documents processed per batch
`AGENT_DOCUMENT_MAX_RETRIES`	`3`	Max retries for failed documents

Bulk CV Import Worker

File: app/services/bulk_cv/worker.py | Cycle: 30 seconds | Condition: BULK_CV_IMPORT_ENABLED=true (default)

Processes bulk CV import jobs created via the createBulkCvImport GraphQL mutation. Downloads ZIP files from S3, extracts CV documents, runs the CV extraction pipeline on each file, and creates User/Talent/CV records with deduplication by email.

Processing Flow

Poll for BulkCvImport records with status=PENDING (every 30s)
  --> Claim job via SELECT FOR UPDATE SKIP LOCKED
  --> Download ZIP from S3, validate, extract CV files
  --> For each CV (concurrency limit: 3):
     - Upload individual CV to S3
     - Billing pre-check (SKIPPED if blocked)
     - Run CV extraction pipeline (Bedrock LLM)
     - Find or create User by email
     - Find or create Talent in company pool
     - Create CV record with structured data
  --> Update progress in DB after each file
  --> Set final status: COMPLETED / PARTIALLY_COMPLETED / FAILED

Real-time progress is streamed via the bulkCvImport GraphQL subscription. See the Bulk CV Import page for full details.

Configuration

Variable	Default	Description
`BULK_CV_IMPORT_ENABLED`	`true`	Enable the worker
`BULK_CV_POLL_INTERVAL`	`30`	Poll interval (seconds)
`BULK_CV_MAX_CONCURRENT`	`3`	Max CVs processed in parallel
`BULK_CV_MAX_RETRIES`	`3`	Max retry attempts per job
`BULK_CV_MAX_FILES`	`500`	Max CV files per ZIP
`BULK_CV_MAX_ZIP_SIZE_BYTES`	`524288000`	Max compressed ZIP size (500 MB)

Circuit Breaker Pattern

All workers use circuit breakers (implemented in app/services/resilience/) to prevent cascading failures.

State Transitions

CLOSED (normal operation)
  |
  |-- 5 consecutive failures -->
  |
  v
OPEN (fail-fast, all operations rejected immediately)
  |
  |-- wait 60 seconds -->
  |
  v
HALF-OPEN (one probe request allowed)
  |
  |-- probe succeeds --> CLOSED
  |-- probe fails ----> OPEN

When the circuit is OPEN, new operations fail immediately without executing the actual operation. After the reset timeout (60 seconds), a single probe request is allowed through. If it succeeds, the circuit closes and normal operation resumes. If it fails, the circuit reopens.

Error Handling Patterns

Exponential Backoff

Used for transient failures such as database connection issues or temporary API errors:

from app.infrastructure.async_utils import retry_with_backoff

result = await retry_with_backoff(
    save_to_database,
    max_retries=3,
    base_delay=0.5,
    exceptions=(ConnectionError, TimeoutError),
)
# Delays: 0.5s, 1.0s, 2.0s

Timeout Protection

All external calls use timeouts to prevent resource exhaustion:

from app.infrastructure.async_utils import with_timeout

@with_timeout(30.0, timeout_message="Bedrock API call timed out")
async def call_bedrock():
    ...

Heavy computation is offloaded to managed AWS services (Bedrock for LLM inference, Rekognition for face detection, Transcribe for speech-to-text, Titan for embeddings), keeping the asyncio event loop responsive.

Health Check Integration

Workers report their status to the health endpoints used by Docker HEALTHCHECK and deployment verification.

GET /health/ready

{
  "status": "healthy",
  "components": {
    "database": "healthy",
    "match_score_generator": "running",
    "email_worker": "running",
    "cv_worker": "disabled",
    "document_worker": "running"
  }
}

Workers can report three states: running (actively processing), disabled (not enabled via config), or stopped (was running but shut down or failed).

Graceful Shutdown

Workers handle SIGTERM and SIGINT signals for clean shutdown during deployments:

async def stop(self):
    self.running = False
    # Wait for current operation to complete (don't interrupt mid-processing)
    if self._current_task:
        await self._current_task
    await self._cleanup()

This ensures that a deployment (which stops the old container and starts a new one) does not interrupt in-progress operations like a half-completed match score calculation, a partially processed email, or a document mid-embedding.

The shutdown sequence has a 10-second timeout for task cancellation before the database connection pool is closed.

Performance Considerations

Concurrency Model

Workers are async but do not perform CPU-intensive computation locally. The asyncio event loop stays responsive even under load because workers spend most of their time waiting for I/O (database queries, AWS API responses), during which other coroutines can run.

Database Optimization

Connection pooling via asyncpg (shared pool across all workers and request handlers)
Batch inserts for match scores to reduce round-trips
Indexed queries on isActive flag for user/job fetching
Duplicate detection before scoring (skip already-scored pairs)

Memory Management

Sequential batch processing prevents memory spikes from loading too many records
Connection pool limits prevent pool exhaustion under concurrent load
Timeouts prevent runaway coroutines from accumulating

Monitoring

All workers use structured logging with contextual fields:

from app.logging_config import get_logger
logger = get_logger(__name__)

logger.info("Processing cycle complete", extra={
    "pairs_processed": 150,
    "duration_ms": 2340,
    "errors": 0
})

Application logs are accessible via the monitoring endpoints:

Endpoint	Description
`/monitoring/stats`	Aggregated counters and uptime
`/monitoring/app-logs?level=ERROR&limit=100`	Application logs filtered by level
`/monitoring/errors`	Recent ERROR level logs
`/monitoring/ai-services`	Health check for Bedrock, S3

Worker Summary

Worker	Cycle	Condition	Creates
Match Score Generator	30s	Always on	UserJobMatchScore records
Inbound Email Worker	30s	`INBOUND_EMAIL_ENABLED=true`	Job records (via job parsing)
Inbound CV Worker	30s	`INBOUND_CV_ENABLED=true`	User + Talent + CV records
Document Worker	30s	`AGENT_DOCUMENT_WORKER_ENABLED=true`	AgentDocumentChunk records (pgvector)
Bulk CV Import Worker	30s	`BULK_CV_IMPORT_ENABLED=true`	User + Talent + CV records (batch from ZIP)
Matching Workers	On-demand	Always on	Real-time matching results (via GraphQL subscriptions)
Background Task Service	Continuous	Always on	Coordinates task lifecycle, health reporting, and worker orchestration

Architecture​

Match Score Generator​

Processing Loop​

Control Endpoints​

Metrics Response​

Inbound Email Worker​

Processing Flow​

Key Patterns​

Configuration​

Inbound CV Worker​

Processing Flow​

Configuration​

Document Worker​

Processing Flow​

Configuration​

Bulk CV Import Worker​

Processing Flow​

Configuration​

Circuit Breaker Pattern​

State Transitions​

Error Handling Patterns​

Exponential Backoff​

Timeout Protection​

Health Check Integration​

Graceful Shutdown​

Performance Considerations​

Concurrency Model​

Database Optimization​

Memory Management​

Monitoring​

Worker Summary​

Architecture

Match Score Generator

Processing Loop

Control Endpoints

Metrics Response

Inbound Email Worker

Processing Flow

Key Patterns

Configuration

Inbound CV Worker

Processing Flow

Configuration

Document Worker

Processing Flow

Configuration

Bulk CV Import Worker

Processing Flow

Configuration

Circuit Breaker Pattern

State Transitions

Error Handling Patterns

Exponential Backoff

Timeout Protection

Health Check Integration

Graceful Shutdown

Performance Considerations

Concurrency Model

Database Optimization

Memory Management

Monitoring

Worker Summary