"""SQLAlchemy-backed assistant-answer feedback storage.

Each method acquires its own short-lived session. Writes are idempotent on a
SHA-256 target key derived from user/thread/message/agent; the digest avoids
oversized compound indexes on legacy MySQL utf8mb4 deployments.
"""

from __future__ import annotations

import hashlib
import uuid
from datetime import UTC, datetime
from typing import Any

from sqlalchemy import case, delete, func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from deerflow.persistence.feedback.model import FeedbackRow
from deerflow.persistence.feedback.scoring import FEEDBACK_REASONS, calculate_answer_quality_score
from deerflow.runtime.user_context import AUTO, _AutoSentinel, resolve_user_id


class FeedbackRepository:
    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
        self._sf = session_factory

    @staticmethod
    def _row_to_dict(row: FeedbackRow) -> dict[str, Any]:
        data = row.to_dict()
        for key in ("created_at", "updated_at"):
            value = data.get(key)
            if isinstance(value, datetime):
                data[key] = value.isoformat()
        data["reasons"] = list(data.get("reasons") or [])
        data["detail"] = data.get("comment")
        return data

    @staticmethod
    def make_target_key(
        *,
        user_id: str | None,
        thread_id: str,
        run_id: str,
        message_id: str | None,
        agent_id: str,
    ) -> str:
        """Build the portable idempotency identity for one answer rating.

        Historical run-level callers have no message id. Their deterministic
        ``run:<id>`` surrogate preserves the old one-rating-per-run behavior.
        """

        message_key = message_id or f"run:{run_id}"
        raw = "\x1f".join(
            (
                user_id or "",
                thread_id,
                message_key,
                agent_id or "lead_agent",
            )
        )
        return hashlib.sha256(raw.encode("utf-8")).hexdigest()

    @staticmethod
    def _normalize_reasons(reasons: list[str] | None) -> list[str]:
        allowed = set(FEEDBACK_REASONS)
        normalized: list[str] = []
        for reason in reasons or []:
            value = str(reason).strip()
            if value in allowed and value not in normalized:
                normalized.append(value)
        return normalized

    async def create(
        self,
        *,
        run_id: str,
        thread_id: str,
        rating: int,
        user_id: str | None | _AutoSentinel = AUTO,
        message_id: str | None = None,
        agent_id: str = "lead_agent",
        reasons: list[str] | None = None,
        comment: str | None = None,
    ) -> dict[str, Any]:
        """Create or update the same answer target idempotently."""

        return await self.upsert(
            run_id=run_id,
            thread_id=thread_id,
            rating=rating,
            user_id=user_id,
            message_id=message_id,
            agent_id=agent_id,
            reasons=reasons,
            comment=comment,
        )

    async def get(
        self,
        feedback_id: str,
        *,
        user_id: str | None | _AutoSentinel = AUTO,
    ) -> dict[str, Any] | None:
        resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.get")
        async with self._sf() as session:
            row = await session.get(FeedbackRow, feedback_id)
            if row is None:
                return None
            if resolved_user_id is not None and row.user_id != resolved_user_id:
                return None
            return self._row_to_dict(row)

    async def get_by_target(
        self,
        *,
        thread_id: str,
        run_id: str,
        message_id: str | None,
        agent_id: str,
        user_id: str | None | _AutoSentinel = AUTO,
    ) -> dict[str, Any] | None:
        resolved_user_id = resolve_user_id(
            user_id,
            method_name="FeedbackRepository.get_by_target",
        )
        target_key = self.make_target_key(
            user_id=resolved_user_id,
            thread_id=thread_id,
            run_id=run_id,
            message_id=message_id,
            agent_id=agent_id,
        )
        async with self._sf() as session:
            row = await session.scalar(
                select(FeedbackRow).where(FeedbackRow.target_key == target_key)
            )
            return self._row_to_dict(row) if row is not None else None

    async def list_by_run(
        self,
        thread_id: str,
        run_id: str,
        *,
        limit: int = 100,
        message_id: str | None = None,
        user_id: str | None | _AutoSentinel = AUTO,
    ) -> list[dict[str, Any]]:
        resolved_user_id = resolve_user_id(
            user_id,
            method_name="FeedbackRepository.list_by_run",
        )
        stmt = select(FeedbackRow).where(
            FeedbackRow.thread_id == thread_id,
            FeedbackRow.run_id == run_id,
        )
        if message_id is not None:
            stmt = stmt.where(FeedbackRow.message_id == message_id)
        if resolved_user_id is not None:
            stmt = stmt.where(FeedbackRow.user_id == resolved_user_id)
        stmt = stmt.order_by(FeedbackRow.created_at.asc()).limit(limit)
        async with self._sf() as session:
            result = await session.execute(stmt)
            return [self._row_to_dict(row) for row in result.scalars()]

    async def list_by_thread(
        self,
        thread_id: str,
        *,
        limit: int = 1000,
        user_id: str | None | _AutoSentinel = AUTO,
    ) -> list[dict[str, Any]]:
        resolved_user_id = resolve_user_id(
            user_id,
            method_name="FeedbackRepository.list_by_thread",
        )
        stmt = select(FeedbackRow).where(FeedbackRow.thread_id == thread_id)
        if resolved_user_id is not None:
            stmt = stmt.where(FeedbackRow.user_id == resolved_user_id)
        stmt = stmt.order_by(FeedbackRow.created_at.asc()).limit(limit)
        async with self._sf() as session:
            result = await session.execute(stmt)
            return [self._row_to_dict(row) for row in result.scalars()]

    async def list_by_message_ids(
        self,
        *,
        thread_id: str,
        message_ids: list[str],
        user_id: str | None | _AutoSentinel = AUTO,
    ) -> list[dict[str, Any]]:
        """Load current-user feedback for a bounded visible-message batch."""

        if not message_ids:
            return []
        resolved_user_id = resolve_user_id(
            user_id,
            method_name="FeedbackRepository.list_by_message_ids",
        )
        stmt = select(FeedbackRow).where(
            FeedbackRow.thread_id == thread_id,
            FeedbackRow.message_id.in_(message_ids),
        )
        if resolved_user_id is not None:
            stmt = stmt.where(FeedbackRow.user_id == resolved_user_id)
        async with self._sf() as session:
            result = await session.execute(stmt)
            return [self._row_to_dict(row) for row in result.scalars()]

    async def delete(
        self,
        feedback_id: str,
        *,
        user_id: str | None | _AutoSentinel = AUTO,
    ) -> bool:
        resolved_user_id = resolve_user_id(
            user_id,
            method_name="FeedbackRepository.delete",
        )
        async with self._sf() as session:
            row = await session.get(FeedbackRow, feedback_id)
            if row is None:
                return False
            if resolved_user_id is not None and row.user_id != resolved_user_id:
                return False
            await session.delete(row)
            await session.commit()
            return True

    async def upsert(
        self,
        *,
        run_id: str,
        thread_id: str,
        rating: int,
        user_id: str | None | _AutoSentinel = AUTO,
        message_id: str | None = None,
        agent_id: str = "lead_agent",
        reasons: list[str] | None = None,
        comment: str | None = None,
    ) -> dict[str, Any]:
        """Create or update feedback for one user/thread/message/agent target."""

        if rating not in (1, -1):
            raise ValueError(f"rating must be +1 or -1, got {rating}")
        resolved_user_id = resolve_user_id(
            user_id,
            method_name="FeedbackRepository.upsert",
        )
        normalized_agent_id = str(agent_id or "lead_agent")[:128]
        normalized_reasons = self._normalize_reasons(reasons)
        target_key = self.make_target_key(
            user_id=resolved_user_id,
            thread_id=thread_id,
            run_id=run_id,
            message_id=message_id,
            agent_id=normalized_agent_id,
        )

        # The unique digest handles cross-worker races. On a simultaneous
        # insert, retry as an update in a fresh transaction.
        for attempt in range(2):
            now = datetime.now(UTC)
            async with self._sf() as session:
                row = await session.scalar(
                    select(FeedbackRow).where(FeedbackRow.target_key == target_key)
                )
                if row is None:
                    row = FeedbackRow(
                        feedback_id=str(uuid.uuid4()),
                        run_id=run_id,
                        thread_id=thread_id,
                        user_id=resolved_user_id,
                        message_id=message_id,
                        agent_id=normalized_agent_id,
                        target_key=target_key,
                        rating=rating,
                        reasons=normalized_reasons,
                        comment=comment,
                        created_at=now,
                        updated_at=now,
                    )
                    session.add(row)
                else:
                    row.run_id = run_id
                    row.rating = rating
                    row.reasons = normalized_reasons
                    row.comment = comment
                    row.updated_at = now
                try:
                    await session.commit()
                except IntegrityError:
                    await session.rollback()
                    if attempt == 0:
                        continue
                    raise
                # Runtime factories use expire_on_commit=False. Avoid a
                # post-commit refresh, unsafe behind a read replica.
                return self._row_to_dict(row)
        raise RuntimeError("unreachable feedback upsert retry state")

    async def delete_by_run(
        self,
        *,
        thread_id: str,
        run_id: str,
        message_id: str | None = None,
        user_id: str | None | _AutoSentinel = AUTO,
    ) -> bool:
        """Delete this user's feedback for a run or one message in the run."""

        resolved_user_id = resolve_user_id(
            user_id,
            method_name="FeedbackRepository.delete_by_run",
        )
        conditions = [
            FeedbackRow.thread_id == thread_id,
            FeedbackRow.run_id == run_id,
            FeedbackRow.user_id == resolved_user_id,
        ]
        if message_id is not None:
            conditions.append(FeedbackRow.message_id == message_id)
        async with self._sf() as session:
            result = await session.execute(delete(FeedbackRow).where(*conditions))
            await session.commit()
            return bool(result.rowcount)

    async def delete_by_target(
        self,
        *,
        thread_id: str,
        run_id: str,
        message_id: str,
        agent_id: str,
        user_id: str | None | _AutoSentinel = AUTO,
    ) -> bool:
        resolved_user_id = resolve_user_id(
            user_id,
            method_name="FeedbackRepository.delete_by_target",
        )
        target_key = self.make_target_key(
            user_id=resolved_user_id,
            thread_id=thread_id,
            run_id=run_id,
            message_id=message_id,
            agent_id=agent_id,
        )
        async with self._sf() as session:
            result = await session.execute(
                delete(FeedbackRow).where(FeedbackRow.target_key == target_key)
            )
            await session.commit()
            return bool(result.rowcount)

    async def list_by_thread_grouped(
        self,
        thread_id: str,
        *,
        user_id: str | None | _AutoSentinel = AUTO,
    ) -> dict[str, dict[str, Any]]:
        """Legacy run-keyed view, retaining the newest row for each run."""

        records = await self.list_by_thread(thread_id, user_id=user_id)
        return {record["run_id"]: record for record in records}

    async def aggregate_by_run(self, thread_id: str, run_id: str) -> dict[str, Any]:
        """Aggregate feedback stats for a run using database-side counting."""

        stmt = select(
            func.count().label("total"),
            func.coalesce(
                func.sum(case((FeedbackRow.rating == 1, 1), else_=0)),
                0,
            ).label("positive"),
            func.coalesce(
                func.sum(case((FeedbackRow.rating == -1, 1), else_=0)),
                0,
            ).label("negative"),
        ).where(
            FeedbackRow.thread_id == thread_id,
            FeedbackRow.run_id == run_id,
        )
        async with self._sf() as session:
            row = (await session.execute(stmt)).one()
            return {
                "run_id": run_id,
                "total": row.total,
                "positive": row.positive,
                "negative": row.negative,
            }

    async def aggregate_by_agent(self, agent_id: str) -> dict[str, Any]:
        """Return the documented user-feedback answer-quality score."""

        normalized_agent_id = str(agent_id or "lead_agent")[:128]
        stmt = select(FeedbackRow.rating, FeedbackRow.reasons).where(
            FeedbackRow.agent_id == normalized_agent_id
        )
        async with self._sf() as session:
            rows = (await session.execute(stmt)).all()
        return calculate_answer_quality_score(
            (
                {"rating": row.rating, "reasons": list(row.reasons or [])}
                for row in rows
            ),
            agent_id=normalized_agent_id,
        )