app/tasks/analytics_task.py

"""Author RAG Chatbot SaaS — Analytics Hourly Aggregation Task.

Runs every hour via Celery beat.
Delegates aggregation logic to core/analytics/aggregator.py.

RULE: This task is idempotent — safe to re-run for the same date.
RULE: Failures here MUST NOT affect live chat — analytics is non-critical path.
"""

import structlog
from app.tasks.celery_app import celery_app

logger = structlog.get_logger(__name__)


@celery_app.task(
    name="app.tasks.analytics_task.aggregate_hourly",
    max_retries=3,
    default_retry_delay=60,
    acks_late=True,
)
def aggregate_hourly():
    """Roll up yesterday's + today's raw events into analytics_daily rows.

    Delegates to core/analytics/aggregator.run_daily_rollup().
    Runs for today (partial day) and yesterday (full day) to catch any missed events.
    """
    from datetime import date, timedelta
    from app.services.analytics_core.aggregator import run_daily_rollup

    try:
        today = date.today()
        yesterday = today - timedelta(days=1)

        # Always process yesterday (ensures complete daily record)
        yesterday_result = run_daily_rollup(yesterday)
        logger.info("Yesterday rollup complete", **yesterday_result, date=str(yesterday))

        # Also process today (partial day — updated each hour)
        today_result = run_daily_rollup(today)
        logger.info("Today rollup complete", **today_result, date=str(today))

    except Exception as exc:
        logger.error("Analytics aggregation failed", error=str(exc))
        raise