"""Sessions: the store, the `SessionRow` mapping the kit reads, the cookie, the loader middleware.

Sources of truth: this module, `app/models/accounts.py` (`AuthSession`), `docs/ACCOUNTS.md`,
`tests/test_accounts_sessions.py`. Design: `.logs/planning/accounts-2026-09.md` §4.3.

An opaque 256-bit token rides in the cookie; its sha256 is the row's key. No signing secret
sits in the session path, so a second host or worker needs no shared key and a database read
yields nothing a browser could present. The cookie name and Secure flag follow
`settings.is_production`, never the request scheme: behind Caddy the scheme is `http`, and a
header any client can send must not pick the cookie. Absolute 30 days, no sliding, at launch.

Every store call opens its own short session (`app.db.SessionLocal`, looked up at call time
so the test suite can swap the engine), never the request's `get_db`: the one GET that
writes, the kit's View As fail-closed auto-stop, is confined to `acting_as`/`acting_mode`
on this table and can never join a domain-row transaction.
"""

from __future__ import annotations

import contextvars
import hashlib
import logging
import secrets
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta

from fastapi import FastAPI, Request
from sqlalchemy import delete, select, update

from app import db as appdb
from app.config import settings
from app.models import Account, AccountCredential, AuthSession
from app.services.throttle import client_key

log = logging.getLogger(__name__)

COOKIE_PRODUCTION = "__Host-dfp_session"
COOKIE_DEV = "dfp_session"
SESSION_SECONDS = 30 * 24 * 3600
KEEP_REVOKED_DAYS = 30


def utcnow() -> datetime:
    """The one clock the account system reads; tests replace it."""
    return datetime.now(UTC)


def aware(value: datetime | None) -> datetime | None:
    """SQLite hands back naive datetimes; treat them as UTC, which is all we ever store."""
    if value is not None and value.tzinfo is None:
        return value.replace(tzinfo=UTC)
    return value


def cookie_name() -> str:
    return COOKIE_PRODUCTION if settings.is_production else COOKIE_DEV


def cookie_secure() -> bool:
    return settings.is_production


def hash_token(token: str) -> str:
    return hashlib.sha256(token.encode()).hexdigest()


@dataclass
class SessionRow:
    """The loaded session, shaped as the mapping the kit reads: exactly the keys `user`,
    `bw_acting_as` and `bw_acting_mode`, plus the facts the access policy needs. A write to
    the kit's keys sets `dirty`; the loader flushes a dirty row after the response."""

    token_hash: str
    account_id: int
    username: str
    display_name: str
    status: str
    must_change_password: bool
    ip: str | None
    acting_as: str | None = None
    acting_mode: str | None = None
    dirty: bool = field(default=False, compare=False)

    _KEYS = ("user", "bw_acting_as", "bw_acting_mode")

    @property
    def active(self) -> bool:
        return self.status == "active"

    @property
    def effective_username(self) -> str:
        return self.acting_as or self.username

    # -- the kit's mapping contract --------------------------------------------------------
    def get(self, key, default=None):
        if key == "user":
            return self.username
        if key == "bw_acting_as":
            return self.acting_as if self.acting_as is not None else default
        if key == "bw_acting_mode":
            return self.acting_mode if self.acting_mode is not None else default
        return default

    def __getitem__(self, key):
        value = self.get(key)
        if value is None and key not in self._KEYS:
            raise KeyError(key)
        return value

    def __contains__(self, key) -> bool:
        return self.get(key) is not None

    def __setitem__(self, key, value) -> None:
        if key == "bw_acting_as":
            self.acting_as, self.dirty = (value or None), True
        elif key == "bw_acting_mode":
            self.acting_mode, self.dirty = (value or None), True
        elif key == "user":
            raise KeyError("the real user of a session never changes")
        else:
            raise KeyError(key)

    def pop(self, key, default=None):
        value = self.get(key, default)
        if key in ("bw_acting_as", "bw_acting_mode"):
            self[key] = None
        return value


# The session of the request being served, for the audit sink and anything else that must
# know who really acted without threading a Request through every call.
current: contextvars.ContextVar[SessionRow | None] = contextvars.ContextVar("dfp_session", default=None)


# --- the store ------------------------------------------------------------------------------

def mint(account_id: int, *, ip: str | None, user_agent: str | None, amr: str = "pwd",
         now: datetime | None = None) -> str:
    """Create a session row and return the token the cookie carries (never stored)."""
    now = now or utcnow()
    token = secrets.token_urlsafe(32)
    with appdb.SessionLocal() as db:
        db.add(AuthSession(
            token_hash=hash_token(token), account_id=account_id, created_at=now,
            expires_at=now + timedelta(seconds=SESSION_SECONDS), last_seen_at=now, amr=amr,
            ip=(ip or "")[:64] or None, user_agent=(user_agent or "")[:300] or None,
        ))
        db.commit()
    return token


