"""Server-side browser sessions for the BFF auth boundary.

Sessions are DB-backed when PostgreSQL is available and fall back to the
process-local store for local degraded development. The browser only receives an
opaque HttpOnly cookie; roles and user identity stay on the server.
"""

from __future__ import annotations

import hashlib
import hmac
import secrets
import time
import uuid
from datetime import datetime, timezone
from dataclasses import dataclass
from typing import Literal

from .config import settings
from .db import get_pool
from .runtime_policy import require_runtime_fallback_allowed, runtime_fallback_allowed


@dataclass(slots=True)
class SessionUser:
    user_id: str
    email: str
    display_name: str
    role: str
    cohort_ids: list[str]
    expires_at: float


@dataclass(slots=True)
class ManagedUser:
    user_id: str
    email: str
    display_name: str
    role: str
    cohort_ids: list[str]
    affiliation: str
    created_at: float
    last_seen_at: float


class InactiveUserError(Exception):
    """Raised when an inactive managed user attempts to create a login session."""


_sessions: dict[str, SessionUser] = {}
_users: dict[str, ManagedUser] = {}
_email_index: dict[str, str] = {}
_inactive_emails: set[str] = set()

DEFAULT_AFFILIATION = settings.default_affiliation.strip()
RoleName = Literal["learner", "teacher", "admin"]
DB_ROLE_BY_APP = {"learner": "learner", "teacher": "instructor", "admin": "admin"}
APP_ROLE_BY_DB = {"learner": "learner", "instructor": "teacher", "admin": "admin"}


def _sid_hash(raw_sid: str) -> str:
    return hmac.new(
        settings.session_secret.encode("utf-8"),
        raw_sid.encode("utf-8"),
        hashlib.sha256,
    ).hexdigest()


def user_id_from_email(email: str) -> str:
    return str(uuid.uuid5(uuid.NAMESPACE_URL, f"vignette:user:{email.strip().lower()}"))


def user_id_from_external_id(external_id: str) -> str:
    return str(uuid.uuid5(uuid.NAMESPACE_URL, f"vignette:user-external:{external_id.strip().lower()}"))


def _normalize_email(email: str) -> str:
    return email.strip().lower()


def _normalize_external_id(external_id: str | None, email: str) -> str:
    value = (external_id or "").strip().lower()
    return value or f"email:{_normalize_email(email)}"


def _db_role(role: str) -> str:
    return DB_ROLE_BY_APP.get(role, role)


def _app_role(role: str) -> str:
    return APP_ROLE_BY_DB.get(role, role)


def _ts(value: datetime | None) -> float:
    return (value or datetime.now(timezone.utc)).timestamp()


def _cohort_ids(cohort: str | None) -> list[str]:
    return [cohort] if cohort else []


def _cohort_value(cohort_ids: list[str] | None) -> str | None:
    return (cohort_ids or [None])[0]


