Blame for backend/app/worker/tasks.py

"""Celery task: execute one outbox job (review inference or product aggregation).

Delivery model (see :mod:`app.models.job`): Redis messages are only wake-ups
and may be duplicated or lost; the ``jobs`` row is the source of truth. Every
run locks its job row, skips finished jobs, and records retries durably with
exponential back-off. :mod:`app.worker.outbox` re-sends anything unfinished.

ML models load lazily through :func:`get_pipeline`, so importing this module
(e.g. from tests) never touches model artifacts.
"""

from __future__ import annotations

import logging
from datetime import UTC, datetime, timedelta
from functools import lru_cache
from typing import TYPE_CHECKING, Any, Final

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.config import settings
from app.database import SessionLocal
from app.models.job import Job, JobKind
from app.models.product import Product
from app.models.review import Review, ReviewStatus
from app.worker.celery_app import celery_app

if TYPE_CHECKING:
    from app.ml.pipeline import VERAPipeline

logger = logging.getLogger(__name__)

MAX_REVIEW_ATTEMPTS: Final = 4
MAX_AGGREGATE_ATTEMPTS: Final = 10
MAX_BACKOFF_SECONDS: Final = 300
#: Arbitrary constant key for the Postgres advisory lock around aggregation.
AGGREGATION_LOCK_KEY: Final = 19842026
FAILED_REVIEW_MESSAGE: Final = "Review analysis failed. Please try again later."


@lru_cache(maxsize=1)
def get_pipeline() -> VERAPipeline:
    """Load production artifacts once per worker process. Never falls back."""
    from app.ml.pipeline import VERAPipeline

    return VERAPipeline(settings.MODEL_DIR, settings.HF_MODEL_REPO, settings.HF_TOKEN)


def _as_utc(value: datetime) -> datetime:
    """Normalise DB datetimes: SQLite returns naive values, Postgres aware ones."""
    return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)


def backoff_seconds(attempts: int) -> int:
    """5 s, 10 s, 20 s, … capped at :data:`MAX_BACKOFF_SECONDS`."""
    return min(MAX_BACKOFF_SECONDS, 5 * 2 ** min(attempts, 6))


# ----------------------------------------------------------------------------
# Job handlers — run inside the caller's transaction; the caller commits.
# ----------------------------------------------------------------------------
def process_review(db: Session, review_id: int) -> None:
    """Run the pipeline for one review and store its result.

    The review row stays locked for the whole inference, so a crash rolls back
    cleanly and a duplicate delivery cannot write concurrently. The status is not
    committed as "processing" separately; clients see "pending" until "done".
    """
    review = db.query(Review).filter(Review.id == review_id).with_for_update().one_or_none()
    if review is None or review.status in (ReviewStatus.DONE.value, ReviewStatus.FAILED.value):
        return

    result = get_pipeline().process(
        text=review.text,
        star_rating=review.star_rating,
        helpful_votes=review.helpful_votes,
        verified_purchase=review.verified_purchase,
        product_title=review.product_title,
    )
    review.corrected_score = result["corrected_score"]
    review.interval_lower, review.interval_upper = result["interval"]
    review.confidence_value = result["confidence"]
    review.star_text_gap = result["star_text_gap"]
    review.review_quality = result["review_quality"]
    review.aspects = result["aspects"]
    review.flags = result["flags"]
    review.explanation = result["explanation"]
    review.error_message = None
    review.status = ReviewStatus.DONE.value

    # Reviews without an ASIN never create aggregation work (PRD weakness #3).
    if review.product_asin:
        db.add(Job(kind=JobKind.AGGREGATE.value))


