"""Thumb up/down votes on strategic-log rows.

The model is one row per (log_id, user_id) — see
``app/models.py::StrategicLogFeedback``. The UI shows aggregate counts
only; user attribution is server-side state, not surface state. A user
can flip up → down (or vice versa) by re-submitting; "clear" deletes
the row.

For email-digest feedback links we sign a short payload with the same
itsdangerous serialiser pattern used elsewhere (``app/auth.py``). The
link is single-purpose: it identifies the (user, log, intended vote)
combination, lets the recipient click without being logged in, and
expires after EMAIL_FEEDBACK_TTL_SECONDS.
"""
from __future__ import annotations

from dataclasses import dataclass

from itsdangerous import BadSignature, SignatureExpired, URLSafeTimedSerializer
from sqlalchemy import delete, func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import get_settings
from app.db import utcnow
from app.models import StrategicLogFeedback


# Email-link tokens are valid for 30 days so a user can click the thumb
# in last week's Sunday digest. Long enough for slow readers; short
# enough that the signing-secret rotation can shake stale tokens off.
EMAIL_FEEDBACK_TTL_SECONDS = 30 * 24 * 60 * 60

# Discriminated enum of acceptable votes. "clear" is a sentinel that
# means "remove your existing vote, if any".
VALID_VOTES = ("up", "down", "clear")


class FeedbackError(ValueError):
    """Raised on bad input. Message is safe to surface to the user."""


@dataclass(frozen=True)
class FeedbackCounts:
    up: int
    down: int
    user_vote: str | None  # 'up' | 'down' | None


# ---------------------------------------------------------------------------
# Service
# ---------------------------------------------------------------------------


def _validate_vote(vote: str) -> str:
    v = (vote or "").strip().lower()
    if v not in VALID_VOTES:
        raise FeedbackError(
            f"vote must be one of {VALID_VOTES!r}; got {vote!r}"
        )
    return v


async def set_vote(
    session: AsyncSession,
    *,
    log_id: int,
    user_id: int,
    vote: str,
) -> FeedbackCounts:
    """Insert / update / clear the user's vote on ``log_id``. Returns the
    resulting aggregate counts plus the user's new vote state.

    "clear" removes the row entirely; "up"/"down" upserts. The unique
    index on (log_id, user_id) guarantees there's never more than one
    row per pair."""
    v = _validate_vote(vote)

    if v == "clear":
        await session.execute(
            delete(StrategicLogFeedback)
            .where(StrategicLogFeedback.log_id == log_id)
            .where(StrategicLogFeedback.user_id == user_id)
        )
        await session.commit()
        return await get_counts(session, log_id=log_id, user_id=user_id)

    # Upsert by hand — SQLAlchemy's portable dialect doesn't expose
    # ON DUPLICATE KEY UPDATE across MySQL+SQLite reliably for our test
    # path, so SELECT then INSERT-or-UPDATE is the simplest correct shape.
    existing = (await session.execute(
        select(StrategicLogFeedback)
        .where(StrategicLogFeedback.log_id == log_id)
        .where(StrategicLogFeedback.user_id == user_id)
    )).scalar_one_or_none()
    if existing is None:
        session.add(StrategicLogFeedback(
            log_id=log_id, user_id=user_id, vote=v,
            created_at=utcnow(), updated_at=utcnow(),
        ))
    else:
        existing.vote = v
        existing.updated_at = utcnow()
    await session.commit()
    return await get_counts(session, log_id=log_id, user_id=user_id)


async def get_counts(
    session: AsyncSession,
    *,
    log_id: int,
    user_id: int | None = None,
) -> FeedbackCounts:
    """Aggregate up/down counts for one log, plus the requesting user's
    own vote (if ``user_id`` is supplied). One indexed query for the
    counts; a second indexed lookup for the personal vote."""
    rows = (await session.execute(
        select(
            StrategicLogFeedback.vote,
            func.count(StrategicLogFeedback.id),
        )
        .where(StrategicLogFeedback.log_id == log_id)
        .group_by(StrategicLogFeedback.vote)
    )).all()
    counts = {vote: int(n) for vote, n in rows}
    user_vote: str | None = None
    if user_id is not None:
        user_vote = (await session.execute(
            select(StrategicLogFeedback.vote)
            .where(StrategicLogFeedback.log_id == log_id)
            .where(StrategicLogFeedback.user_id == user_id)
        )).scalar_one_or_none()
    return FeedbackCounts(
        up=counts.get("up", 0),
        down=counts.get("down", 0),
        user_vote=user_vote,
    )


# ---------------------------------------------------------------------------
# Email-link token helpers (parallel to app.auth.sign_pending)
# ---------------------------------------------------------------------------


def _feedback_serializer() -> URLSafeTimedSerializer:
    s = get_settings()
    secret = s.CASSANDRA_SESSION_SECRET or s.CASSANDRA_TOKEN or "dev-insecure-secret"
    return URLSafeTimedSerializer(secret, salt="cassandra-log-feedback-v1")


def sign_feedback_token(user_id: int, log_id: int, vote: str) -> str:
    """Signed token for an email-digest thumb link. Encodes the intended
    (user, log, vote) tuple. The recipient clicks the link without being
    logged in; the receiving endpoint verifies the signature, applies the
    vote, and shows a thank-you page."""
    v = _validate_vote(vote)
    if v == "clear":
        raise FeedbackError("clear is not a valid email-link vote")
    return _feedback_serializer().dumps({
        "uid": int(user_id), "lid": int(log_id), "v": v,
    })


def verify_feedback_token(token: str) -> dict | None:
    """Returns {"user_id": int, "log_id": int, "vote": "up"|"down"} on
    valid + un-expired tokens, or None on bad signature / expired / bad
    payload. The TTL is ``EMAIL_FEEDBACK_TTL_SECONDS``."""
    try:
        data = _feedback_serializer().loads(
            token, max_age=EMAIL_FEEDBACK_TTL_SECONDS,
        )
        return {
            "user_id": int(data["uid"]),
            "log_id": int(data["lid"]),
            "vote": str(data["v"]),
        }
    except (BadSignature, SignatureExpired, KeyError, TypeError, ValueError):
        return None