async def _runtime_tables_ready(conn) -> bool:
    row = await conn.fetchrow(
        """
        SELECT
          EXISTS (
            SELECT 1 FROM information_schema.columns
            WHERE table_schema = 'app'
              AND table_name = 'app_user'
              AND column_name IN ('affiliation', 'last_seen_at', 'updated_at')
            GROUP BY table_schema, table_name
            HAVING count(*) = 3
          ) AS has_user_columns,
          to_regclass('app.auth_session') IS NOT NULL AS has_auth_session,
          to_regclass('app.user_preferences') IS NOT NULL AS has_preferences,
          to_regclass('app.admin_engine_config') IS NOT NULL AS has_engine_config,
          EXISTS (
            SELECT 1 FROM information_schema.columns
            WHERE table_schema = 'app'
              AND table_name = 'sessions'
              AND column_name IN (
                'runtime_case_id',
                'persona_code',
                'persona_display_name',
                'persona_difficulty',
                'prev_rapport_credit'
              )
            GROUP BY table_schema, table_name
            HAVING count(*) = 5
          ) AS has_session_columns,
          EXISTS (
            SELECT 1 FROM information_schema.columns
            WHERE table_schema = 'app'
              AND table_name = 'session_state'
              AND column_name = 'turns_in_stage'
          ) AS has_state_columns,
          (
            SELECT count(*) = 4
            FROM app.stage_def
            WHERE stage_code IN ('라포','탐색','개입','정리')
          ) AS has_stage_defs,
          EXISTS (
            SELECT 1 FROM pg_policies
            WHERE schemaname = 'app'
              AND tablename = 'sessions'
              AND policyname IN ('p_sessions_insert','p_sessions_update','p_sessions_delete')
            GROUP BY schemaname, tablename
            HAVING count(*) = 3
          ) AS has_session_write_policies,
          NOT EXISTS (
            SELECT 1 FROM pg_policies
            WHERE schemaname = 'app'
              AND tablename = 'sessions'
              AND policyname = 'p_sessions_modify'
          ) AS removed_old_session_policy,
          EXISTS (
            SELECT 1 FROM pg_policies
            WHERE schemaname = 'app'
              AND tablename = 'turns'
              AND policyname IN ('p_turns_insert','p_turns_update','p_turns_delete')
            GROUP BY schemaname, tablename
            HAVING count(*) = 3
          ) AS has_turn_write_policies,
          NOT EXISTS (
            SELECT 1 FROM pg_policies
            WHERE schemaname = 'app'
              AND tablename = 'turns'
              AND policyname = 'p_turns_modify'
          ) AS removed_old_turn_policy
        """
    )
    return bool(
        row
        and row["has_user_columns"]
        and row["has_auth_session"]
        and row["has_preferences"]
        and row["has_engine_config"]
        and row["has_session_columns"]
        and row["has_state_columns"]
        and row["has_stage_defs"]
        and row["has_session_write_policies"]
        and row["removed_old_session_policy"]
        and row["has_turn_write_policies"]
        and row["removed_old_turn_policy"]
    )


