"""PostgreSQL state shared by every API worker, without a Redis service.

Every read enforces expiry using the database clock. Writes and counters are
atomic. Authentication failures propagate rather than silently allowing access.
Expired rows can be pruned with `python -m app.core.state_store` from a daily cron.
"""

import asyncio
import hashlib
import hmac
import logging
import secrets
from dataclasses import dataclass
from datetime import timedelta
from typing import Any

from sqlalchemy import case, delete, func, select
from sqlalchemy.dialects.postgresql import insert

from app.config import settings
from app.database import AsyncSessionLocal
from app.models.expiring_state import ExpiringState

_logger = logging.getLogger(__name__)


def _hash_token(token: str) -> str:
    return hashlib.sha256(token.encode("utf-8")).hexdigest()


def _key(namespace: str, identifier: str) -> str:
    # Opaque link tokens and email addresses are never persisted in key names.
    return f"{namespace}:{_hash_token(identifier)}"


def _otp_hash(email: str, otp: str) -> str:
    return hmac.new(
        settings.APP_SECRET_KEY.encode("utf-8"),
        f"{email.strip().lower()}:{otp}".encode(),
        hashlib.sha256,
    ).hexdigest()


async def _put(key: str, value: dict[str, Any], ttl: int) -> None:
    statement = insert(ExpiringState).values(
        key=key,
        value=value,
        expires_at=func.statement_timestamp() + timedelta(seconds=max(int(ttl), 1)),
    )
    statement = statement.on_conflict_do_update(
        index_elements=[ExpiringState.key],
        set_={"value": statement.excluded.value, "expires_at": statement.excluded.expires_at},
    )
    async with AsyncSessionLocal.begin() as session:
        await session.execute(statement)


async def _get(key: str) -> dict[str, Any] | None:
    async with AsyncSessionLocal() as session:
        value = await session.scalar(
            select(ExpiringState.value).where(
                ExpiringState.key == key,
                ExpiringState.expires_at > func.statement_timestamp(),
            )
        )
        return value


async def _delete(key: str) -> None:
    async with AsyncSessionLocal.begin() as session:
        await session.execute(delete(ExpiringState).where(ExpiringState.key == key))


async def cleanup_expired_state() -> None:
    """Remove expired entries; expiry is enforced even when cleanup has not run."""
    async with AsyncSessionLocal.begin() as session:
        await session.execute(
            delete(ExpiringState).where(ExpiringState.expires_at <= func.statement_timestamp())
        )


async def store_refresh_token(user_id: str, token: str, expires_in_seconds: int) -> None:
    await _put(_key("refresh", user_id), {"digest": _hash_token(token)}, expires_in_seconds)


async def is_refresh_token_valid(user_id: str, token: str) -> bool:
    stored = await _get(_key("refresh", user_id))
    return stored is not None and secrets.compare_digest(str(stored["digest"]), _hash_token(token))


async def revoke_refresh_token(user_id: str) -> None:
    await _delete(_key("refresh", user_id))


async def blacklist_access_token(token: str, expires_in_seconds: int = 1800) -> None:
    await _put(_key("blacklist", token), {"revoked": True}, expires_in_seconds)


async def is_access_token_blacklisted(token: str) -> bool:
    return await _get(_key("blacklist", token)) is not None


async def store_client_token(
    token: str, payload: dict[str, Any], expires_in_seconds: int = 72 * 3600
) -> None:
    await _put(_key("client_link", token), payload, expires_in_seconds)


async def get_client_token_payload(token: str) -> dict[str, Any] | None:
    return await _get(_key("client_link", token))


async def revoke_client_token(token: str) -> None:
    await _delete(_key("client_link", token))


async def store_dashboard_cache(
    cache_key: str, payload: dict[str, Any], expires_in_seconds: int = 300
) -> None:
    try:
        await _put(_key("dashboard", cache_key), payload, expires_in_seconds)
    except Exception as exc:
        _logger.warning("Failed to store optional dashboard cache (%s)", type(exc).__name__)


async def get_dashboard_cache(cache_key: str) -> dict[str, Any] | None:
    try:
        return await _get(_key("dashboard", cache_key))
    except Exception as exc:
        _logger.warning("Failed to read optional dashboard cache (%s)", type(exc).__name__)
        return None


async def store_password_reset_otp(email: str, otp: str, user_id: str, expires_in_seconds: int = 300) -> None:
    await _put(
        _key("reset_otp", email.strip().lower()),
        {"otp_hash": _otp_hash(email, otp), "user_id": user_id, "attempts": 0},
        expires_in_seconds,
    )


async def delete_password_reset_otp(email: str) -> None:
    await _delete(_key("reset_otp", email.strip().lower()))


@dataclass(frozen=True)
class OTPVerification:
    """Outcome of atomically verifying/consuming a password reset code."""

    status: str
    user_id: str | None = None
    remaining_attempts: int = 0


async def verify_password_reset_otp(email: str, otp: str) -> OTPVerification:
    """Lock the code across workers, count failures, and consume it exactly once."""
    key = _key("reset_otp", email.strip().lower())
    async with AsyncSessionLocal.begin() as session:
        entry = await session.scalar(
            select(ExpiringState)
            .where(ExpiringState.key == key, ExpiringState.expires_at > func.statement_timestamp())
            .with_for_update()
        )
        if entry is None:
            return OTPVerification("expired")
        attempts = int(entry.value.get("attempts", 0))
        if attempts >= 5:
            await session.delete(entry)
            return OTPVerification("exhausted")
        if not secrets.compare_digest(str(entry.value["otp_hash"]), _otp_hash(email, otp.strip())):
            attempts += 1
            if attempts >= 5:
                await session.delete(entry)
                return OTPVerification("exhausted")
            entry.value = {**entry.value, "attempts": attempts}
            return OTPVerification("invalid", remaining_attempts=5 - attempts)
        await session.delete(entry)
        return OTPVerification("valid", user_id=str(entry.value["user_id"]))


async def check_forgot_password_rate_limit(
    email: str, max_requests: int = 5, window_seconds: int = 900
) -> bool:
    """Count requests atomically in a fixed window shared across all workers."""
    expired = ExpiringState.expires_at <= func.statement_timestamp()
    statement = insert(ExpiringState).values(
        key=_key("reset_rate", email.strip().lower()),
        value={"count": 1},
        expires_at=func.statement_timestamp() + timedelta(seconds=max(int(window_seconds), 1)),
    )
    counter_statement = statement.on_conflict_do_update(
        index_elements=[ExpiringState.key],
        set_={
            "value": case(
                (expired, statement.excluded.value),
                else_=func.jsonb_build_object("count", ExpiringState.value["count"].as_integer() + 1),
            ),
            "expires_at": case((expired, statement.excluded.expires_at), else_=ExpiringState.expires_at),
        },
    ).returning(ExpiringState.value["count"].as_integer())
    async with AsyncSessionLocal.begin() as session:
        count = await session.scalar(counter_statement)
    return count is not None and count <= max_requests


if __name__ == "__main__":
    asyncio.run(cleanup_expired_state())