MobilityOps/backend/app/api/routers/demo.py

from __future__ import annotations

import threading
import time
import uuid
from datetime import UTC, datetime

from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.api.deps import get_current_user, get_db, require_operations_manager
from app.core.config import get_settings
from app.core.db import begin_exclusive_demo_reset, end_exclusive_demo_reset, engine
from app.core.security import SessionPayload, create_session_token, read_session_token
from app.models.audit import AuditEvent
from app.models.user import User
from app.schemas import CurrentUser, DemoLoginRequest, DemoManifestOut
from app.seed_loader import reset_and_seed
from app.services.audit import record_audit_event
from app.services.demo_manifest import build_demo_manifest, scenario_integrity_report
from app.services.sessions import revoke_session

router = APIRouter(prefix="/api/v1/demo", tags=["demo"])
settings = get_settings()
_reset_guard = threading.Lock()
_RESET_ADVISORY_LOCK_ID = 706_533_149


@router.get("/manifest", response_model=DemoManifestOut)
def demo_manifest(db: Session = Depends(get_db)) -> DemoManifestOut:
    if not settings.mobilityops_demo_mode:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Demo mode is disabled")
    # Deliberately unauthenticated: the demo-entry screen and the permanent demo badge
    # both need this before any session exists. Nothing here is sensitive — it's the same
    # honest "what is this demo" summary a logged-in user would see.
    return build_demo_manifest(db)


@router.post("/login", response_model=CurrentUser)
def demo_login(
    body: DemoLoginRequest, response: Response, db: Session = Depends(get_db)
) -> CurrentUser:
    if not settings.mobilityops_demo_mode:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Demo mode is disabled")
    public_ref = "USR-OPS" if body.role == "operations_manager" else "USR-EMP"
    user = db.scalar(select(User).where(User.public_ref == public_ref))
    if user is None:
        raise LookupError("Demo users are missing; run the seed loader first.")

    token = create_session_token(
        SessionPayload(
            user_id=str(user.id),
            public_ref=user.public_ref,
            role=user.role,
            display_name=user.display_name,
            issued_at=int(time.time()),
            session_id=str(uuid.uuid4()),
        )
    )
    response.set_cookie(
        settings.session_cookie_name,
        token,
        httponly=True,
        samesite="lax",
        secure=settings.session_cookie_secure,
        max_age=settings.session_ttl_seconds,
    )
    record_audit_event(
        db,
        actor_type="user",
        actor_id=user.id,
        actor_label=user.display_name,
        action="demo_login",
        entity_type="user",
        entity_id=user.id,
    )
    db.commit()
    return CurrentUser(public_ref=user.public_ref, display_name=user.display_name, role=body.role)


@router.get("/session", response_model=CurrentUser)
def get_session(response: Response, user: CurrentUser = Depends(get_current_user)) -> CurrentUser:
    # Never let the browser (or an intermediary) cache an authentication check — a stale
    # cached 200 here would keep showing a logged-out browser as authenticated.
    response.headers["Cache-Control"] = "no-store"
    return user


@router.post("/logout")
def demo_logout(request: Request, response: Response, db: Session = Depends(get_db)) -> dict:
    token = request.cookies.get(settings.session_cookie_name)
    payload = read_session_token(token) if token else None
    if payload is not None and token is not None:
        revoke_session(db, token, payload)
        record_audit_event(
            db,
            actor_type="user",
            actor_id=uuid.UUID(payload.user_id),
            actor_label=payload.display_name,
            action="demo_logout",
            entity_type="user",
        )
        db.commit()
    response.delete_cookie(settings.session_cookie_name)
    return {"status": "logged_out"}


@router.post("/reset")
def demo_reset(
    response: Response,
    db: Session = Depends(get_db),
    user: CurrentUser = Depends(require_operations_manager),
) -> dict:
    if not settings.mobilityops_demo_mode:
        # Outside demo mode the reset endpoint must not exist at all: it wipes
        # operational data and replaces it with synthetic records.
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Demo mode is disabled")
    if not settings.demo_allow_reset:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Demo reset is disabled on this deployment.",
        )
    if not _reset_guard.acquire(blocking=False):
        raise HTTPException(status_code=409, detail="A demo reset is already running.")
    replica_lock_connection = None
    try:
        # A session-level lock on its own connection rejects another replica immediately;
        # the main DB session can then safely end its auth read transaction and wait on
        # the normal shared/exclusive data barrier without releasing this replica guard.
        replica_lock_connection = engine.connect()
        locked = replica_lock_connection.scalar(
            select(func.pg_try_advisory_lock(_RESET_ADVISORY_LOCK_ID))
        )
        if not locked:
            raise HTTPException(status_code=409, detail="A demo reset is already running.")
        begin_exclusive_demo_reset(db)
        # The audit timestamp is shared by every replica. A process-local monotonic
        # timestamp cannot protect a multi-replica deployment.
        last_reset_at = db.scalar(
            select(AuditEvent.occurred_at)
            .where(AuditEvent.action == "demo_reset")
            .order_by(AuditEvent.occurred_at.desc())
            .limit(1)
        )
        elapsed = (datetime.now(UTC) - last_reset_at).total_seconds() if last_reset_at else None
        if elapsed is not None and elapsed < settings.demo_reset_cooldown_seconds:
            retry_after = max(1, int(settings.demo_reset_cooldown_seconds - elapsed + 0.999))
            raise HTTPException(
                status_code=429,
                detail=f"Demo reset is cooling down. Retry in {retry_after} seconds.",
                headers={"Retry-After": str(retry_after)},
            )
        result = reset_and_seed(db, preserve_integration_telemetry=True, commit=False)
        integrity = scenario_integrity_report(db)
        record_audit_event(
            db,
            actor_type="user",
            actor_label=user.display_name,
            action="demo_reset",
            entity_type="system",
            metadata={
                "counts": result.counts,
                "anchor_date": result.anchor_date.isoformat(),
                "scenario_integrity": integrity,
            },
        )
        db.commit()
        end_exclusive_demo_reset(db)
    finally:
        if replica_lock_connection is not None:
            try:
                replica_lock_connection.scalar(
                    select(func.pg_advisory_unlock(_RESET_ADVISORY_LOCK_ID))
                )
            finally:
                replica_lock_connection.close()
        _reset_guard.release()
    response.delete_cookie(settings.session_cookie_name)
    return {
        "status": "reset",
        "counts": result.counts,
        "anchor_date": result.anchor_date.isoformat(),
        "seeded_at": result.seeded_at.isoformat(),
        "scenario_integrity": integrity,
    }