async def ensure_runtime_tables() -> None:
    """Ensure DB-backed auth/user runtime tables exist when a pool is available."""
    pool = get_pool()
    async with pool.acquire() as conn:
        await conn.execute(
            """
            DO $$
            BEGIN
              IF EXISTS (
                SELECT 1 FROM information_schema.columns
                WHERE table_schema = 'app'
                  AND table_name = 'app_user'
                  AND column_name = 'affiliation'
              ) THEN
                ALTER TABLE app.app_user ALTER COLUMN affiliation SET DEFAULT '';
              END IF;
            END $$;
            """
        )
        if await _runtime_tables_ready(conn):
            return
        await conn.execute(
            """
            ALTER TABLE app.app_user
              ADD COLUMN IF NOT EXISTS affiliation TEXT NOT NULL DEFAULT '',
              ADD COLUMN IF NOT EXISTS last_seen_at TIMESTAMPTZ NOT NULL DEFAULT now(),
              ADD COLUMN IF NOT EXISTS updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
            """
        )
        await conn.execute(
            """
            CREATE TABLE IF NOT EXISTS app.auth_session (
              sid_hash TEXT PRIMARY KEY,
              user_id UUID NOT NULL REFERENCES app.app_user(user_id) ON DELETE CASCADE,
              role TEXT NOT NULL,
              display_name TEXT NOT NULL,
              cohort_ids JSONB NOT NULL DEFAULT '[]'::jsonb,
              expires_at TIMESTAMPTZ NOT NULL,
              revoked_at TIMESTAMPTZ,
              created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
              last_seen_at TIMESTAMPTZ NOT NULL DEFAULT now()
            )
            """
        )
        await conn.execute(
            """
            CREATE INDEX IF NOT EXISTS idx_auth_session_user_active
              ON app.auth_session(user_id, expires_at)
              WHERE revoked_at IS NULL
            """
        )
        await conn.execute(
            """
            CREATE TABLE IF NOT EXISTS app.user_preferences (
              user_id UUID PRIMARY KEY REFERENCES app.app_user(user_id) ON DELETE CASCADE,
              theme TEXT NOT NULL DEFAULT 'system',
              voice_preset_id TEXT NOT NULL DEFAULT 'soft-young-fem',
              voice_rate REAL NOT NULL DEFAULT 1.0,
              notifications JSONB NOT NULL DEFAULT '{}'::jsonb,
              updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
            )
            """
        )
        await conn.execute(
            """
            CREATE TABLE IF NOT EXISTS app.admin_engine_config (
              id BOOLEAN PRIMARY KEY DEFAULT TRUE CHECK (id),
              engine_mode TEXT NOT NULL,
              engine_url TEXT NOT NULL,
              model TEXT NOT NULL,
              updated_by TEXT,
              updated_at TIMESTAMPTZ
            )
            """
        )
        await conn.execute(
            """
            ALTER TABLE app.sessions
              ADD COLUMN IF NOT EXISTS runtime_case_id UUID,
              ADD COLUMN IF NOT EXISTS persona_code TEXT,
              ADD COLUMN IF NOT EXISTS persona_display_name TEXT,
              ADD COLUMN IF NOT EXISTS persona_difficulty TEXT,
              ADD COLUMN IF NOT EXISTS prev_rapport_credit REAL NOT NULL DEFAULT 0.0
            """
        )
        await conn.execute(
            """
            ALTER TABLE app.session_state
              ADD COLUMN IF NOT EXISTS turns_in_stage INT NOT NULL DEFAULT 0
            """
        )
        await conn.execute(
            """
            INSERT INTO app.stage_def (stage_code, display_name, seq, base_openness)
            VALUES
              ('라포', '라포', 1, 0.15),
              ('탐색', '탐색', 2, 0.35),
              ('개입', '개입', 3, 0.55),
              ('정리', '정리', 4, 0.45)
            ON CONFLICT (stage_code) DO UPDATE SET
              display_name = EXCLUDED.display_name,
              seq = EXCLUDED.seq,
              base_openness = EXCLUDED.base_openness
            """
        )
        await conn.execute(
            """
            DROP POLICY IF EXISTS p_sessions_modify ON app.sessions;
            DROP POLICY IF EXISTS p_sessions_insert ON app.sessions;
            DROP POLICY IF EXISTS p_sessions_update ON app.sessions;
            DROP POLICY IF EXISTS p_sessions_delete ON app.sessions;

            CREATE POLICY p_sessions_insert ON app.sessions FOR INSERT WITH CHECK (
              app.is_ai_context() OR app.current_role_name() IN ('admin','instructor')
              OR learner_id = app.current_uid()
            );
            CREATE POLICY p_sessions_update ON app.sessions FOR UPDATE USING (
              app.is_ai_context() OR app.current_role_name() IN ('admin','instructor')
              OR learner_id = app.current_uid()
            ) WITH CHECK (
              app.is_ai_context() OR app.current_role_name() IN ('admin','instructor')
              OR learner_id = app.current_uid()
            );
            CREATE POLICY p_sessions_delete ON app.sessions FOR DELETE USING (
              app.is_ai_context() OR app.current_role_name() IN ('admin','instructor')
              OR learner_id = app.current_uid()
            )
            """
        )
        await conn.execute(
            """
            DROP POLICY IF EXISTS p_case_insert ON app.case_profile;
            DROP POLICY IF EXISTS p_case_update ON app.case_profile;

            CREATE POLICY p_case_insert ON app.case_profile FOR INSERT WITH CHECK (
              app.is_ai_context() OR app.current_role_name() IN ('admin','instructor')
              OR learner_id = app.current_uid()
            );
            CREATE POLICY p_case_update ON app.case_profile FOR UPDATE USING (
              app.is_ai_context() OR app.current_role_name() IN ('admin','instructor')
              OR learner_id = app.current_uid()
            ) WITH CHECK (
              app.is_ai_context() OR app.current_role_name() IN ('admin','instructor')
              OR learner_id = app.current_uid()
            )
            """
        )
        await conn.execute(
            """
            DROP POLICY IF EXISTS p_turns_modify ON app.turns;
            DROP POLICY IF EXISTS p_turns_insert ON app.turns;
            DROP POLICY IF EXISTS p_turns_update ON app.turns;
            DROP POLICY IF EXISTS p_turns_delete ON app.turns;

            CREATE POLICY p_turns_insert ON app.turns FOR INSERT WITH CHECK (
              app.is_ai_context()
              OR app.current_role_name() IN ('admin','instructor')
              OR EXISTS (
                SELECT 1 FROM app.sessions s
                WHERE s.id = app.turns.session_id
                  AND s.learner_id = app.current_uid()
              )
            );
            CREATE POLICY p_turns_update ON app.turns FOR UPDATE USING (
              app.is_ai_context()
              OR app.current_role_name() IN ('admin','instructor')
              OR EXISTS (
                SELECT 1 FROM app.sessions s
                WHERE s.id = app.turns.session_id
                  AND s.learner_id = app.current_uid()
              )
            ) WITH CHECK (
              app.is_ai_context()
              OR app.current_role_name() IN ('admin','instructor')
              OR EXISTS (
                SELECT 1 FROM app.sessions s
                WHERE s.id = app.turns.session_id
                  AND s.learner_id = app.current_uid()
              )
            );
            CREATE POLICY p_turns_delete ON app.turns FOR DELETE USING (
              app.is_ai_context()
              OR app.current_role_name() IN ('admin','instructor')
              OR EXISTS (
                SELECT 1 FROM app.sessions s
                WHERE s.id = app.turns.session_id
                  AND s.learner_id = app.current_uid()
              )
            )
            """
        )


