"""Postgres-backed per-IP rate limiting (design D10, spec rsvp-flow FR-2.12).

Deliberately not in-process: the API may run several uvicorn workers, and an in-memory
counter would silently allow N times the intended limit — one bucket per process.
"""

import hashlib
from datetime import UTC, datetime, timedelta
from typing import Any, cast

from sqlalchemy import CursorResult, delete, select
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import get_settings
from app.models import RateLimit

RSVP_BUCKET = "rsvp_submit"
RSVP_LIMIT = 5
RSVP_WINDOW = timedelta(minutes=10)

#: Password sign-in attempts from one address (design D15). Paired with, not a substitute
#: for, the per-account lockout: this one stops an attacker spreading a few guesses over many
#: accounts to stay under every individual account's threshold.
PASSWORD_SIGNIN_BUCKET = "password_signin"

#: Pending accounts one address may create by signing in with Google (design D6). Each one
#: still costs the attacker a real verified Google account; this bounds the rate.
ADMIN_SIGNUP_BUCKET = "admin_signup"


def _hash_ip(ip: str) -> str:
    pepper = get_settings().token_pepper
    return hashlib.sha256(f"{pepper}:{ip}".encode()).hexdigest()


def _window_start(window: timedelta, now: datetime | None = None) -> datetime:
    """Snap to a fixed window so concurrent requests agree on which row to increment."""
    moment = now or datetime.now(UTC)
    seconds = int(window.total_seconds())
    epoch = int(moment.timestamp())
    return datetime.fromtimestamp(epoch - (epoch % seconds), tz=UTC)


async def check_and_increment(
    session: AsyncSession,
    ip: str | None,
    *,
    bucket: str = RSVP_BUCKET,
    limit: int = RSVP_LIMIT,
    window: timedelta = RSVP_WINDOW,
) -> bool:
    """Return True if the request is allowed, incrementing the counter.

    The upsert is atomic, so two simultaneous submissions cannot both read 4 and both
    proceed.
    """
    if not ip:
        return True

    ip_hash = _hash_ip(ip)
    start = _window_start(window)

    stmt = (
        insert(RateLimit)
        .values(bucket=bucket, ip_hash=ip_hash, window_start=start, count=1)
        .on_conflict_do_update(
            index_elements=["bucket", "ip_hash", "window_start"],
            set_={"count": RateLimit.__table__.c.count + 1},
        )
        .returning(RateLimit.count)
    )
    current = await session.scalar(stmt)
    return current is not None and current <= limit


async def peek(
    session: AsyncSession,
    ip: str,
    *,
    bucket: str = RSVP_BUCKET,
    window: timedelta = RSVP_WINDOW,
) -> int:
    """Current count without incrementing — for tests and diagnostics."""
    count = await session.scalar(
        select(RateLimit.count).where(
            RateLimit.bucket == bucket,
            RateLimit.ip_hash == _hash_ip(ip),
            RateLimit.window_start == _window_start(window),
        )
    )
    return count or 0


async def cleanup(session: AsyncSession, older_than: timedelta = timedelta(hours=24)) -> int:
    """Delete expired windows. Called periodically by the worker so the table stays small."""
    cutoff = datetime.now(UTC) - older_than
    result = await session.execute(delete(RateLimit).where(RateLimit.window_start < cutoff))
    # `rowcount` is declared on CursorResult; `execute` is typed as returning the
    # generic Result, so the narrowing is explicit rather than ignored.
    return cast(CursorResult[Any], result).rowcount or 0