def aggregate_products(db: Session, current_job_id: str | None = None) -> int:
    """Recompute EB aggregates and ranks for ALL products. Returns product count.

    The EB prior is a population quantity, so every product is refreshed under
    one lock (Postgres advisory transaction lock — released on commit/rollback,
    no TTL to expire mid-run). Other unfinished aggregate jobs visible *before*
    the review snapshot is read are marked finished: their reviews are already
    committed and therefore included, so re-running them would be pure waste.
    """
    from app.ml.calibration.empirical_bayes import (
        EmpiricalBayesAggregator,
        ReviewSignal,
        rank_products,
    )

    if db.get_bind().dialect.name == "postgresql":
        db.execute(text("SELECT pg_advisory_xact_lock(:key)"), {"key": AGGREGATION_LOCK_KEY})

    superseded = (
        db.query(Job)
        .filter(
            Job.kind == JobKind.AGGREGATE.value,
            Job.finished_at.is_(None),
            Job.id != (current_job_id or ""),
        )
        .with_for_update(skip_locked=True)
        .all()
    )

    rows = (
        db.query(
            Review.product_asin,
            Review.corrected_score,
            Review.review_quality,
            Review.star_rating,
            Review.aspects,
        )
        .filter(
            Review.status == ReviewStatus.DONE.value,
            Review.product_asin.is_not(None),
            Review.corrected_score.is_not(None),
        )
        .all()
    )
    now = datetime.now(UTC)
    for job in superseded:
        job.finished_at = now

    signals = [
        ReviewSignal(
            product_asin=row.product_asin,
            corrected_score=float(row.corrected_score),
            quality_weight=float(row.review_quality) if row.review_quality is not None else 0.5,
            star_rating=float(row.star_rating),
            aspects=row.aspects or {},
        )
        for row in rows
    ]
    if not signals:
        return 0

    _, aggregates = EmpiricalBayesAggregator().aggregate_all(signals)
    for rank, asin in enumerate(rank_products(aggregates), start=1):
        aggregate = aggregates[asin]
        product = db.get(Product, asin)
        if product is None:  # FK makes this impossible unless data was edited by hand
            raise RuntimeError(f"Completed review references missing product {asin!r}")
        product.previous_rank_position = product.rank_position
        product.rank_position = rank
        product.corrected_score = aggregate.corrected_score
        product.uncertainty = aggregate.uncertainty
        product.trust_weighted_score = aggregate.trust_weighted_score
        product.star_average = aggregate.star_average
        product.eb_shrinkage_factor = aggregate.eb_shrinkage_factor
        product.review_count = aggregate.review_count
        product.aspect_summary = aggregate.aspect_summary
        product.last_aggregated_at = now
    return len(aggregates)


# ----------------------------------------------------------------------------
# Celery entry point
# ----------------------------------------------------------------------------
def _record_failure(db: Session, job_id: str) -> None:
    """Increment attempts; schedule a retry or mark the job (and review) failed."""
    job = db.query(Job).filter(Job.id == job_id).with_for_update().one_or_none()
    if job is None or job.finished_at is not None:
        return  # a duplicate delivery finished it meanwhile

    job.attempts += 1
    limit = MAX_REVIEW_ATTEMPTS if job.kind == JobKind.REVIEW.value else MAX_AGGREGATE_ATTEMPTS
    now = datetime.now(UTC)
    if job.attempts >= limit:
        job.finished_at = now
        if job.kind == JobKind.REVIEW.value and job.review_id is not None:
            review = db.get(Review, job.review_id)
            if review is not None and review.status != ReviewStatus.DONE.value:
                review.status = ReviewStatus.FAILED.value
                review.error_message = FAILED_REVIEW_MESSAGE
        logger.error("Job %s (%s) failed permanently after %d attempts", job.id, job.kind, limit)
    else:
        job.available_at = now + timedelta(seconds=backoff_seconds(job.attempts))
        job.last_dispatched_at = None  # let the outbox re-send once due
    db.commit()


def execute_job(job_id: str) -> str:
    """Run one job to completion or record its failure. Returns an outcome label."""
    with SessionLocal() as db:
        try:
            job = (
                db.query(Job)
                .filter(Job.id == job_id)
                .with_for_update(skip_locked=True)
                .one_or_none()
            )
            if job is None or job.finished_at is not None:
                return "ignored"  # unknown, finished, or locked by a concurrent delivery
            if _as_utc(job.available_at) > datetime.now(UTC):
                return "deferred"

            if job.kind == JobKind.REVIEW.value and job.review_id is not None:
                process_review(db, job.review_id)
            elif job.kind == JobKind.AGGREGATE.value:
                aggregate_products(db, current_job_id=job.id)
            else:
                raise ValueError(f"Unknown job kind {job.kind!r}")

            job.finished_at = datetime.now(UTC)
            db.commit()
            return "done"
        except Exception:
            db.rollback()
            logger.exception("Job %s failed; recording durable retry", job_id)
            _record_failure(db, job_id)
            return "retry_or_failed"


@celery_app.task(name="vera.run_job", ignore_result=True)
def run_job(job_id: str) -> dict[str, Any]:
    """Celery wrapper around :func:`execute_job`."""
    return {"job_id": job_id, "status": execute_job(job_id)}