def _managed_user_from_row(row) -> ManagedUser:
    return ManagedUser(
        user_id=str(row["user_id"]),
        email=row["email"],
        display_name=row["display_name"] or row["email"],
        role=_app_role(row["role"]),
        cohort_ids=_cohort_ids(row.get("cohort") if hasattr(row, "get") else row["cohort"]),
        affiliation=row["affiliation"] or DEFAULT_AFFILIATION,
        created_at=_ts(row["created_at"]),
        last_seen_at=_ts(row["last_seen_at"]),
    )


def _memory_upsert_managed_user(
    *,
    email: str,
    display_name: str,
    role: str,
    cohort_ids: list[str] | None = None,
    user_id: str | None = None,
    affiliation: str | None = None,
    reactivate: bool = False,
) -> ManagedUser:
    now = time.time()
    normalized_email = _normalize_email(email)
    if normalized_email in _inactive_emails and not reactivate:
        raise InactiveUserError("user is inactive")
    if reactivate:
        _inactive_emails.discard(normalized_email)
    uid = user_id or _email_index.get(normalized_email) or user_id_from_email(normalized_email)
    current = _users.get(uid)
    user = ManagedUser(
        user_id=uid,
        email=normalized_email,
        display_name=(display_name.strip() if display_name else "") or normalized_email,
        role=role,
        cohort_ids=list(cohort_ids or current.cohort_ids if current else cohort_ids or []),
        affiliation=(
            affiliation.strip()
            if affiliation
            else (current.affiliation if current else DEFAULT_AFFILIATION)
        ),
        created_at=current.created_at if current else now,
        last_seen_at=now,
    )
    _users[uid] = user
    _email_index[normalized_email] = uid
    return user