def load(token: str | None, *, now: datetime | None = None) -> SessionRow | None:
    """The live session for a cookie value, or None (unknown, revoked, expired)."""
    if not token or len(token) > 200:
        return None
    now = now or utcnow()
    with appdb.SessionLocal() as db:
        row = db.execute(
            select(AuthSession, Account, AccountCredential.must_change_password)
            .join(Account, Account.id == AuthSession.account_id)
            .outerjoin(AccountCredential, AccountCredential.account_id == Account.id)
            .where(AuthSession.token_hash == hash_token(token))
        ).first()
    if row is None:
        return None
    session, account, must_change = row
    if session.revoked_at is not None or aware(session.expires_at) <= now:
        return None
    return SessionRow(
        token_hash=session.token_hash, account_id=account.id, username=account.username or "",
        display_name=account.display_name, status=account.status,
        must_change_password=bool(must_change), ip=session.ip,
        acting_as=session.acting_as, acting_mode=session.acting_mode,
    )


def flush(row: SessionRow) -> None:
    """Persist the kit's View As keys after they changed: the one GET that writes."""
    if not row.dirty:
        return
    with appdb.SessionLocal() as db:
        db.execute(update(AuthSession).where(AuthSession.token_hash == row.token_hash)
                   .values(acting_as=row.acting_as, acting_mode=row.acting_mode))
        db.commit()
    row.dirty = False


def touch(token_hash: str, *, now: datetime | None = None) -> None:
    """The heartbeat: last_seen_at moves; expiry does not (absolute at launch)."""
    with appdb.SessionLocal() as db:
        db.execute(update(AuthSession).where(AuthSession.token_hash == token_hash)
                   .values(last_seen_at=now or utcnow()))
        db.commit()


def revoke(token_hash: str, *, reason: str, by_id: int | None = None, now: datetime | None = None) -> int:
    with appdb.SessionLocal() as db:
        result = db.execute(
            update(AuthSession)
            .where(AuthSession.token_hash == token_hash, AuthSession.revoked_at.is_(None))
            .values(revoked_at=now or utcnow(), revoke_reason=reason[:40], revoked_by_id=by_id)
        )
        db.commit()
    return result.rowcount or 0


def revoke_account(account_id: int, *, reason: str, except_hash: str | None = None,
                   by_id: int | None = None, now: datetime | None = None) -> int:
    """Every live session of one account, optionally sparing the one that asked."""
    with appdb.SessionLocal() as db:
        stmt = (update(AuthSession)
                .where(AuthSession.account_id == account_id, AuthSession.revoked_at.is_(None))
                .values(revoked_at=now or utcnow(), revoke_reason=reason[:40], revoked_by_id=by_id))
        if except_hash:
            stmt = stmt.where(AuthSession.token_hash != except_hash)
        result = db.execute(stmt)
        db.commit()
    return result.rowcount or 0


def revoke_all(*, reason: str, by_id: int | None = None, now: datetime | None = None) -> int:
    """The break-glass: everyone out, at once."""
    with appdb.SessionLocal() as db:
        result = db.execute(update(AuthSession).where(AuthSession.revoked_at.is_(None))
                            .values(revoked_at=now or utcnow(), revoke_reason=reason[:40], revoked_by_id=by_id))
        db.commit()
    return result.rowcount or 0


def prune(*, now: datetime | None = None) -> int:
    """Delete expired rows and revoked rows older than KEEP_REVOKED_DAYS."""
    now = now or utcnow()
    with appdb.SessionLocal() as db:
        result = db.execute(delete(AuthSession).where(
            (AuthSession.expires_at <= now)
            | ((AuthSession.revoked_at.isnot(None)) & (AuthSession.revoked_at <= now - timedelta(days=KEEP_REVOKED_DAYS)))
        ))
        db.commit()
    return result.rowcount or 0


def active_impersonations() -> list[dict]:
    """Every live session with a View As on it: one SELECT, the owner's oversight read."""
    now = utcnow()
    with appdb.SessionLocal() as db:
        rows = db.execute(
            select(Account.username, AuthSession.acting_as, AuthSession.acting_mode,
                   AuthSession.last_seen_at, AuthSession.created_at)
            .join(Account, Account.id == AuthSession.account_id)
            .where(AuthSession.acting_as.isnot(None), AuthSession.revoked_at.is_(None),
                   AuthSession.expires_at > now)
        ).all()
    return [{"real": r[0], "target": r[1], "mode": r[2],
             "last_seen_at": aware(r[3]).isoformat() if r[3] else None,
             "since": aware(r[4]).isoformat() if r[4] else None} for r in rows]


# --- the request side -----------------------------------------------------------------------

def set_cookie(response, token: str) -> None:
    response.set_cookie(
        cookie_name(), token, max_age=SESSION_SECONDS, path="/",
        secure=cookie_secure(), httponly=True, samesite="lax",
    )


def clear_cookie(response) -> None:
    response.delete_cookie(cookie_name(), path="/", secure=cookie_secure(), httponly=True, samesite="lax")


def of_request(request: Request) -> SessionRow | None:
    """The session the loader attached, or None."""
    return getattr(request.state, "session", None)


def install(app: FastAPI) -> None:
    """The loader: outermost of the auth middlewares, so the access policy and every route
    see `request.state.session`; a dirty row (a View As started, stopped or auto-stopped) is
    flushed after the response."""

    @app.middleware("http")
    async def session_loader(request: Request, call_next):
        row = load(request.cookies.get(cookie_name()))
        request.state.session = row
        request.state.client_ip = client_key(request)
        token = current.set(row)
        try:
            response = await call_next(request)
        finally:
            current.reset(token)
        if row is not None and row.dirty:
            try:
                flush(row)
            except Exception:  # noqa: BLE001 - never turn a served page into a 500 after the fact
                log.exception("session flush failed")
        return response