app/services/analytics_core/aggregator.py

"""Author RAG Chatbot SaaS — Analytics Aggregator.

Runs as a Celery beat task every hour.
Aggregates raw analytics_events into pre-computed analytics_daily rows.

RULE: This task is idempotent — safe to re-run for the same date.
RULE: Uses INSERT ... ON CONFLICT (upsert) so partial runs don't create duplicates.
RULE: Failures here MUST NOT affect live chat — analytics is non-critical path.
"""

import asyncio
import uuid
from datetime import date, datetime, timedelta, timezone

import structlog

logger = structlog.get_logger(__name__)


async def _run_daily_rollup(target_date: date) -> dict:
    """Aggregate analytics_events for target_date into analytics_daily.

    For each author that had events on target_date:
      - Count total chat turns
      - Count unique visitor fingerprints
      - Sum tokens used
      - Sum link clicks
      - Average session turns
      - Average faithfulness score

    Args:
        target_date: The calendar date to aggregate (UTC).

    Returns:
        Dict summarising rows written: {authors_processed, rows_written}.
    """
    from sqlalchemy import func, select, text
    from app.dependencies import _get_session_factory
    from app.models.analytics import AnalyticsEvent, AnalyticsDaily

    async with _get_session_factory()() as db:
        # Fetch per-author aggregates for the target date
        day_start = datetime(target_date.year, target_date.month, target_date.day, tzinfo=timezone.utc)
        day_end = day_start + timedelta(days=1)

        result = await db.execute(
            select(
                AnalyticsEvent.author_id,
                func.count(AnalyticsEvent.id).label("total_chats"),
                func.count(func.distinct(AnalyticsEvent.visitor_fingerprint)).label("unique_visitors"),
                func.sum(AnalyticsEvent.prompt_tokens + AnalyticsEvent.completion_tokens).label("total_tokens"),
                # DESIGN-1 fix: was summing link_shown (button displayed) — should be
                # link_clicked (button actually clicked). Fixes inflated conversion funnel.
                func.sum(AnalyticsEvent.link_clicked.cast("int")).label("total_link_clicks"),
                func.avg(AnalyticsEvent.faithfulness_score).label("avg_faithfulness"),
            ).where(
                AnalyticsEvent.timestamp >= day_start,
                AnalyticsEvent.timestamp < day_end,
            ).group_by(AnalyticsEvent.author_id)
        )
        rows = result.all()

        if not rows:
            logger.info("Aggregator: no events for date", date=target_date.isoformat())
            return {"authors_processed": 0, "rows_written": 0}

        rows_written = 0
        for row in rows:
            author_id = row.author_id

            # Upsert into analytics_daily (raw SQL for portability)
            await db.execute(
                text("""
                    INSERT INTO analytics_daily
                        (id, author_id, date, total_chats, unique_visitors,
                         total_tokens_used, total_link_clicks, avg_session_turns, avg_faithfulness_score)
                    VALUES
                        (:row_id, :author_id, :date, :total_chats, :unique_visitors,
                         :total_tokens, :total_link_clicks, 0.0, :avg_faithfulness)
                    ON CONFLICT (author_id, date)
                    DO UPDATE SET
                        total_chats          = EXCLUDED.total_chats,
                        unique_visitors      = EXCLUDED.unique_visitors,
                        total_tokens_used    = EXCLUDED.total_tokens_used,
                        total_link_clicks    = EXCLUDED.total_link_clicks,
                        avg_faithfulness_score = EXCLUDED.avg_faithfulness_score
                """),
                {
                    "row_id": str(uuid.uuid4()),
                    "author_id": author_id,
                    "date": target_date.isoformat(),
                    "total_chats": int(row.total_chats or 0),
                    "unique_visitors": int(row.unique_visitors or 0),
                    "total_tokens": int(row.total_tokens or 0),
                    "total_link_clicks": int(row.total_link_clicks or 0),
                    "avg_faithfulness": float(row.avg_faithfulness or 0.0),
                },
            )
            rows_written += 1

        await db.commit()
        logger.info(
            "Daily rollup complete",
            date=target_date.isoformat(),
            authors=len(rows),
            rows_written=rows_written,
        )
        return {"authors_processed": len(rows), "rows_written": rows_written}


def run_daily_rollup(target_date: date | None = None) -> dict:
    """Synchronous wrapper — called by Celery analytics_task.

    Args:
        target_date: Date to aggregate. Defaults to yesterday (UTC).

    Returns:
        Aggregation result dict.
    """
    if target_date is None:
        target_date = (datetime.now(timezone.utc) - timedelta(days=1)).date()

    logger.info("Starting daily analytics rollup", date=target_date.isoformat())
    return asyncio.run(_run_daily_rollup(target_date))