async def upsert_managed_user(
    *,
    email: str,
    display_name: str,
    role: str,
    cohort_ids: list[str] | None = None,
    user_id: str | None = None,
    external_id: str | None = None,
    affiliation: str | None = None,
    reactivate: bool = False,
) -> ManagedUser:
    normalized_email = _normalize_email(email)
    normalized_external_id = _normalize_external_id(external_id, normalized_email)
    try:
      pool = get_pool()
      async with pool.acquire() as conn:
        row = await conn.fetchrow(
            """
            INSERT INTO app.app_user (
              external_id, email, display_name, role, cohort, affiliation, last_seen_at, updated_at
            )
            VALUES ($1, $2, $3, $4, $5, $6, now(), now())
            ON CONFLICT (external_id) DO UPDATE SET
              email = EXCLUDED.email,
              display_name = COALESCE(NULLIF(EXCLUDED.display_name, ''), app.app_user.display_name),
              role = EXCLUDED.role,
              cohort = COALESCE(EXCLUDED.cohort, app.app_user.cohort),
              affiliation = COALESCE(NULLIF(EXCLUDED.affiliation, ''), app.app_user.affiliation),
              is_active = CASE WHEN $7 THEN TRUE ELSE app.app_user.is_active END,
              last_seen_at = now(),
              updated_at = now()
            WHERE app.app_user.is_active OR $7
            RETURNING user_id, email, display_name, role, cohort, affiliation, created_at, last_seen_at
            """,
            normalized_external_id,
            normalized_email,
            (display_name.strip() if display_name else normalized_email),
            _db_role(role),
            _cohort_value(cohort_ids),
            affiliation or DEFAULT_AFFILIATION,
            reactivate,
        )
        if row is None:
            _inactive_emails.add(normalized_email)
            raise InactiveUserError("user is inactive")
        user = _managed_user_from_row(row)
        _memory_upsert_managed_user(
            email=user.email,
            display_name=user.display_name,
            role=user.role,
            cohort_ids=user.cohort_ids,
            user_id=user.user_id,
            affiliation=user.affiliation,
            reactivate=True,
        )
        return user
    except InactiveUserError:
        raise
    except Exception:
        require_runtime_fallback_allowed("managed user")
        return _memory_upsert_managed_user(
            email=normalized_email,
            display_name=display_name,
            role=role,
            cohort_ids=cohort_ids,
            user_id=user_id or user_id_from_external_id(normalized_external_id),
            affiliation=affiliation,
            reactivate=reactivate,
        )


async def get_managed_user(user_id: str) -> ManagedUser | None:
    try:
        pool = get_pool()
        async with pool.acquire() as conn:
            row = await conn.fetchrow(
                """
                SELECT user_id, email, display_name, role, cohort, affiliation, created_at, last_seen_at
                FROM app.app_user
                WHERE user_id = $1::uuid AND is_active
                """,
                user_id,
            )
            if row is not None:
                return _managed_user_from_row(row)
    except Exception:
        require_runtime_fallback_allowed("managed user")
    if not runtime_fallback_allowed():
        return None
    return _users.get(user_id)


async def list_managed_users() -> tuple[list[ManagedUser], bool]:
    try:
        pool = get_pool()
        async with pool.acquire() as conn:
            rows = await conn.fetch(
                """
                SELECT user_id, email, display_name, role, cohort, affiliation, created_at, last_seen_at
                FROM app.app_user
                WHERE is_active
                ORDER BY last_seen_at DESC
                """
            )
            return [_managed_user_from_row(row) for row in rows], True
    except Exception:
        require_runtime_fallback_allowed("managed user list")
    return sorted(_users.values(), key=lambda u: u.last_seen_at, reverse=True), False


