app/worker/celery_app.py

from celery import Celery
from kombu import Exchange, Queue

from app.core.config import get_settings

settings = get_settings()

# Two named queues:
#   high  — small repos (< 10 MB), expected to complete in < 30s
#   low   — large repos (>= 10 MB), may take several minutes
#
# Both are durable (survives Redis restart). The worker consumes high
# first due to queue ordering in the -Q argument (see docker-compose.yml).

default_exchange = Exchange("default", type="direct")

HIGH_QUEUE = "high"
LOW_QUEUE  = "low"

celery_app = Celery(
    "ai_code_review",
    broker=settings.redis_url,
    backend=settings.celery_result_backend or settings.redis_url,
)

celery_app.conf.update(
    # Queue definitions
    task_queues=(
        Queue(HIGH_QUEUE, default_exchange, routing_key="high", durable=True),
        Queue(LOW_QUEUE,  default_exchange, routing_key="low",  durable=True),
    ),
    task_default_queue=HIGH_QUEUE,
    task_default_exchange="default",
    task_default_routing_key="high",

    # Route analyze_repository_task to high or low based on a size hint
    # passed as a task header (set in tasks.py before .apply_async())
    task_routes={
        "analyze_repository_task": {"queue": HIGH_QUEUE},  # default; overridden at call site
    },

    # Serialisation
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],

    # Results expire after 24 h — matches the cache TTL
    result_expires=86400,

    # Eagerness (test mode)
    task_always_eager=settings.celery_task_always_eager,

    # Acknowledge only after the task completes, not on receipt.
    # Prevents a crashed worker from silently dropping a job.
    task_acks_late=True,
    task_reject_on_worker_lost=True,

    # Worker settings
    worker_prefetch_multiplier=1,   # one task at a time per worker slot
)