app/analytics/repository.py

from __future__ import annotations

from datetime import datetime, timezone

from sqlalchemy import or_, select
from sqlalchemy.exc import IntegrityError

from app.analytics.errors import (
    AnalyticsIdempotencyConflictError,
    AnalyticsNotFoundError,
    AnalyticsValidationError,
)
from app.analytics.models import (
    AnalyticsMetricSnapshot,
    AnalyticsPlatformMetric,
    AnalyticsPostMetric,
    AnalyticsSyncRun,
)
from app.social.database import SocialDatabase
from app.social.models import SocialPost, SocialPostTarget


class AnalyticsRepository:
    def __init__(self, database: SocialDatabase) -> None:
        self.database = database

    async def create_sync(self, run: AnalyticsSyncRun) -> AnalyticsSyncRun:
        try:
            async with self.database.session(run.workspace_id) as session:
                session.add(run)
                await session.commit()
                await session.refresh(run)
                return run
        except IntegrityError as exc:
            existing = await self.get_sync_by_idempotency(run.workspace_id, run.idempotency_key)
            if existing is not None:
                if (
                    existing.date_from != run.date_from
                    or existing.date_to != run.date_to
                    or existing.provider != run.provider
                    or existing.project_id != run.project_id
                ):
                    raise AnalyticsIdempotencyConflictError(
                        "The analytics idempotency key was already used for a different request."
                    ) from exc
                return existing
            raise

    async def get_sync_by_idempotency(
        self, workspace_id: str, idempotency_key: str
    ) -> AnalyticsSyncRun | None:
        async with self.database.session(workspace_id) as session:
            return await session.scalar(
                select(AnalyticsSyncRun).where(
                    AnalyticsSyncRun.workspace_id == workspace_id,
                    AnalyticsSyncRun.idempotency_key == idempotency_key,
                )
            )

    async def get_sync(self, workspace_id: str, run_id: str) -> AnalyticsSyncRun:
        async with self.database.session(workspace_id) as session:
            run = await session.scalar(
                select(AnalyticsSyncRun).where(
                    AnalyticsSyncRun.id == run_id,
                    AnalyticsSyncRun.workspace_id == workspace_id,
                )
            )
            if run is None:
                raise AnalyticsNotFoundError("Analytics sync run was not found.")
            return run

    async def list_syncs(
        self, workspace_id: str, *, offset: int, limit: int
    ) -> list[AnalyticsSyncRun]:
        async with self.database.session(workspace_id) as session:
            return list(
                (
                    await session.scalars(
                        select(AnalyticsSyncRun)
                        .where(AnalyticsSyncRun.workspace_id == workspace_id)
                        .order_by(AnalyticsSyncRun.created_at.desc())
                        .offset(offset)
                        .limit(limit)
                    )
                ).all()
            )

    async def claim_due_syncs(self, *, limit: int = 4) -> list[AnalyticsSyncRun]:
        async with self.database.worker_session() as session:
            now = datetime.now(timezone.utc)
            statement = (
                select(AnalyticsSyncRun)
                .where(
                    AnalyticsSyncRun.status.in_(["queued", "retrying"]),
                    (AnalyticsSyncRun.next_attempt_at.is_(None))
                    | (AnalyticsSyncRun.next_attempt_at <= now),
                )
                .order_by(AnalyticsSyncRun.created_at)
                .limit(limit)
                .with_for_update(skip_locked=True)
            )
            runs = list((await session.scalars(statement)).all())
            for run in runs:
                run.status = "running"
                run.started_at = now
                run.attempt_count += 1
                run.updated_at = now
            await session.commit()
            return runs

    async def update_sync(
        self,
        run_id: str,
        *,
        status: str,
        metrics_count: int | None = None,
        error_code: str | None = None,
        error_message: str | None = None,
        next_attempt_at: datetime | None = None,
    ) -> AnalyticsSyncRun:
        async with self.database.worker_session() as session:
            run = await session.get(AnalyticsSyncRun, run_id)
            if run is None:
                raise AnalyticsNotFoundError("Analytics sync run was not found.")
            run.status = status
            run.error_code = error_code
            run.error_message = error_message
            run.next_attempt_at = next_attempt_at
            if metrics_count is not None:
                run.metrics_count = metrics_count
            if status in {"succeeded", "partial", "failed", "cancelled"}:
                run.completed_at = datetime.now(timezone.utc)
            run.updated_at = datetime.now(timezone.utc)
            await session.commit()
            await session.refresh(run)
            return run

    async def cancel_sync(self, workspace_id: str, run_id: str) -> AnalyticsSyncRun:
        async with self.database.session(workspace_id) as session:
            run = await session.scalar(
                select(AnalyticsSyncRun).where(
                    AnalyticsSyncRun.id == run_id,
                    AnalyticsSyncRun.workspace_id == workspace_id,
                )
            )
            if run is None:
                raise AnalyticsNotFoundError("Analytics sync run was not found.")
            if run.status not in {"queued", "running"}:
                raise AnalyticsValidationError(
                    "Only queued or running analytics synchronization can be cancelled."
                )
            run.status = "cancelled"
            run.completed_at = datetime.now(timezone.utc)
            run.updated_at = datetime.now(timezone.utc)
            await session.commit()
            await session.refresh(run)
            return run

    async def published_targets(
        self,
        workspace_id: str,
        *,
        project_id: str | None = None,
        provider: str | None = None,
        date_from: datetime | None = None,
        date_to: datetime | None = None,
    ) -> list[tuple[SocialPost, SocialPostTarget]]:
        async with self.database.session(workspace_id) as session:
            statement = (
                select(SocialPost, SocialPostTarget)
                .join(SocialPostTarget, SocialPostTarget.social_post_id == SocialPost.id)
                .where(
                    SocialPost.workspace_id == workspace_id,
                    SocialPostTarget.external_post_id.is_not(None),
                    SocialPostTarget.status == "published",
                )
            )
            if project_id:
                statement = statement.where(SocialPost.project_id == project_id)
            if provider:
                statement = statement.where(SocialPostTarget.provider == provider)
            if date_from:
                statement = statement.where(SocialPostTarget.published_at >= date_from)
            if date_to:
                statement = statement.where(SocialPostTarget.published_at <= date_to)
            return list((await session.execute(statement)).all())

    async def upsert_post_metric(
        self,
        *,
        workspace_id: str,
        post: SocialPost,
        target: SocialPostTarget,
        metric: dict[str, object],
        collected_at: datetime,
    ) -> AnalyticsPostMetric:
        bucket = target.published_at or collected_at
        async with self.database.session(workspace_id) as session:
            existing = await session.scalar(
                select(AnalyticsPostMetric).where(
                    AnalyticsPostMetric.workspace_id == workspace_id,
                    AnalyticsPostMetric.social_post_target_id == target.id,
                    AnalyticsPostMetric.metric_date == bucket,
                )
            )
            values = dict(
                provider=target.provider,
                views=metric.get("views"),
                impressions=metric.get("impressions"),
                likes=metric.get("likes"),
                comments=metric.get("comments"),
                shares=metric.get("shares"),
                engagement_rate=metric.get("engagement_rate"),
                dimensions=(
                    metric.get("raw_metrics") if isinstance(metric.get("raw_metrics"), dict) else {}
                ),
                source="provider",
                collected_at=collected_at,
                external_post_id=target.external_post_id,
            )
            if existing is None:
                existing = AnalyticsPostMetric(
                    workspace_id=workspace_id,
                    project_id=post.project_id,
                    social_post_id=post.id,
                    social_post_target_id=target.id,
                    social_account_id=target.social_account_id,
                    metric_date=bucket,
                    **values,
                )
                session.add(existing)
            else:
                for key, value in values.items():
                    setattr(existing, key, value)
            if target.external_post_id:
                for metric_name in (
                    "views",
                    "impressions",
                    "likes",
                    "comments",
                    "shares",
                    "engagement_rate",
                ):
                    metric_value = metric.get(metric_name)
                    if not isinstance(metric_value, (int, float)) or isinstance(metric_value, bool):
                        continue
                    snapshot = await session.scalar(
                        select(AnalyticsMetricSnapshot).where(
                            AnalyticsMetricSnapshot.workspace_id == workspace_id,
                            AnalyticsMetricSnapshot.provider == target.provider,
                            AnalyticsMetricSnapshot.external_object_id == target.external_post_id,
                            AnalyticsMetricSnapshot.metric_name == metric_name,
                            AnalyticsMetricSnapshot.bucket_start == bucket,
                        )
                    )
                    if snapshot is None:
                        session.add(
                            AnalyticsMetricSnapshot(
                                workspace_id=workspace_id,
                                project_id=post.project_id,
                                social_post_id=post.id,
                                social_account_id=target.social_account_id,
                                provider=target.provider,
                                external_object_id=target.external_post_id,
                                metric_name=metric_name,
                                metric_value=float(metric_value),
                                bucket_start=bucket,
                                dimensions={},
                                source="provider",
                                collected_at=collected_at,
                            )
                        )
                    else:
                        snapshot.metric_value = float(metric_value)
                        snapshot.collected_at = collected_at
            await session.flush()
            day_start = bucket.astimezone(timezone.utc).replace(
                hour=0, minute=0, second=0, microsecond=0
            )
            day_end = day_start.replace(hour=23, minute=59, second=59, microsecond=999999)
            account_rows = list(
                (
                    await session.scalars(
                        select(AnalyticsPostMetric).where(
                            AnalyticsPostMetric.workspace_id == workspace_id,
                            AnalyticsPostMetric.social_account_id == target.social_account_id,
                            AnalyticsPostMetric.metric_date >= day_start,
                            AnalyticsPostMetric.metric_date <= day_end,
                        )
                    )
                ).all()
            )
            platform = await session.scalar(
                select(AnalyticsPlatformMetric).where(
                    AnalyticsPlatformMetric.workspace_id == workspace_id,
                    AnalyticsPlatformMetric.provider == target.provider,
                    AnalyticsPlatformMetric.social_account_id == target.social_account_id,
                    AnalyticsPlatformMetric.metric_date == day_start,
                )
            )
            aggregates: dict[str, int | float | None] = {}
            for name in (
                "views",
                "impressions",
                "likes",
                "comments",
                "shares",
                "engagement_rate",
            ):
                metric_values = [
                    getattr(item, name) for item in account_rows if getattr(item, name) is not None
                ]
                aggregates[name] = (
                    (
                        sum(metric_values) / len(metric_values)
                        if name == "engagement_rate"
                        else sum(metric_values)
                    )
                    if metric_values
                    else None
                )
            if platform is None:
                platform = AnalyticsPlatformMetric(
                    workspace_id=workspace_id,
                    project_id=post.project_id,
                    social_account_id=target.social_account_id,
                    provider=target.provider,
                    metric_date=day_start,
                )
                session.add(platform)
            platform.posts_count = len(account_rows)
            platform.collected_at = collected_at
            for name, value in aggregates.items():
                setattr(platform, name, value)
            await session.commit()
            await session.refresh(existing)
            return existing

    async def metric_rows(
        self,
        workspace_id: str,
        query: dict[str, object],
    ) -> list[AnalyticsPostMetric]:
        async with self.database.session(workspace_id) as session:
            statement = select(AnalyticsPostMetric).where(
                AnalyticsPostMetric.workspace_id == workspace_id,
                AnalyticsPostMetric.metric_date >= query["date_from"],
                AnalyticsPostMetric.metric_date <= query["date_to"],
            )
            for field, model_field in (
                ("project_id", AnalyticsPostMetric.project_id),
                ("provider", AnalyticsPostMetric.provider),
                ("social_account_id", AnalyticsPostMetric.social_account_id),
                ("post_id", AnalyticsPostMetric.social_post_id),
            ):
                value = query.get(field)
                if value:
                    statement = statement.where(model_field == value)
            if query.get("search"):
                pattern = f"%{str(query['search']).strip()}%"
                statement = statement.join(
                    SocialPost, SocialPost.id == AnalyticsPostMetric.social_post_id
                ).where(
                    or_(
                        SocialPost.canonical_caption.ilike(pattern),
                        SocialPost.id.ilike(pattern),
                    )
                )
            return list((await session.scalars(statement)).all())

    async def latest_sync(
        self, workspace_id: str, *, provider: str | None = None, project_id: str | None = None
    ) -> AnalyticsSyncRun | None:
        async with self.database.session(workspace_id) as session:
            statement = select(AnalyticsSyncRun).where(
                AnalyticsSyncRun.workspace_id == workspace_id,
                AnalyticsSyncRun.status.in_(["succeeded", "partial"]),
            )
            if provider:
                statement = statement.where(AnalyticsSyncRun.provider == provider)
            if project_id:
                statement = statement.where(AnalyticsSyncRun.project_id == project_id)
            return await session.scalar(
                statement.order_by(AnalyticsSyncRun.completed_at.desc()).limit(1)
            )