async def update_managed_user(
    user_id: str,
    *,
    display_name: str | None = None,
    role: RoleName | None = None,
    affiliation: str | None = None,
    cohort_ids: list[str] | None = None,
) -> ManagedUser | None:
    try:
        pool = get_pool()
        async with pool.acquire() as conn:
            row = await conn.fetchrow(
                """
                UPDATE app.app_user SET
                  display_name = COALESCE($2, display_name),
                  role = COALESCE($3, role),
                  cohort = CASE WHEN $4 THEN $5 ELSE cohort END,
                  affiliation = COALESCE($6, affiliation),
                  updated_at = now(),
                  last_seen_at = now()
                WHERE user_id = $1::uuid AND is_active
                RETURNING user_id, email, display_name, role, cohort, affiliation, created_at, last_seen_at
                """,
                user_id,
                display_name.strip() if display_name is not None else None,
                _db_role(role) if role is not None else None,
                cohort_ids is not None,
                _cohort_value(cohort_ids),
                affiliation.strip() if affiliation is not None else None,
            )
            if row is not None:
                next_user = _managed_user_from_row(row)
                _memory_upsert_managed_user(
                    email=next_user.email,
                    display_name=next_user.display_name,
                    role=next_user.role,
                    cohort_ids=next_user.cohort_ids,
                    user_id=next_user.user_id,
                    affiliation=next_user.affiliation,
                )
                for session in _sessions.values():
                    if session.user_id == user_id:
                        session.display_name = next_user.display_name
                        session.role = next_user.role
                        session.cohort_ids = list(next_user.cohort_ids)
                return next_user
    except Exception:
        require_runtime_fallback_allowed("managed user update")

    if not runtime_fallback_allowed():
        return None
    current = _users.get(user_id)
    if current is None:
        return None

    next_user = ManagedUser(
        user_id=current.user_id,
        email=current.email,
        display_name=display_name.strip() if display_name is not None else current.display_name,
        role=role if role is not None else current.role,
        cohort_ids=list(cohort_ids) if cohort_ids is not None else current.cohort_ids,
        affiliation=affiliation.strip() if affiliation is not None else current.affiliation,
        created_at=current.created_at,
        last_seen_at=time.time(),
    )
    _users[user_id] = next_user

    for session in _sessions.values():
        if session.user_id == user_id:
            session.display_name = next_user.display_name
            session.role = next_user.role
            session.cohort_ids = list(next_user.cohort_ids)
    return next_user


async def deactivate_managed_user(user_id: str) -> bool:
    """Deactivate a managed user and revoke their active browser sessions."""
    changed = False
    inactive_email: str | None = None
    try:
        pool = get_pool()
        async with pool.acquire() as conn:
            row = await conn.fetchrow(
                "SELECT email FROM app.app_user WHERE user_id = $1::uuid",
                user_id,
            )
            if row is not None and row["email"]:
                inactive_email = _normalize_email(row["email"])
            result = await conn.execute(
                """
                UPDATE app.app_user SET
                  is_active = FALSE,
                  updated_at = now(),
                  last_seen_at = now()
                WHERE user_id = $1::uuid AND is_active
                """,
                user_id,
            )
            changed = result.endswith(" 1")
            await conn.execute(
                "UPDATE app.auth_session SET revoked_at = now() WHERE user_id = $1::uuid",
                user_id,
            )
    except Exception:
        require_runtime_fallback_allowed("managed user deactivation")

    if not runtime_fallback_allowed():
        return changed
    memory_user = _users.pop(user_id, None)
    if memory_user is not None:
        inactive_email = _normalize_email(memory_user.email)
    if inactive_email:
        _inactive_emails.add(inactive_email)
    for email, uid in list(_email_index.items()):
        if uid == user_id:
            _email_index.pop(email, None)
    for sid_hash, session in list(_sessions.items()):
        if session.user_id == user_id:
            _sessions.pop(sid_hash, None)
    return changed or memory_user is not None


async def active_session_count(user_id: str) -> int:
    try:
        pool = get_pool()
        async with pool.acquire() as conn:
            return int(
                await conn.fetchval(
                    """
                    SELECT count(*)
                    FROM app.auth_session
                    WHERE user_id = $1::uuid
                      AND revoked_at IS NULL
                      AND expires_at > now()
                    """,
                    user_id,
                )
                or 0
            )
    except Exception:
        require_runtime_fallback_allowed("auth session count")
    now = time.time()
    count = 0
    for session in _sessions.values():
        if session.user_id == user_id and session.expires_at > now:
            count += 1
    return count


async def create_session(
    *,
    email: str,
    display_name: str,
    role: str,
    cohort_ids: list[str] | None = None,
    user_id: str | None = None,
    external_id: str | None = None,
) -> tuple[str, SessionUser]:
    raw_sid = secrets.token_urlsafe(32)
    normalized_email = _normalize_email(email)
    managed = await upsert_managed_user(
        email=normalized_email,
        display_name=display_name,
        role=role,
        cohort_ids=cohort_ids,
        user_id=user_id,
        external_id=external_id,
        reactivate=False,
    )
    expires_at = time.time() + settings.session_ttl_seconds
    user = SessionUser(
        user_id=managed.user_id,
        email=normalized_email,
        display_name=managed.display_name,
        role=managed.role,
        cohort_ids=list(managed.cohort_ids),
        expires_at=expires_at,
    )
    sid_hash = _sid_hash(raw_sid)
    try:
        pool = get_pool()
        async with pool.acquire() as conn:
            await conn.execute(
                """
                INSERT INTO app.auth_session (
                  sid_hash, user_id, role, display_name, cohort_ids, expires_at, last_seen_at
                )
                VALUES ($1, $2::uuid, $3, $4, $5::jsonb, $6, now())
                ON CONFLICT (sid_hash) DO UPDATE SET
                  user_id = EXCLUDED.user_id,
                  role = EXCLUDED.role,
                  display_name = EXCLUDED.display_name,
                  cohort_ids = EXCLUDED.cohort_ids,
                  expires_at = EXCLUDED.expires_at,
                  revoked_at = NULL,
                  last_seen_at = now()
                """,
                sid_hash,
                managed.user_id,
                managed.role,
                managed.display_name,
                list(managed.cohort_ids),
                datetime.fromtimestamp(expires_at, tz=timezone.utc),
            )
    except Exception:
        require_runtime_fallback_allowed("browser session")
    _sessions[sid_hash] = user
    return raw_sid, user


async def get_session(raw_sid: str | None) -> SessionUser | None:
    if not raw_sid:
        return None
    key = _sid_hash(raw_sid)
    try:
        pool = get_pool()
        async with pool.acquire() as conn:
            row = await conn.fetchrow(
                """
                SELECT
                  s.expires_at,
                  u.user_id,
                  u.email,
                  COALESCE(u.display_name, s.display_name, u.email) AS display_name,
                  u.role,
                  u.cohort
                FROM app.auth_session s
                JOIN app.app_user u ON u.user_id = s.user_id
                WHERE s.sid_hash = $1
                  AND s.revoked_at IS NULL
                  AND s.expires_at > now()
                  AND u.is_active
                """,
                key,
            )
            if row is not None:
                await conn.execute(
                    "UPDATE app.auth_session SET last_seen_at = now() WHERE sid_hash = $1",
                    key,
                )
                await conn.execute(
                    "UPDATE app.app_user SET last_seen_at = now() WHERE user_id = $1",
                    row["user_id"],
                )
                return SessionUser(
                    user_id=str(row["user_id"]),
                    email=row["email"],
                    display_name=row["display_name"],
                    role=_app_role(row["role"]),
                    cohort_ids=_cohort_ids(row["cohort"]),
                    expires_at=_ts(row["expires_at"]),
                )
    except Exception:
        require_runtime_fallback_allowed("browser session")

    if not runtime_fallback_allowed():
        return None
    user = _sessions.get(key)
    if user is None:
        return None
    if user.expires_at <= time.time():
        _sessions.pop(key, None)
        return None
    managed = _users.get(user.user_id)
    if managed is not None:
        managed.last_seen_at = time.time()
    return user


async def revoke_session(raw_sid: str | None) -> None:
    if not raw_sid:
        return
    key = _sid_hash(raw_sid)
    _sessions.pop(key, None)
    try:
        pool = get_pool()
        async with pool.acquire() as conn:
            await conn.execute(
                "UPDATE app.auth_session SET revoked_at = now() WHERE sid_hash = $1",
                key,
            )
    except Exception:
        require_runtime_fallback_allowed("browser session revoke")