MotorSignal-App/backend/motorsignal_api/app/main.py

import uuid
import logging
import os
import smtplib
import secrets
import hashlib
import time
from datetime import datetime, timedelta, timezone
from email.message import EmailMessage
from typing import Any, Literal, Optional

from email_validator import EmailNotValidError, validate_email
from fastapi import BackgroundTasks, Body, Depends, FastAPI, Header, HTTPException, Query, Request, status
from pydantic import BaseModel, ConfigDict, Field, field_validator
from sqlalchemy import func, or_, select, text
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session as DbSession
from sqlalchemy.orm.attributes import flag_modified
import copy

from auth import (
    create_access_token,
    create_refresh_token,
    create_one_time_token,
    decode_access_token,
    decode_firebase_id_token,
    hash_password,
    hash_refresh_token,
    hash_one_time_token,
    normalize_email,
    refresh_expiry,
    verify_password,
)
from db import check_database, engine
from models import (
    OpsAction,
    AuditEvent,
    AuthRateLimit,
    EmailVerificationToken,
    Entitlement,
    Operator,
    OpsAuditEvent,
    OpsIncident,
    OpsIncidentNote,
    OpsSession,
    OpsSupportTicket,
    OpsSupportTicketNote,
    OpsCmsDocument,
    OpsCmsRevision,
    PasswordResetToken,
    Ride,
    Session,
    Subscription,
    TelemetryPoint,
    User,
    Vehicle,
)
from infrastructure_metrics import get_infrastructure_metrics, get_recent_php_probes, run_network_speedtest
import ops_actions_service



logger = logging.getLogger(__name__)


app = FastAPI(
    title="MotorSignal API",
    version="1.0.0",
    docs_url=None,
    redoc_url=None,
    openapi_url=None,
)

from road_alerts import road_alerts_router
app.include_router(road_alerts_router, prefix="/api/v1/alerts", tags=["Road Alerts"])


@app.on_event("startup")
def startup_reconcile_actions():
    try:
        with DbSession(engine) as db:
            ops_actions_service.reconcile_stale_actions(db)
    except Exception as exc:
        logger.warning("Could not reconcile stale ops actions on startup: %s", exc)



class RegisterRequest(BaseModel):
    email: str
    password: str = Field(min_length=12, max_length=128)
    device_name: str | None = Field(default=None, max_length=255)


class LoginRequest(BaseModel):
    email: str
    password: str
    device_name: str | None = Field(default=None, max_length=255)


class RefreshRequest(BaseModel):
    refresh_token: str


class EmailRequest(BaseModel):
    email: str


class TokenRequest(BaseModel):
    token: str = Field(min_length=20, max_length=256)


class FirebaseLinkRequest(BaseModel):
    id_token: str = Field(min_length=100, max_length=8192)


class PasswordResetRequest(TokenRequest):
    password: str = Field(min_length=12, max_length=128)


class TokenResponse(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str = "bearer"
    ops_token: str | None = None



class SessionResponse(BaseModel):
    id: uuid.UUID
    device_name: str | None
    created_at: datetime
    expires_at: datetime
    last_used_at: datetime | None
    current: bool


class VehicleUpdate(BaseModel):
    nickname: str | None = Field(default=None, min_length=1, max_length=120)
    make: str | None = Field(default=None, min_length=1, max_length=120)
    model: str | None = Field(default=None, min_length=1, max_length=120)
    year: int | None = Field(default=None, ge=1885, le=2100)
    vin: str | None = Field(default=None, min_length=11, max_length=17)


class UserVehicleCreate(BaseModel):
    nickname: str = Field(min_length=1, max_length=120)
    make: str = Field(min_length=1, max_length=120)
    model: str = Field(min_length=1, max_length=120)
    year: int | None = Field(default=None, ge=1885, le=2100)
    vin: str | None = Field(default=None, min_length=11, max_length=17)


class UserRideCreate(BaseModel):
    vehicle_id: uuid.UUID | None = None
    started_at: datetime
    ended_at: datetime | None = None
    distance_miles: float | None = Field(default=None, ge=0)
    top_speed_mph: float | None = Field(default=None, ge=0)
    avg_speed_mph: float | None = Field(default=None, ge=0)
    status: str = Field(default="completed", min_length=1, max_length=32)


class TelemetryPointInput(BaseModel):
    recorded_at: datetime
    latitude: float | None = Field(default=None, ge=-90, le=90)
    longitude: float | None = Field(default=None, ge=-180, le=180)
    speed_mph: float | None = Field(default=None, ge=0)
    lean_angle_deg: float | None = Field(default=None, ge=-90, le=90)
    g_force: float | None = Field(default=None, ge=0)
    accel_g: float | None = Field(default=None, ge=-10, le=10)
    brake_g: float | None = Field(default=None, ge=-10, le=10)


class TelemetryBatch(BaseModel):
    points: list[TelemetryPointInput] = Field(min_length=1, max_length=1000)


def get_db():
    db = DbSession(engine)

    try:
        yield db
    finally:
        db.close()


def enforce_auth_rate_limit(request: Request, db: DbSession, action: str, limit: int = 10, window_seconds: int = 900) -> None:
    client = request.client.host if request.client else "unknown"
    key = f"{action}:{client}"
    now = datetime.now(timezone.utc)
    bucket = db.get(AuthRateLimit, key, with_for_update=True)
    if bucket is None:
        db.add(AuthRateLimit(key=key, window_started=now, attempts=1))
        db.commit()
        return
    elapsed = (now - bucket.window_started).total_seconds()
    if elapsed >= window_seconds:
        bucket.window_started = now
        bucket.attempts = 1
    elif bucket.attempts >= limit:
        retry_after = max(1, int(window_seconds - elapsed))
        raise HTTPException(status_code=429, detail="Too many authentication attempts", headers={"Retry-After": str(retry_after)})
    else:
        bucket.attempts += 1
    db.commit()


def extract_bearer_token(authorization: str | None) -> str:
    if not authorization:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Missing authorization token",
        )

    scheme, _, token = authorization.partition(" ")

    if scheme.lower() != "bearer" or not token:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authorization header",
        )

    return token


def get_request_context(request: Request) -> tuple[str | None, str | None]:
    return request.client.host if request.client else None, request.headers.get("user-agent")


def audit(
    db: DbSession,
    event_type: str,
    actor_user_id: uuid.UUID | None = None,
    target_user_id: uuid.UUID | None = None,
    request: Request | None = None,
    metadata: dict | None = None,
) -> None:
    ip_address = request.client.host if request and request.client else None
    user_agent = request.headers.get("user-agent")[:512] if request and request.headers.get("user-agent") else None
    db.add(AuditEvent(event_type=event_type, actor_user_id=actor_user_id, target_user_id=target_user_id, ip_address=ip_address, user_agent=user_agent, metadata_json=metadata))


def send_email(recipient: str, subject: str, body: str) -> None:
    host = os.getenv("SMTP_HOST")
    if not host:
        return
    try:
        message = EmailMessage()
        message["From"] = os.environ["SMTP_FROM"]
        message["To"] = recipient
        message["Subject"] = subject
        message.set_content(body)
        port = int(os.getenv("SMTP_PORT", "587"))
        with smtplib.SMTP(host, port, timeout=10) as smtp:
            if os.getenv("SMTP_STARTTLS", "true").lower() == "true":
                smtp.starttls()
            if os.getenv("SMTP_USER"):
                smtp.login(os.environ["SMTP_USER"], os.environ["SMTP_PASSWORD"])
            smtp.send_message(message)
    except (KeyError, OSError, smtplib.SMTPException):
        logger.exception("MotorSignal email delivery failed")


def get_current_identity(
    authorization: str | None = Header(default=None),
    db: DbSession = Depends(get_db),
) -> tuple[User, Session]:
    token = extract_bearer_token(authorization)

    try:
        payload = decode_access_token(token)
    except Exception:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or expired access token",
        )

    if payload.get("type") != "access":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid access token",
        )

    try:
        user_id = uuid.UUID(payload["sub"])
        session_id = uuid.UUID(payload["sid"])
    except Exception:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid access token",
        )

    session = db.get(Session, session_id)

    if (
        session is None
        or session.user_id != user_id
        or session.revoked_at is not None
        or session.expires_at <= datetime.now(timezone.utc)
    ):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Session is no longer valid",
        )

    user = db.get(User, user_id)

    if user is None or not user.is_active:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="User is unavailable",
        )

    return user, session


def get_current_user(identity: tuple[User, Session] = Depends(get_current_identity)) -> User:
    return identity[0]



@app.get("/api/v1/health")
def health():
    return {
        "status": "ok",
        "service": "motorsignal-api",
    }


@app.get("/api/v1/ready")
def readiness():
    try:
        check_database()

        return {
            "status": "ready",
            "database": "ok",
        }
    except Exception:
        raise HTTPException(
            status_code=503,
            detail="Service unavailable",
        )


@app.post(
    "/api/v1/auth/register",
    response_model=TokenResponse,
    status_code=201,
)
def register(
    request: RegisterRequest,
    http_request: Request,
    db: DbSession = Depends(get_db),
):
    enforce_auth_rate_limit(http_request, db, "register")
    try:
        validated = validate_email(
            request.email,
            check_deliverability=False,
        )
        email = normalize_email(validated.normalized)

    except EmailNotValidError:
        raise HTTPException(
            status_code=400,
            detail="Invalid email address",
        )

    existing = db.scalar(
        select(User).where(User.email == email)
    )

    if existing is not None:
        raise HTTPException(
            status_code=409,
            detail="Account already exists",
        )

    user = User(
        email=email,
        password_hash=hash_password(request.password),
    )

    db.add(user)
    db.flush()

    raw_refresh, refresh_hash = create_refresh_token()

    session = Session(
        user_id=user.id,
        refresh_token_hash=refresh_hash,
        device_name=request.device_name,
        expires_at=refresh_expiry(),
    )

    db.add(session)
    verification_raw, verification_hash = create_one_time_token()
    db.add(EmailVerificationToken(user_id=user.id, token_hash=verification_hash, expires_at=datetime.now(timezone.utc) + timedelta(hours=24)))
    audit(db, "auth.register", actor_user_id=user.id, target_user_id=user.id)
    db.commit()

    web_base = os.getenv("WEB_BASE_URL", "https://motorsignal.app")
    send_email(email, "Verify your MotorSignal email", f"Verify your email: {web_base}/verify-email?token={verification_raw}")

    return TokenResponse(
        access_token=create_access_token(user.id, session.id),
        refresh_token=raw_refresh,
    )


@app.post(
    "/api/v1/auth/login",
    response_model=TokenResponse,
)
def login(
    request: LoginRequest,
    http_request: Request,
    db: DbSession = Depends(get_db),
):
    enforce_auth_rate_limit(http_request, db, "login")
    email = normalize_email(request.email)

    user = db.scalar(
        select(User).where(User.email == email)
    )

    if user is None or not verify_password(
        request.password,
        user.password_hash,
    ):
        raise HTTPException(
            status_code=401,
            detail="Invalid email or password",
        )

    if not user.is_active:
        raise HTTPException(
            status_code=403,
            detail="Account disabled",
        )

    raw_refresh, refresh_hash = create_refresh_token()

    session = Session(
        user_id=user.id,
        refresh_token_hash=refresh_hash,
        device_name=request.device_name,
        expires_at=refresh_expiry(),
    )

    db.add(session)
    audit(db, "auth.login", actor_user_id=user.id, target_user_id=user.id)
    db.commit()

    ops_token: str | None = None
    stmt_op = select(Operator).where(
        Operator.identity == email,
        Operator.status == "active",
    )
    operator = db.scalar(stmt_op)

    if operator is not None:
        raw_ops_token = secrets.token_urlsafe(32)
        token_hash = hashlib.sha256(raw_ops_token.encode("utf-8")).hexdigest()
        now = datetime.now(timezone.utc)
        expires = now + timedelta(hours=8)
        auth_ver = getattr(operator, "authority_version", 1) or 1
        db.execute(text("""
            INSERT INTO ops.sessions (
                id, operator_id, token_digest, authority_version,
                firebase_auth_time, created_at, last_seen_at, absolute_expires_at,
                idle_timeout_seconds, session_token_hash, expires_at, last_activity_at
            ) VALUES (
                gen_random_uuid(), :op_id, :digest, :ver,
                :auth_time, :now, :now, :expires,
                900, :token_hash, :expires, :now
            )
        """), {
            "op_id": operator.id,
            "digest": token_hash,
            "ver": auth_ver,
            "auth_time": int(now.timestamp()),
            "now": now,
            "expires": expires,
            "token_hash": token_hash,
        })
        ops_audit(
            db=db,
            action="operator.login",
            actor_identity=operator.identity,
            actor_role=operator.role,
            target_resource="ops.session",
            result="success",
        )
        db.commit()
        ops_token = raw_ops_token

    return TokenResponse(
        access_token=create_access_token(user.id, session.id),
        refresh_token=raw_refresh,
        ops_token=ops_token,
    )



@app.post(
    "/api/v1/auth/refresh",
    response_model=TokenResponse,
)
def refresh(
    request: RefreshRequest,
    db: DbSession = Depends(get_db),
):
    old_hash = hash_refresh_token(request.refresh_token)

    session = db.scalar(
        select(Session).where(
            Session.refresh_token_hash == old_hash
        )
    )

    now = datetime.now(timezone.utc)

    if (
        session is None
        or session.revoked_at is not None
        or session.expires_at <= now
    ):
        raise HTTPException(
            status_code=401,
            detail="Invalid refresh token",
        )

    user = db.get(User, session.user_id)

    if user is None or not user.is_active:
        raise HTTPException(
            status_code=401,
            detail="Invalid refresh token",
        )

    new_raw, new_hash = create_refresh_token()

    session.refresh_token_hash = new_hash
    session.last_used_at = now
    session.expires_at = refresh_expiry()

    db.commit()

    return TokenResponse(
        access_token=create_access_token(user.id, session.id),
        refresh_token=new_raw,
    )


@app.post("/api/v1/auth/logout", status_code=204)
def logout(
    request: RefreshRequest,
    db: DbSession = Depends(get_db),
):
    refresh_hash = hash_refresh_token(request.refresh_token)

    session = db.scalar(
        select(Session).where(
            Session.refresh_token_hash == refresh_hash
        )
    )

    if session is not None and session.revoked_at is None:
        session.revoked_at = datetime.now(timezone.utc)
        audit(db, "auth.logout", actor_user_id=session.user_id, target_user_id=session.user_id)
        db.commit()

    return None


@app.get("/api/v1/me")
def me(user: User = Depends(get_current_user)):
    return {
        "id": str(user.id),
        "email": user.email,
        "email_verified": user.email_verified,
        "firebase_linked": user.firebase_uid is not None,
        "role": user.role,
        "created_at": user.created_at,
    }


@app.post("/api/v1/auth/firebase/link")
def link_firebase_identity(request: FirebaseLinkRequest, user: User = Depends(get_current_user), db: DbSession = Depends(get_db)):
    try:
        claims = decode_firebase_id_token(request.id_token)
    except Exception:
        raise HTTPException(status_code=401, detail="Invalid Firebase identity")
    firebase_email = claims.get("email")
    if claims.get("email_verified") is not True or not isinstance(firebase_email, str) or normalize_email(firebase_email) != user.email:
        raise HTTPException(status_code=403, detail="Firebase email must be verified and match this account")
    firebase_uid = claims["sub"]
    linked_user = db.scalar(select(User).where(User.firebase_uid == firebase_uid))
    if linked_user is not None and linked_user.id != user.id:
        raise HTTPException(status_code=409, detail="Firebase identity is already linked")
    if user.firebase_uid is not None and user.firebase_uid != firebase_uid:
        raise HTTPException(status_code=409, detail="Account is already linked to another Firebase identity")
    if user.firebase_uid is None:
        user.firebase_uid = firebase_uid
        user.email_verified = True
        audit(db, "auth.firebase_linked", actor_user_id=user.id, target_user_id=user.id)
        db.commit()
    return {"firebase_linked": True}


@app.post("/api/v1/auth/firebase/login", response_model=TokenResponse)
def login_with_firebase(request: FirebaseLinkRequest, db: DbSession = Depends(get_db)):
    try:
        claims = decode_firebase_id_token(request.id_token)
    except Exception:
        raise HTTPException(status_code=401, detail="Invalid Firebase identity token")

    firebase_email = claims.get("email")
    if not isinstance(firebase_email, str) or not firebase_email.strip():
        raise HTTPException(status_code=400, detail="Firebase identity missing email")

    email = normalize_email(firebase_email)
    firebase_uid = claims["sub"]

    user = db.scalar(select(User).where((User.firebase_uid == firebase_uid) | (User.email == email)))
    if user is None:
        random_pwd = secrets.token_urlsafe(32)
        user = User(
            email=email,
            password_hash=hash_password(random_pwd),
            email_verified=claims.get("email_verified", True),
            firebase_uid=firebase_uid,
            role="user",
        )
        db.add(user)
        db.flush()
    else:
        if user.firebase_uid is None:
            user.firebase_uid = firebase_uid
        if claims.get("email_verified") is True:
            user.email_verified = True

    if not user.is_active:
        raise HTTPException(status_code=403, detail="Account disabled")

    raw_refresh, refresh_hash = create_refresh_token()
    session = Session(
        user_id=user.id,
        refresh_token_hash=refresh_hash,
        device_name="Google Sign-In",
        expires_at=refresh_expiry(),
    )
    db.add(session)
    audit(db, "auth.firebase_login", actor_user_id=user.id, target_user_id=user.id)
    db.commit()

    ops_token: str | None = None
    operator = db.scalar(select(Operator).where(Operator.identity == email, Operator.status == "active"))
    if operator is not None:
        raw_ops_token = secrets.token_urlsafe(32)
        token_hash = hashlib.sha256(raw_ops_token.encode("utf-8")).hexdigest()
        now = datetime.now(timezone.utc)
        expires = now + timedelta(hours=8)
        auth_ver = getattr(operator, "authority_version", 1) or 1
        db.execute(text("""
            INSERT INTO ops.sessions (
                id, operator_id, token_digest, authority_version,
                firebase_auth_time, created_at, last_seen_at, absolute_expires_at,
                idle_timeout_seconds, session_token_hash, expires_at, last_activity_at
            ) VALUES (
                gen_random_uuid(), :op_id, :digest, :ver,
                :auth_time, :now, :now, :expires,
                900, :token_hash, :expires, :now
            )
        """), {
            "op_id": operator.id,
            "digest": token_hash,
            "ver": auth_ver,
            "auth_time": int(now.timestamp()),
            "now": now,
            "expires": expires,
            "token_hash": token_hash,
        })
        ops_audit(
            db=db,
            action="operator.firebase_login",
            actor_identity=operator.identity,
            actor_role=operator.role,
            target_resource="ops.session",
            result="success",
        )
        db.commit()
        ops_token = raw_ops_token

    return TokenResponse(
        access_token=create_access_token(user.id, session.id),
        refresh_token=raw_refresh,
        ops_token=ops_token,
    )



@app.post("/api/v1/auth/verify-email/request", status_code=202)
def request_email_verification(payload: EmailRequest, http_request: Request, db: DbSession = Depends(get_db)):
    enforce_auth_rate_limit(http_request, db, "verify-email-request")
    email = normalize_email(payload.email)
    user = db.scalar(select(User).where(User.email == email))
    if user is not None and not user.email_verified and user.is_active:
        raw, token_hash = create_one_time_token()
        db.add(EmailVerificationToken(user_id=user.id, token_hash=token_hash, expires_at=datetime.now(timezone.utc) + timedelta(hours=24)))
        audit(db, "auth.email_verification_requested", target_user_id=user.id)
        db.commit()
        web_base = os.getenv("WEB_BASE_URL", "https://motorsignal.app")
        send_email(email, "Verify your MotorSignal email", f"Verify your email: {web_base}/verify-email?token={raw}")
    return {"detail": "If the account can be verified, instructions have been sent."}


@app.post("/api/v1/auth/verify-email", status_code=204)
def verify_email(request: TokenRequest, http_request: Request, db: DbSession = Depends(get_db)):
    enforce_auth_rate_limit(http_request, db, "verify-email")
    token = db.scalar(select(EmailVerificationToken).where(EmailVerificationToken.token_hash == hash_one_time_token(request.token)))
    now = datetime.now(timezone.utc)
    if token is None or token.used_at is not None or token.expires_at <= now:
        raise HTTPException(status_code=400, detail="Invalid or expired verification token")
    user = db.get(User, token.user_id)
    if user is None:
        raise HTTPException(status_code=400, detail="Invalid verification token")
    user.email_verified = True
    token.used_at = now
    audit(db, "auth.email_verified", actor_user_id=user.id, target_user_id=user.id)
    db.commit()
    return None


@app.post("/api/v1/auth/password-reset/request", status_code=202)
def request_password_reset(payload: EmailRequest, http_request: Request, db: DbSession = Depends(get_db)):
    enforce_auth_rate_limit(http_request, db, "password-reset-request")
    email = normalize_email(payload.email)
    user = db.scalar(select(User).where(User.email == email))
    if user is not None and user.is_active:
        raw, token_hash = create_one_time_token()
        db.add(PasswordResetToken(user_id=user.id, token_hash=token_hash, expires_at=datetime.now(timezone.utc) + timedelta(hours=1)))
        audit(db, "auth.password_reset_requested", target_user_id=user.id)
        db.commit()
        web_base = os.getenv("WEB_BASE_URL", "https://motorsignal.app")
        send_email(email, "Reset your MotorSignal password", f"Reset your password: {web_base}/forgot-password?token={raw}")
    return {"detail": "If the account exists, reset instructions have been sent."}


@app.post("/api/v1/auth/password-reset", status_code=204)
def reset_password(request: PasswordResetRequest, http_request: Request, db: DbSession = Depends(get_db)):
    enforce_auth_rate_limit(http_request, db, "password-reset")
    token = db.scalar(select(PasswordResetToken).where(PasswordResetToken.token_hash == hash_one_time_token(request.token)))
    now = datetime.now(timezone.utc)
    if token is None or token.used_at is not None or token.expires_at <= now:
        raise HTTPException(status_code=400, detail="Invalid or expired reset token")
    user = db.get(User, token.user_id)
    if user is None or not user.is_active:
        raise HTTPException(status_code=400, detail="Invalid reset token")
    user.password_hash = hash_password(request.password)
    token.used_at = now
    db.query(Session).filter(Session.user_id == user.id, Session.revoked_at.is_(None)).update({Session.revoked_at: now}, synchronize_session=False)
    audit(db, "auth.password_reset_completed", actor_user_id=user.id, target_user_id=user.id)
    db.commit()
    return None


@app.get("/api/v1/auth/sessions", response_model=list[SessionResponse])
def list_sessions(identity: tuple[User, Session] = Depends(get_current_identity), db: DbSession = Depends(get_db)):
    user, current = identity
    sessions = db.scalars(select(Session).where(Session.user_id == user.id, Session.revoked_at.is_(None)).order_by(Session.created_at.desc())).all()
    return [SessionResponse(id=item.id, device_name=item.device_name, created_at=item.created_at, expires_at=item.expires_at, last_used_at=item.last_used_at, current=item.id == current.id) for item in sessions]


@app.delete("/api/v1/auth/sessions/{session_id}", status_code=204)
def revoke_session(session_id: uuid.UUID, identity: tuple[User, Session] = Depends(get_current_identity), db: DbSession = Depends(get_db)):
    user, _ = identity
    session = db.scalar(select(Session).where(Session.id == session_id, Session.user_id == user.id))
    if session is None:
        raise HTTPException(status_code=404, detail="Session not found")
    if session.revoked_at is None:
        session.revoked_at = datetime.now(timezone.utc)
        audit(db, "auth.session_revoked", actor_user_id=user.id, target_user_id=user.id, metadata={"session_id": str(session.id)})
        db.commit()
    return None


@app.post("/api/v1/auth/logout-all", status_code=204)
def logout_all(identity: tuple[User, Session] = Depends(get_current_identity), db: DbSession = Depends(get_db)):
    user, _ = identity
    now = datetime.now(timezone.utc)
    db.query(Session).filter(Session.user_id == user.id, Session.revoked_at.is_(None)).update({Session.revoked_at: now}, synchronize_session=False)
    audit(db, "auth.logout_all", actor_user_id=user.id, target_user_id=user.id)
    db.commit()
    return None


def vehicle_payload(vehicle: Vehicle) -> dict:
    return {"id": str(vehicle.id), "user_id": str(vehicle.user_id), "nickname": vehicle.nickname, "make": vehicle.make, "model": vehicle.model, "year": vehicle.year, "vin": vehicle.vin, "created_at": vehicle.created_at, "updated_at": vehicle.updated_at}


def ride_payload(ride: Ride) -> dict:
    return {"id": str(ride.id), "user_id": str(ride.user_id), "vehicle_id": str(ride.vehicle_id) if ride.vehicle_id else None, "started_at": ride.started_at, "ended_at": ride.ended_at, "distance_miles": ride.distance_miles, "top_speed_mph": ride.top_speed_mph, "avg_speed_mph": ride.avg_speed_mph, "status": ride.status, "created_at": ride.created_at}


@app.get("/api/v1/vehicles")
def list_my_vehicles(user: User = Depends(get_current_user), db: DbSession = Depends(get_db)):
    vehicles = db.scalars(select(Vehicle).where(Vehicle.user_id == user.id).order_by(Vehicle.created_at.desc())).all()
    return [vehicle_payload(vehicle) for vehicle in vehicles]


@app.post("/api/v1/vehicles", status_code=201)
def create_my_vehicle(request: UserVehicleCreate, user: User = Depends(get_current_user), db: DbSession = Depends(get_db)):
    vehicle = Vehicle(user_id=user.id, **request.model_dump())
    db.add(vehicle)
    db.flush()
    audit(db, "vehicle.created", actor_user_id=user.id, target_user_id=user.id, metadata={"vehicle_id": str(vehicle.id)})
    db.commit()
    db.refresh(vehicle)
    return vehicle_payload(vehicle)


@app.get("/api/v1/vehicles/{vehicle_id}")
def get_my_vehicle(vehicle_id: uuid.UUID, user: User = Depends(get_current_user), db: DbSession = Depends(get_db)):
    vehicle = db.scalar(select(Vehicle).where(Vehicle.id == vehicle_id, Vehicle.user_id == user.id))
    if vehicle is None:
        raise HTTPException(status_code=404, detail="Vehicle not found")
    return vehicle_payload(vehicle)


@app.patch("/api/v1/vehicles/{vehicle_id}")
def update_my_vehicle(vehicle_id: uuid.UUID, request: VehicleUpdate, user: User = Depends(get_current_user), db: DbSession = Depends(get_db)):
    vehicle = db.scalar(select(Vehicle).where(Vehicle.id == vehicle_id, Vehicle.user_id == user.id))
    if vehicle is None:
        raise HTTPException(status_code=404, detail="Vehicle not found")
    changes = request.model_dump(exclude_unset=True)
    for field, value in changes.items():
        setattr(vehicle, field, value)
    audit(db, "vehicle.updated", actor_user_id=user.id, target_user_id=user.id, metadata={"vehicle_id": str(vehicle.id), "fields": list(changes)})
    db.commit()
    db.refresh(vehicle)
    return vehicle_payload(vehicle)


@app.delete("/api/v1/vehicles/{vehicle_id}", status_code=204)
def delete_my_vehicle(vehicle_id: uuid.UUID, user: User = Depends(get_current_user), db: DbSession = Depends(get_db)):
    vehicle = db.scalar(select(Vehicle).where(Vehicle.id == vehicle_id, Vehicle.user_id == user.id))
    if vehicle is None:
        raise HTTPException(status_code=404, detail="Vehicle not found")
    audit(db, "vehicle.deleted", actor_user_id=user.id, target_user_id=user.id, metadata={"vehicle_id": str(vehicle.id)})
    db.delete(vehicle)
    db.commit()
    return None


@app.get("/api/v1/rides")
def list_my_rides(page: int = Query(default=1, ge=1), page_size: int = Query(default=50, ge=1, le=100), user: User = Depends(get_current_user), db: DbSession = Depends(get_db)):
    query = select(Ride).where(Ride.user_id == user.id)
    total = db.scalar(select(func.count()).select_from(Ride).where(Ride.user_id == user.id)) or 0
    rides = db.scalars(query.order_by(Ride.started_at.desc()).offset((page - 1) * page_size).limit(page_size)).all()
    return {"items": [ride_payload(ride) for ride in rides], "page": page, "page_size": page_size, "total": total}


@app.post("/api/v1/rides", status_code=201)
def create_my_ride(request: UserRideCreate, user: User = Depends(get_current_user), db: DbSession = Depends(get_db)):
    if request.vehicle_id is not None:
        vehicle = db.scalar(select(Vehicle).where(Vehicle.id == request.vehicle_id, Vehicle.user_id == user.id))
        if vehicle is None:
            raise HTTPException(status_code=400, detail="Vehicle not found")
    ride = Ride(user_id=user.id, **request.model_dump())
    db.add(ride)
    db.flush()
    audit(db, "ride.created", actor_user_id=user.id, target_user_id=user.id, metadata={"ride_id": str(ride.id)})
    db.commit()
    db.refresh(ride)
    return ride_payload(ride)


@app.get("/api/v1/rides/{ride_id}")
def get_my_ride(ride_id: uuid.UUID, user: User = Depends(get_current_user), db: DbSession = Depends(get_db)):
    ride = db.scalar(select(Ride).where(Ride.id == ride_id, Ride.user_id == user.id))
    if ride is None:
        raise HTTPException(status_code=404, detail="Ride not found")
    return ride_payload(ride)


def telemetry_payload(point: TelemetryPoint) -> dict:
    return {"id": str(point.id), "ride_id": str(point.ride_id), "recorded_at": point.recorded_at, "latitude": point.latitude, "longitude": point.longitude, "speed_mph": point.speed_mph, "lean_angle_deg": point.lean_angle_deg, "g_force": point.g_force, "accel_g": point.accel_g, "brake_g": point.brake_g}


def get_owned_ride(ride_id: uuid.UUID, user_id: uuid.UUID, db: DbSession) -> Ride:
    ride = db.scalar(select(Ride).where(Ride.id == ride_id, Ride.user_id == user_id))
    if ride is None:
        raise HTTPException(status_code=404, detail="Ride not found")
    return ride


@app.post("/api/v1/rides/{ride_id}/telemetry", status_code=201)
def ingest_telemetry(ride_id: uuid.UUID, request: TelemetryBatch, user: User = Depends(get_current_user), db: DbSession = Depends(get_db)):
    get_owned_ride(ride_id, user.id, db)
    points = [TelemetryPoint(ride_id=ride_id, **point.model_dump()) for point in request.points]
    db.add_all(points)
    audit(db, "telemetry.ingested", actor_user_id=user.id, target_user_id=user.id, metadata={"ride_id": str(ride_id), "point_count": len(points)})
    db.commit()
    return {"accepted": len(points)}


@app.get("/api/v1/rides/{ride_id}/telemetry")
def list_my_telemetry(ride_id: uuid.UUID, limit: int = Query(default=1000, ge=1, le=5000), user: User = Depends(get_current_user), db: DbSession = Depends(get_db)):
    get_owned_ride(ride_id, user.id, db)
    points = db.scalars(select(TelemetryPoint).where(TelemetryPoint.ride_id == ride_id).order_by(TelemetryPoint.recorded_at.asc()).limit(limit)).all()
    return {"items": [telemetry_payload(point) for point in points], "count": len(points)}


def telemetry_summary(ride_id: uuid.UUID, db: DbSession) -> dict:
    count, first_recorded, last_recorded, max_speed, max_lean, max_g = db.execute(
        select(func.count(TelemetryPoint.id), func.min(TelemetryPoint.recorded_at), func.max(TelemetryPoint.recorded_at), func.max(TelemetryPoint.speed_mph), func.max(func.abs(TelemetryPoint.lean_angle_deg)), func.max(TelemetryPoint.g_force)).where(TelemetryPoint.ride_id == ride_id)
    ).one()
    duration_seconds = (last_recorded - first_recorded).total_seconds() if first_recorded and last_recorded else 0
    return {"ride_id": str(ride_id), "point_count": count, "duration_seconds": duration_seconds, "max_speed_mph": max_speed, "max_lean_angle_deg": max_lean, "max_g_force": max_g}


@app.get("/api/v1/rides/{ride_id}/telemetry/summary")
def my_telemetry_summary(ride_id: uuid.UUID, user: User = Depends(get_current_user), db: DbSession = Depends(get_db)):
    get_owned_ride(ride_id, user.id, db)
    return telemetry_summary(ride_id, db)



def subscription_payload(subscription: Subscription) -> dict:
    return {"id": str(subscription.id), "user_id": str(subscription.user_id), "plan": subscription.plan, "status": subscription.status, "current_period_start": subscription.current_period_start, "current_period_end": subscription.current_period_end, "created_at": subscription.created_at, "updated_at": subscription.updated_at}


def entitlement_payload(entitlement: Entitlement) -> dict:
    return {"id": str(entitlement.id), "user_id": str(entitlement.user_id), "key": entitlement.key, "value": entitlement.value, "expires_at": entitlement.expires_at, "created_at": entitlement.created_at}


def get_or_create_subscription(user_id: uuid.UUID, db: DbSession) -> Subscription:
    subscription = db.scalar(select(Subscription).where(Subscription.user_id == user_id))
    if subscription is None:
        subscription = Subscription(user_id=user_id, plan="free", status="active")
        db.add(subscription)
        db.flush()
    return subscription


@app.get("/api/v1/me/subscription")
def get_my_subscription(user: User = Depends(get_current_user), db: DbSession = Depends(get_db)):
    subscription = get_or_create_subscription(user.id, db)
    db.commit()
    db.refresh(subscription)
    return subscription_payload(subscription)


@app.get("/api/v1/me/entitlements")
def list_my_entitlements(user: User = Depends(get_current_user), db: DbSession = Depends(get_db)):
    entitlements = db.scalars(select(Entitlement).where(Entitlement.user_id == user.id).order_by(Entitlement.created_at.desc())).all()
    return [entitlement_payload(item) for item in entitlements if item.expires_at is None or item.expires_at > datetime.now(timezone.utc)]



# ============================================================
# Operations (OPS-3B / OPS-3C / OPS-4) Platform Endpoints
# ============================================================

import hashlib


def get_current_operator(
    request: Request,
    db: DbSession = Depends(get_db),
) -> tuple[Operator, OpsSession]:
    auth_header = request.headers.get("authorization")
    token: str | None = None
    if auth_header:
        scheme, _, param = auth_header.partition(" ")
        if scheme.lower() == "bearer" and param:
            token = param
    if not token:
        token = request.cookies.get("__Host-motorsignal_ops") or request.cookies.get("motorsignal_ops")

    if not token:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Missing Operations session credentials",
        )

    token_hash = hashlib.sha256(token.encode("utf-8")).hexdigest()
    now = datetime.now(timezone.utc)

    stmt = (
        select(OpsSession, Operator)
        .join(Operator, OpsSession.operator_id == Operator.id)
        .where(
            OpsSession.session_token_hash == token_hash,
            OpsSession.revoked_at.is_(None),
            OpsSession.expires_at > now,
            Operator.status == "active",
        )
    )
    result = db.execute(stmt).first()
    if result is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Operations session expired or invalid",
        )

    ops_session, operator = result
    ops_session.last_activity_at = now
    db.commit()
    return operator, ops_session


def require_ops_roles(*roles: str):
    def dependency(
        auth: tuple[Operator, OpsSession] = Depends(get_current_operator),
    ) -> tuple[Operator, OpsSession]:
        operator, session = auth
        if operator.role not in roles:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Insufficient Operations authority",
            )
        return operator, session

    return dependency


def ops_audit(
    db: DbSession,
    action: str,
    actor_identity: str,
    actor_role: str | None,
    target_resource: str | None = None,
    reason_code: str | None = None,
    case_reference: str | None = None,
    result: str = "success",
    metadata: dict | None = None,
) -> OpsAuditEvent:
    event = OpsAuditEvent(
        timestamp=datetime.now(timezone.utc),
        action=action,
        actor_identity=actor_identity,
        actor_role=actor_role,
        target_resource=target_resource,
        reason_code=reason_code,
        case_reference=case_reference,
        result=result,
        metadata_json=metadata,
    )
    db.add(event)
    return event


def generate_incident_case_reference(db: DbSession) -> str:
    year = datetime.now(timezone.utc).year
    seq_val = db.execute(text("SELECT nextval('ops.incident_case_seq')")).scalar()
    return f"OPS-{year}-{int(seq_val):06d}"


class OpsIncidentCreate(BaseModel):
    model_config = ConfigDict(extra="forbid")
    title: str = Field(min_length=1, max_length=255)
    summary: str = Field(min_length=1, max_length=4000)
    severity: str = Field(default="medium")


class OpsIncidentPatch(BaseModel):
    model_config = ConfigDict(extra="forbid")
    severity: str | None = None
    status: str | None = None
    resolution: str | None = Field(default=None, max_length=4000)
    assigned_to: str | None = Field(default=None, max_length=320)


class OpsIncidentNoteCreate(BaseModel):
    model_config = ConfigDict(extra="forbid")
    body: str = Field(min_length=1, max_length=4000)


class OpsUserStatusUpdate(BaseModel):
    model_config = ConfigDict(extra="forbid")
    is_active: bool
    reason: str = Field(min_length=3)


class OpsUserActionRequest(BaseModel):
    model_config = ConfigDict(extra="forbid")
    reason: str = Field(min_length=3)


class OpsSupportDiagnosticBundle(BaseModel):
    model_config = ConfigDict(extra="ignore")
    app_version: str
    os_version: str
    device_model: str
    stack_trace: str | None = None
    sensor_status: str
    logs_snippet: str | None = None
    submitted_at: str



class OpsActionCreate(BaseModel):
    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
    action_type: Literal["deploy_website", "restart_website", "restart_api", "reboot_server"]
    idempotency_key: str | None = Field(default=None, min_length=8, max_length=128)
    confirmation_text: str | None = Field(default=None, max_length=64)
    confirmation: str | None = Field(default=None, max_length=64)

class OpsActionRead(BaseModel):
    id: uuid.UUID
    action_type: str
    requested_by_id: uuid.UUID
    request_timestamp: datetime
    start_timestamp: datetime | None
    completion_timestamp: datetime | None
    state: str
    status_message: str | None
    deployed_revision: str | None
    target_revision: str | None
    preflight_result: str | None
    health_result: str | None
    rollback_status: str | None
    failure_code: str | None
    idempotency_key: str | None

class OpsSupportTicketCreate(BaseModel):
    model_config = ConfigDict(extra="forbid")
    ticket_type: str = Field(default="general")
    reporter_email: str = Field(min_length=3, max_length=255)
    subject: str = Field(min_length=1, max_length=255)
    description: str = Field(min_length=1, max_length=10000)
    severity: str = Field(default="medium")
    diagnostic_bundle: OpsSupportDiagnosticBundle | dict | None = None

    @field_validator("reporter_email")
    @classmethod
    def validate_reporter_email(cls, v: str) -> str:
        s = v.strip()
        if len(s) < 3:
            raise ValueError("Reporter email must not be empty or whitespace only.")
        return s

    @field_validator("subject", "description")
    @classmethod
    def validate_non_whitespace(cls, v: str) -> str:
        s = v.strip()
        if not s:
            raise ValueError("Field cannot be empty or contain only whitespace.")
        return s


class OpsSupportTicketPatch(BaseModel):
    model_config = ConfigDict(extra="forbid")
    status: str | None = None
    severity: str | None = None


class OpsSupportTicketNoteCreate(BaseModel):
    model_config = ConfigDict(extra="forbid")
    body: str = Field(min_length=1, max_length=10000)


@app.get("/api/v1/ops/me")
def ops_get_me(
    auth: tuple[Operator, OpsSession] = Depends(get_current_operator),
):
    operator, session = auth
    return {
        "identity": operator.identity,
        "role": operator.role,
        "status": operator.status,
        "created_at": operator.created_at.isoformat(),
        "last_activity_at": session.last_activity_at.isoformat() if session.last_activity_at else None,
        "expires_at": session.expires_at.isoformat(),
    }


@app.get("/api/v1/ops/health")
def ops_get_health(db: DbSession = Depends(get_db)):
    now_iso = datetime.now(timezone.utc).isoformat()
    db_ok = False
    try:
        if db is not None:
            db.execute(text("SELECT 1"))
            db_ok = True
        else:
            db_ok = check_database()
    except Exception:
        db_ok = False

    missing_action_outcomes = 0
    if db_ok and db is not None:
        try:
            five_mins_ago = datetime.now(timezone.utc) - timedelta(minutes=5)
            missing_action_outcomes = db.scalar(
                select(func.count()).select_from(OpsAction)
                .where(
                    OpsAction.completion_timestamp.is_(None),
                    OpsAction.state.in_(["queued", "running"]),
                    OpsAction.request_timestamp < five_mins_ago,
                )
            ) or 0
        except Exception:
            missing_action_outcomes = 0

    audit_status = "degraded" if missing_action_outcomes > 0 else ("available" if db_ok else "degraded")
    overall = "available" if (db_ok and missing_action_outcomes == 0) else "degraded"
    return {
        "status": overall,
        "checked_at": now_iso,
        "components": {
            "operations_api": {
                "status": "available",
                "description": "FastAPI core endpoint runtime & ingress",
            },
            "operations_database": {
                "status": "available" if db_ok else "unavailable",
                "description": "PostgreSQL ops schema relational persistence",
            },
            "session_validation": {
                "status": "available" if db_ok else "degraded",
                "description": "Checks staff sessions and enforces owner/operator permissions.",
            },
            "audit_persistence": {
                "status": audit_status,
                "description": "Records administrative actions and their outcomes.",
            },
        },
    }


@app.get("/api/v1/ops/infrastructure")
def ops_get_infrastructure(
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    db_ok = False
    try:
        if db is not None:
            db.execute(text("SELECT 1"))
            db_ok = True
        else:
            db_ok = check_database()
    except Exception:
        db_ok = False

    return get_infrastructure_metrics(database_ok=db_ok)


@app.post("/api/v1/ops/infrastructure/speedtest")
def ops_run_speedtest(
    payload: Optional[dict[str, Any]] = Body(default=None),
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    operator, _ = auth
    if payload and payload.get("mode") == "client_to_server":
        ping_val = payload.get("latency_ms") or payload.get("ping_ms")
        dl_val = payload.get("download_mbps")
        ul_val = payload.get("upload_mbps")
        client_ip = payload.get("client_ip")
        ops_audit(
            db=db,
            action="infrastructure.speedtest.client",
            actor_identity=operator.identity,
            actor_role=operator.role,
            target_resource="host:network:client",
            reason_code="client_network_benchmark",
            result="success",
            metadata={
                "ping_ms": ping_val,
                "download_mbps": dl_val,
                "upload_mbps": ul_val,
                "client_ip": client_ip,
            },
        )
        db.commit()
        return {
            "status": "completed",
            "mode": "client_to_server",
            "latency_ms": ping_val,
            "download_mbps": dl_val,
            "upload_mbps": ul_val,
            "server_target": "MotorSignal Host Edge",
            "client_ip": client_ip,
            "direction_label": "Operator Browser ➔ MotorSignal Server",
            "tested_at": datetime.now(timezone.utc).isoformat(),
        }

    result = run_network_speedtest()
    result["mode"] = "server_to_external"
    result["direction_label"] = "MotorSignal Server ➔ Upstream Edge (1.1.1.1)"
    ops_audit(
        db=db,
        action="infrastructure.speedtest",
        actor_identity=operator.identity,
        actor_role=operator.role,
        target_resource="host:network",
        reason_code="network_benchmark",
        result="success" if result.get("status") == "completed" else "failure",
        metadata={
            "ping_ms": result.get("ping_ms"),
            "download_mbps": result.get("download_mbps"),
            "upload_mbps": result.get("upload_mbps"),
        },
    )
    db.commit()
    return result



@app.get("/api/v1/ops/overview")
def ops_get_overview(
    auth: tuple[Operator, OpsSession] = Depends(get_current_operator),
    db: DbSession = Depends(get_db),
):
    now = datetime.now(timezone.utc)
    t_24h = now - timedelta(hours=24)
    t_7d = now - timedelta(days=7)
    t_30d = now - timedelta(days=30)
    now_iso = now.isoformat()

    # Exclude staff Operations sessions from customer activity counts
    staff_emails = [op.identity for op in db.scalars(select(Operator)).all()]
    staff_user_ids = db.scalars(select(User.id).where(User.email.in_(staff_emails))).all() if staff_emails else []

    total_users = db.scalar(select(func.count()).select_from(User)) or 0
    new_users_24h = db.scalar(select(func.count()).select_from(User).where(User.created_at >= t_24h)) or 0
    new_users_7d = db.scalar(select(func.count()).select_from(User).where(User.created_at >= t_7d)) or 0
    new_users_30d = db.scalar(select(func.count()).select_from(User).where(User.created_at >= t_30d)) or 0

    def cust_sess_count(t_since):
        q = select(func.count(func.distinct(Session.user_id))).where(
            Session.last_used_at >= t_since,
            Session.revoked_at.is_(None),
        )
        if staff_user_ids:
            q = q.where(~Session.user_id.in_(staff_user_ids))
        return db.scalar(q) or 0

    active_users_24h = cust_sess_count(t_24h)
    active_users_7d = cust_sess_count(t_7d)
    active_users_30d = cust_sess_count(t_30d)

    total_subscribers = db.scalar(select(func.count()).select_from(Subscription)) or 0
    pro_members = db.scalar(
        select(func.count()).select_from(Subscription)
        .where(Subscription.plan.in_(["pro", "premium"]), Subscription.status == "active")
    ) or 0
    free_members = db.scalar(
        select(func.count()).select_from(Subscription)
        .where((Subscription.plan == "free") | (Subscription.status != "active"))
    ) or (total_users - pro_members if total_users > pro_members else 0)
    conversion_rate = round((pro_members / max(1, total_users)) * 100, 1)

    # Web customer sessions excluding staff
    cust_web_q = select(func.count()).select_from(Session).where(
        (Session.device_name.ilike("%web%") | Session.device_name.ilike("%browser%")),
        Session.last_used_at >= t_24h,
        Session.revoked_at.is_(None),
    )
    if staff_user_ids:
        cust_web_q = cust_web_q.where(~Session.user_id.in_(staff_user_ids))
    web_sessions_24h = db.scalar(cust_web_q) or 0

    cust_active_sess_q = select(func.count()).select_from(Session).where(
        Session.expires_at > now,
        Session.revoked_at.is_(None),
    )
    if staff_user_ids:
        cust_active_sess_q = cust_active_sess_q.where(~Session.user_id.in_(staff_user_ids))
    cust_active_sessions_now = db.scalar(cust_active_sess_q) or 0

    def cust_logins_count(t_since):
        q = select(func.count()).select_from(AuditEvent).where(
            AuditEvent.event_type.in_(["auth.login", "auth.login_success"]),
            AuditEvent.created_at >= t_since,
        )
        if staff_user_ids:
            q = q.where(~AuditEvent.actor_user_id.in_(staff_user_ids))
        return db.scalar(q) or 0

    web_logins_24h = cust_logins_count(t_24h)
    web_logins_7d = cust_logins_count(t_7d)
    web_logins_30d = cust_logins_count(t_30d)

    # Compact daily activity trends for the past 7 days
    daily_trends = []
    for day_offset in range(6, -1, -1):
        day_start = (now - timedelta(days=day_offset)).replace(hour=0, minute=0, second=0, microsecond=0)
        day_end = day_start + timedelta(days=1)
        day_u_q = select(func.count(func.distinct(Session.user_id))).where(
            Session.last_used_at >= day_start,
            Session.last_used_at < day_end,
            Session.revoked_at.is_(None),
        )
        if staff_user_ids:
            day_u_q = day_u_q.where(~Session.user_id.in_(staff_user_ids))
        day_users = db.scalar(day_u_q) or 0
        day_signins = db.scalar(
            select(func.count()).select_from(AuditEvent).where(
                AuditEvent.event_type.in_(["auth.login", "auth.login_success"]),
                AuditEvent.created_at >= day_start,
                AuditEvent.created_at < day_end,
            )
        ) or 0
        daily_trends.append({
            "date": day_start.strftime("%Y-%m-%d"),
            "active_users": day_users,
            "signins": day_signins,
        })

    mobile_sessions_24h = db.scalar(
        select(func.count()).select_from(Session)
        .where(
            ~Session.device_name.ilike("%web%"),
            ~Session.device_name.ilike("%browser%"),
            Session.last_used_at >= t_24h,
            Session.revoked_at.is_(None)
        )
    ) or 0
    mobile_rides_today = db.scalar(
        select(func.count()).select_from(Ride)
        .where(Ride.started_at >= t_24h)
    ) or 0
    mobile_rides_active = db.scalar(
        select(func.count()).select_from(Ride)
        .where(Ride.status.in_(["recording", "in_progress", "active"]))
    ) or 0
    mobile_datapoints_24h = db.scalar(
        select(func.count()).select_from(TelemetryPoint)
        .where(TelemetryPoint.recorded_at >= t_24h)
    ) or 0
    mobile_distance_24h = db.scalar(
        select(func.coalesce(func.sum(Ride.distance_miles), 0.0))
        .where(Ride.started_at >= t_24h)
    ) or 0.0
    mobile_top_speed_24h = db.scalar(
        select(func.coalesce(func.max(Ride.top_speed_mph), 0.0))
        .where(Ride.started_at >= t_24h)
    ) or 0.0
    mobile_avg_speed_24h = db.scalar(
        select(func.coalesce(func.avg(Ride.avg_speed_mph), 0.0))
        .where(Ride.started_at >= t_24h)
    ) or 0.0

    # Subsystems telemetry
    t0 = time.perf_counter()
    db_ok = False
    try:
        db.execute(text("SELECT 1"))
        db_ok = True
        db_latency_ms = round((time.perf_counter() - t0) * 1000, 1)
    except Exception:
        db_ok = False
        db_latency_ms = None

    latest_audit = db.scalars(select(OpsAuditEvent).order_by(OpsAuditEvent.timestamp.desc()).limit(1)).first()
    last_audit_write_iso = latest_audit.timestamp.isoformat() if latest_audit else None
    failed_audit_writes_24h = db.scalar(
        select(func.count()).select_from(OpsAuditEvent)
        .where(OpsAuditEvent.result.in_(["failure", "failed"]), OpsAuditEvent.timestamp >= t_24h)
    ) or 0

    five_mins_ago = now - timedelta(minutes=5)
    missing_action_outcomes = db.scalar(
        select(func.count()).select_from(OpsAction)
        .where(
            OpsAction.completion_timestamp.is_(None),
            OpsAction.state.in_(["queued", "running"]),
            OpsAction.request_timestamp < five_mins_ago,
        )
    ) or 0

    latest_ride = db.scalar(select(func.max(Ride.created_at)))
    latest_tp = db.scalar(select(func.max(TelemetryPoint.recorded_at)))
    last_upload_dt = max(filter(None, [latest_ride, latest_tp]), default=None)
    last_upload_iso = last_upload_dt.isoformat() if last_upload_dt else None

    failed_uploads_24h = db.scalar(
        select(func.count()).select_from(OpsAuditEvent)
        .where(
            OpsAuditEvent.action.in_(["ride.upload", "telemetry.upload", "upload"]),
            OpsAuditEvent.result.in_(["failure", "failed"]),
            OpsAuditEvent.timestamp >= t_24h,
        )
    ) or 0

    queued_actions_backlog = db.scalar(
        select(func.count()).select_from(OpsAction).where(OpsAction.state == "queued")
    )

    if not db_ok:
        api_status_label = "Unavailable"
    elif last_upload_dt is None or last_upload_dt < (now - timedelta(days=7)):
        api_status_label = "No recent activity"
    elif (queued_actions_backlog or 0) > 10:
        api_status_label = "Delayed"
    else:
        api_status_label = "Healthy"

    validation_errors_24h = db.scalar(
        select(func.count()).select_from(OpsAuditEvent)
        .where(
            OpsAuditEvent.action.ilike("%auth%"),
            OpsAuditEvent.result.in_(["denied", "failure", "failed"]),
            OpsAuditEvent.timestamp >= t_24h,
        )
    ) or 0

    # Active Incidents (Sorted: Critical/High first, then unassigned/stale)
    open_incidents = db.scalars(
        select(OpsIncident)
        .where(OpsIncident.status.in_(["open", "investigating", "monitoring"]))
    ).all()
    total_open = len(open_incidents)
    crit_count = sum(1 for i in open_incidents if i.severity == "critical")
    high_count = sum(1 for i in open_incidents if i.severity == "high")
    med_count = sum(1 for i in open_incidents if i.severity == "medium")
    low_count = sum(1 for i in open_incidents if i.severity == "low")
    unassigned_count = sum(1 for i in open_incidents if not getattr(i, "assigned_to", None))

    def incident_sort_key(inc):
        sev_order = 0 if inc.severity == "critical" else (1 if inc.severity == "high" else (2 if inc.severity == "medium" else 3))
        unassigned_order = 0 if not getattr(inc, "assigned_to", None) else 1
        ts = (inc.updated_at or inc.created_at).timestamp() if (inc.updated_at or inc.created_at) else 0
        return (sev_order, unassigned_order, ts)

    sorted_incidents = sorted(open_incidents, key=incident_sort_key)
    recent_open = [
        {
            "case_reference": inc.case_reference,
            "title": inc.title,
            "severity": inc.severity,
            "status": inc.status,
            "assigned_to": getattr(inc, "assigned_to", None),
            "created_at": inc.created_at.isoformat(),
            "updated_at": inc.updated_at.isoformat(),
        }
        for inc in sorted_incidents[:10]
    ]

    # Error & Access events
    error_events = db.scalars(
        select(OpsAuditEvent)
        .where(OpsAuditEvent.result.in_(["failure", "denied"]))
        .order_by(OpsAuditEvent.timestamp.desc())
        .limit(20)
    ).all()
    errors_24h = db.scalar(
        select(func.count()).select_from(OpsAuditEvent)
        .where(OpsAuditEvent.result.in_(["failure", "denied"]), OpsAuditEvent.timestamp >= t_24h)
    ) or 0
    denials_24h = db.scalar(
        select(func.count()).select_from(OpsAuditEvent)
        .where(OpsAuditEvent.result == "denied", OpsAuditEvent.timestamp >= t_24h)
    ) or 0

    recent_errors = [
        {
            "id": str(e.id),
            "timestamp": e.timestamp.isoformat(),
            "action": e.action,
            "actor_identity": e.actor_identity,
            "reason_code": e.reason_code,
            "result": e.result,
            "target_resource": e.target_resource,
            "affected_service": e.target_resource or "Operations API",
            "occurrences": 1,
        }
        for e in error_events
    ]

    php_probes = get_recent_php_probes(max_entries=15)
    php_count = len(php_probes)
    for p in php_probes:
        p["affected_service"] = "Nginx Web Gateway"
        p["occurrences"] = 1

    combined_errors = sorted(php_probes + recent_errors, key=lambda x: x.get("timestamp", ""), reverse=True)[:25]

    health_info = ops_get_health(db=db)
    if missing_action_outcomes > 0:
        health_info["components"]["audit_persistence"]["status"] = "degraded"
        health_info["status"] = "degraded"

    return {
        "generated_at": now_iso,
        "users": {
            "total_registered": total_users,
            "active_24h": active_users_24h,
            "active_7d": active_users_7d,
            "new_this_week": new_users_7d,
        },
        "membership": {
            "total_subscribers": total_subscribers,
            "premium_members": pro_members,
            "pro_members": pro_members,
            "free_members": free_members,
            "conversion_rate_pct": conversion_rate,
        },
        "activity": {
            "web_active_sessions_24h": web_sessions_24h,
            "web_logins_24h": web_logins_24h,
            "web_signups_7d": new_users_7d,
            "mobile_active_sessions_24h": mobile_sessions_24h,
            "mobile_rides_today": mobile_rides_today,
            "mobile_rides_active": mobile_rides_active,
            "mobile_datapoints_24h": mobile_datapoints_24h,
            "mobile_distance_miles_24h": round(float(mobile_distance_24h or 0.0), 1),
            "mobile_top_speed_mph_today": round(float(mobile_top_speed_24h or 0.0), 1),
            "mobile_avg_speed_mph_today": round(float(mobile_avg_speed_24h or 0.0), 1),
            "mobile_ingress_rate_hz": round(mobile_datapoints_24h / 86400.0, 2) if mobile_datapoints_24h > 0 else 0.0,
        },
        "website_activity": {
            "24h": {
                "active_users": active_users_24h,
                "successful_signins": web_logins_24h,
                "new_accounts": new_users_24h,
                "active_sessions": cust_active_sessions_now,
            },
            "7d": {
                "active_users": active_users_7d,
                "successful_signins": web_logins_7d,
                "new_accounts": new_users_7d,
                "active_sessions": cust_active_sessions_now,
            },
            "30d": {
                "active_users": active_users_30d,
                "successful_signins": web_logins_30d,
                "new_accounts": new_users_30d,
                "active_sessions": cust_active_sessions_now,
            },
            "daily_trends": daily_trends,
        },
        "subsystems": {
            "operations_api": {
                "status": api_status_label,
                "last_successful_upload": last_upload_iso,
                "failed_uploads_24h": failed_uploads_24h,
                "processing_backlog": queued_actions_backlog,
            },
            "operations_database": {
                "connection": "Connected" if db_ok else "Disconnected",
                "query_latency_ms": db_latency_ms if db_latency_ms is not None else 18.0,
                "audit_persistence": "Verified" if db_ok else "Failing",
                "schema_compatibility": "Current",
                "last_checked": now_iso,
            },
            "operations_access": {
                "session_checks": "Healthy" if db_ok else "Unavailable",
                "permission_enforcement": "Verified",
                "validation_errors_24h": validation_errors_24h,
                "last_checked": now_iso,
            },
            "audit_logging": {
                "last_successful_write": last_audit_write_iso,
                "failed_audit_writes_24h": failed_audit_writes_24h,
                "missing_action_outcomes": missing_action_outcomes,
                "verified_status": "Verified" if db_ok else "Failing",
                "last_checked": now_iso,
            },
        },
        "incidents": {
            "total_open": total_open,
            "critical_count": crit_count,
            "high_count": high_count,
            "medium_count": med_count,
            "low_count": low_count,
            "unassigned_count": unassigned_count,
            "recent_open": recent_open,
        },
        "errors": {
            "errors_24h": errors_24h + php_count,
            "denials_24h": denials_24h + php_count,
            "php_probes_24h": php_count,
            "access_denied_24h": denials_24h,
            "suspicious_requests_24h": php_count,
            "monitoring_errors_24h": max(0, errors_24h - denials_24h),
            "recent": combined_errors,
        },
        "health": health_info,
    }


@app.delete("/api/v1/ops/sessions/current", status_code=204)
def ops_logout_current_session(
    auth: tuple[Operator, OpsSession] = Depends(get_current_operator),
    db: DbSession = Depends(get_db),
):
    operator, session = auth
    session.revoked_at = datetime.now(timezone.utc)
    ops_audit(
        db=db,
        action="auth.session_revoked",
        actor_identity=operator.identity,
        actor_role=operator.role,
        target_resource="ops.session",
        result="success",
    )
    db.commit()


@app.post("/api/v1/ops/session/bootstrap")
def ops_bootstrap_session(
    current_user: User = Depends(get_current_user),
    db: DbSession = Depends(get_db),
):
    stmt = select(Operator).where(
        Operator.identity == current_user.email,
        Operator.status == "active",
    )
    operator = db.scalar(stmt)

    if operator is None:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="User identity is not an active Operations operator",
        )

    raw_ops_token = secrets.token_urlsafe(32)
    token_hash = hashlib.sha256(raw_ops_token.encode("utf-8")).hexdigest()
    now = datetime.now(timezone.utc)
    expires = now + timedelta(hours=8)
    auth_ver = getattr(operator, "authority_version", 1) or 1

    db.execute(text("""
        INSERT INTO ops.sessions (
            id, operator_id, token_digest, authority_version,
            firebase_auth_time, created_at, last_seen_at, absolute_expires_at,
            idle_timeout_seconds, session_token_hash, expires_at, last_activity_at
        ) VALUES (
            gen_random_uuid(), :op_id, :digest, :ver,
            :auth_time, :now, :now, :expires,
            900, :token_hash, :expires, :now
        )
    """), {
        "op_id": operator.id,
        "digest": token_hash,
        "ver": auth_ver,
        "auth_time": int(now.timestamp()),
        "now": now,
        "expires": expires,
        "token_hash": token_hash,
    })

    ops_audit(
        db=db,
        action="operator.session_bootstrap",
        actor_identity=operator.identity,
        actor_role=operator.role,
        target_resource="ops.session",
        result="success",
    )
    db.commit()

    return {
        "ops_token": raw_ops_token,
        "expires_at": expires.isoformat(),
        "role": operator.role,
        "identity": operator.identity,
    }


@app.get("/api/v1/ops/audit-events")

def ops_list_audit_events(
    cursor: str | None = None,
    limit: int = Query(default=25, ge=1, le=100),
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("owner")),
    db: DbSession = Depends(get_db),
):
    stmt = select(OpsAuditEvent).order_by(OpsAuditEvent.timestamp.desc())
    if cursor:
        try:
            cursor_dt = datetime.fromisoformat(cursor)
            stmt = stmt.where(OpsAuditEvent.timestamp < cursor_dt)
        except ValueError:
            pass

    events = db.scalars(stmt.limit(limit + 1)).all()
    has_more = len(events) > limit
    returned_events = events[:limit]

    next_cursor = returned_events[-1].timestamp.isoformat() if has_more and returned_events else None
    total_count = db.scalar(select(func.count(OpsAuditEvent.id))) or 0

    items = [
        {
            "id": str(e.id),
            "timestamp": e.timestamp.isoformat(),
            "action": e.action,
            "actor_identity": e.actor_identity,
            "actor_role": e.actor_role,
            "target_resource": e.target_resource,
            "reason_code": e.reason_code,
            "case_reference": e.case_reference,
            "result": e.result,
            "metadata": e.metadata_json or {},
        }
        for e in returned_events
    ]

    return {
        "items": items,
        "next_cursor": next_cursor,
        "total": total_count,
    }


# ============================================================
# Incidents Endpoints (OPS-4)
# ============================================================

VALID_SEVERITIES = {"low", "medium", "high", "critical"}
VALID_STATUSES = {"open", "investigating", "monitoring", "resolved", "closed", "archived"}

ALLOWED_TRANSITIONS = {
    "open": {"investigating", "resolved", "closed"},
    "investigating": {"monitoring", "resolved", "closed"},
    "monitoring": {"investigating", "resolved", "closed"},
    "resolved": {"closed", "investigating"},
    "closed": {"investigating"},  # Reopen (Owner only)
    "archived": set(),  # Terminal (no further mutation)
}


@app.get("/api/v1/ops/incidents")
def ops_list_incidents(
    status: str | None = None,
    severity: str | None = None,
    case_reference: str | None = None,
    cursor: str | None = None,
    limit: int = Query(default=25, ge=1, le=100),
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    stmt = select(OpsIncident).order_by(OpsIncident.created_at.desc())

    if case_reference:
        stmt = stmt.where(OpsIncident.case_reference == case_reference.strip())
    if status:
        stmt = stmt.where(OpsIncident.status == status.strip().lower())
    if severity:
        stmt = stmt.where(OpsIncident.severity == severity.strip().lower())

    if cursor:
        try:
            cursor_dt = datetime.fromisoformat(cursor)
            stmt = stmt.where(OpsIncident.created_at < cursor_dt)
        except ValueError:
            pass

    incidents = db.scalars(stmt.limit(limit + 1)).all()
    has_more = len(incidents) > limit
    returned_incidents = incidents[:limit]
    next_cursor = returned_incidents[-1].created_at.isoformat() if has_more and returned_incidents else None

    items = [
        {
            "case_reference": inc.case_reference,
            "title": inc.title,
            "severity": inc.severity,
            "status": inc.status,
            "assigned_to": getattr(inc, "assigned_to", None),
            "created_at": inc.created_at.isoformat(),
            "updated_at": inc.updated_at.isoformat(),
        }
        for inc in returned_incidents
    ]

    return {
        "items": items,
        "next_cursor": next_cursor,
    }


@app.post("/api/v1/ops/incidents", status_code=201)
def ops_create_incident(
    request: OpsIncidentCreate,
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    operator, _ = auth
    sev = request.severity.strip().lower()
    if sev not in VALID_SEVERITIES:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=f"Invalid severity '{request.severity}'. Must be one of: {sorted(VALID_SEVERITIES)}",
        )

    case_ref = generate_incident_case_reference(db)
    now = datetime.now(timezone.utc)

    incident = OpsIncident(
        case_reference=case_ref,
        title=request.title.strip(),
        summary=request.summary.strip(),
        severity=sev,
        status="open",
        created_at=now,
        created_by_operator_id=operator.id,
        updated_at=now,
    )
    db.add(incident)
    db.flush()

    # Transactional Audit
    ops_audit(
        db=db,
        action="incident.created",
        actor_identity=operator.identity,
        actor_role=operator.role,
        target_resource="ops.incident",
        case_reference=case_ref,
        result="success",
        metadata={"severity": sev},
    )

    db.commit()
    db.refresh(incident)

    return {
        "case_reference": incident.case_reference,
        "title": incident.title,
        "summary": incident.summary,
        "severity": incident.severity,
        "status": incident.status,
        "created_at": incident.created_at.isoformat(),
        "updated_at": incident.updated_at.isoformat(),
        "closed_at": None,
        "resolution": None,
        "created_by_identity": operator.identity,
        "closed_by_identity": None,
    }


@app.get("/api/v1/ops/incidents/{case_reference}")
def ops_get_incident_detail(
    case_reference: str,
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    stmt = select(OpsIncident).where(OpsIncident.case_reference == case_reference.strip())
    incident = db.scalar(stmt)
    if incident is None:
        raise HTTPException(status_code=404, detail="Incident not found")

    creator = db.get(Operator, incident.created_by_operator_id)
    closer = db.get(Operator, incident.closed_by_operator_id) if incident.closed_by_operator_id else None

    return {
        "case_reference": incident.case_reference,
        "title": incident.title,
        "summary": incident.summary,
        "severity": incident.severity,
        "status": incident.status,
        "created_at": incident.created_at.isoformat(),
        "updated_at": incident.updated_at.isoformat(),
        "closed_at": incident.closed_at.isoformat() if incident.closed_at else None,
        "resolution": incident.resolution,
        "assigned_to": getattr(incident, "assigned_to", None),
        "created_by_identity": creator.identity if creator else "operator",
        "closed_by_identity": closer.identity if closer else None,
    }


@app.patch("/api/v1/ops/incidents/{case_reference}")
def ops_patch_incident(
    case_reference: str,
    patch: OpsIncidentPatch,
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    operator, _ = auth
    stmt = select(OpsIncident).where(OpsIncident.case_reference == case_reference.strip())
    incident = db.scalar(stmt)
    if incident is None:
        raise HTTPException(status_code=404, detail="Incident not found")

    if incident.status == "archived":
        raise HTTPException(status_code=409, detail="Archived incidents cannot be modified")

    now = datetime.now(timezone.utc)
    target_status = patch.status.strip().lower() if patch.status else None
    target_severity = patch.severity.strip().lower() if patch.severity else None
    new_resolution = patch.resolution.strip() if patch.resolution is not None else None

    # Severity update
    if target_severity:
        if target_severity not in VALID_SEVERITIES:
            raise HTTPException(
                status_code=422,
                detail=f"Invalid severity '{patch.severity}'. Must be: {sorted(VALID_SEVERITIES)}",
            )
        if target_severity != incident.severity:
            old_sev = incident.severity
            incident.severity = target_severity
            incident.updated_at = now
            ops_audit(
                db=db,
                action="incident.severity_changed",
                actor_identity=operator.identity,
                actor_role=operator.role,
                target_resource="ops.incident",
                case_reference=incident.case_reference,
                result="success",
                metadata={"from": old_sev, "to": target_severity},
            )

    # Resolution update
    if new_resolution is not None and new_resolution != incident.resolution:
        incident.resolution = new_resolution
        incident.updated_at = now
        ops_audit(
            db=db,
            action="incident.resolved",
            actor_identity=operator.identity,
            actor_role=operator.role,
            target_resource="ops.incident",
            case_reference=incident.case_reference,
            result="success",
        )

    # Status transition
    if target_status and target_status != incident.status:
        if target_status not in VALID_STATUSES:
            raise HTTPException(
                status_code=422,
                detail=f"Invalid status '{patch.status}'. Must be: {sorted(VALID_STATUSES)}",
            )

        # Owner only transitions
        if target_status == "archived" and operator.role != "owner":
            raise HTTPException(status_code=403, detail="Only Owners may archive incidents")

        if incident.status == "closed" and target_status == "investigating" and operator.role != "owner":
            raise HTTPException(status_code=403, detail="Only Owners may reopen closed incidents")

        # Allowed transitions
        allowed = ALLOWED_TRANSITIONS.get(incident.status, set())
        # Add archived as valid destination for non-archived if owner
        if operator.role == "owner" and incident.status != "archived":
            allowed = allowed | {"archived"}

        if target_status not in allowed:
            raise HTTPException(
                status_code=409,
                detail=f"Invalid status transition from '{incident.status}' to '{target_status}'",
            )

        # Resolution required for resolved or closed
        effective_res = new_resolution if new_resolution is not None else incident.resolution
        if target_status in {"resolved", "closed"} and (not effective_res or not effective_res.strip()):
            raise HTTPException(
                status_code=422,
                detail="Resolution text is required before resolving or closing an incident",
            )

        old_status = incident.status
        incident.status = target_status
        incident.updated_at = now

        if target_status == "closed":
            incident.closed_at = now
            incident.closed_by_operator_id = operator.id
            audit_action = "incident.closed"
        elif target_status == "resolved":
            audit_action = "incident.resolved"
        elif old_status == "closed" and target_status == "investigating":
            incident.closed_at = None
            incident.closed_by_operator_id = None
            audit_action = "incident.reopened"
        elif target_status == "archived":
            audit_action = "incident.archived"
        else:
            audit_action = "incident.status_changed"

        ops_audit(
            db=db,
            action=audit_action,
            actor_identity=operator.identity,
            actor_role=operator.role,
            target_resource="ops.incident",
            case_reference=incident.case_reference,
            result="success",
            metadata={"from": old_status, "to": target_status},
        )

    # Assigned to update
    if patch.assigned_to is not None:
        new_assigned = patch.assigned_to.strip() if patch.assigned_to.strip() else None
        old_assigned = getattr(incident, "assigned_to", None)
        if new_assigned != old_assigned:
            incident.assigned_to = new_assigned
            incident.updated_at = now
            ops_audit(
                db=db,
                action="incident.assigned",
                actor_identity=operator.identity,
                actor_role=operator.role,
                target_resource="ops.incident",
                case_reference=incident.case_reference,
                result="success",
                metadata={"from": old_assigned, "to": new_assigned},
            )

    db.commit()
    db.refresh(incident)

    creator = db.get(Operator, incident.created_by_operator_id)
    closer = db.get(Operator, incident.closed_by_operator_id) if incident.closed_by_operator_id else None

    return {
        "case_reference": incident.case_reference,
        "title": incident.title,
        "summary": incident.summary,
        "severity": incident.severity,
        "status": incident.status,
        "created_at": incident.created_at.isoformat(),
        "updated_at": incident.updated_at.isoformat(),
        "closed_at": incident.closed_at.isoformat() if incident.closed_at else None,
        "resolution": incident.resolution,
        "assigned_to": getattr(incident, "assigned_to", None),
        "created_by_identity": creator.identity if creator else "operator",
        "closed_by_identity": closer.identity if closer else None,
    }


@app.get("/api/v1/ops/incidents/{case_reference}/notes")
def ops_list_incident_notes(
    case_reference: str,
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    stmt = select(OpsIncident).where(OpsIncident.case_reference == case_reference.strip())
    incident = db.scalar(stmt)
    if incident is None:
        raise HTTPException(status_code=404, detail="Incident not found")

    notes_stmt = (
        select(OpsIncidentNote, Operator.identity)
        .join(Operator, OpsIncidentNote.created_by_operator_id == Operator.id)
        .where(OpsIncidentNote.incident_id == incident.id)
        .order_by(OpsIncidentNote.created_at.desc())
    )
    notes_rows = db.execute(notes_stmt).all()

    return {
        "notes": [
            {
                "created_at": note.created_at.isoformat(),
                "created_by_identity": op_identity,
                "body": note.body,
            }
            for note, op_identity in notes_rows
        ]
    }


@app.post("/api/v1/ops/incidents/{case_reference}/notes", status_code=201)
def ops_create_incident_note(
    case_reference: str,
    request: OpsIncidentNoteCreate,
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    operator, _ = auth
    stmt = select(OpsIncident).where(OpsIncident.case_reference == case_reference.strip())
    incident = db.scalar(stmt)
    if incident is None:
        raise HTTPException(status_code=404, detail="Incident not found")

    if incident.status == "archived":
        raise HTTPException(status_code=409, detail="Archived incidents cannot be modified")

    body_text = request.body.strip()
    if not body_text:
        raise HTTPException(status_code=422, detail="Note body cannot be empty")

    now = datetime.now(timezone.utc)
    note = OpsIncidentNote(
        incident_id=incident.id,
        created_at=now,
        created_by_operator_id=operator.id,
        body=body_text,
    )
    db.add(note)
    incident.updated_at = now
    db.flush()

    # Transactional Audit
    ops_audit(
        db=db,
        action="incident.note_added",
        actor_identity=operator.identity,
        actor_role=operator.role,
        target_resource="ops.incident",
        case_reference=incident.case_reference,
        result="success",
    )

    db.commit()
    db.refresh(note)

    return {
        "created_at": note.created_at.isoformat(),
        "created_by_identity": operator.identity,
        "body": note.body,
    }


# ============================================================
# Operations (ops) - Section 2: User Management
# ============================================================


@app.get("/api/v1/ops/users")
def ops_list_users(
    q: str | None = None,
    cursor: str | None = None,
    limit: int = Query(default=25, ge=1, le=100),
    status_filter: str | None = None,
    verified_filter: str | None = None,
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    base_stmt = select(User)

    if q and q.strip():
        clean_q = q.strip()
        try:
            uuid_q = uuid.UUID(clean_q)
            base_stmt = base_stmt.where(or_(User.id == uuid_q, User.email.ilike(f"%{clean_q}%")))
        except ValueError:
            base_stmt = base_stmt.where(User.email.ilike(f"%{clean_q}%"))

    if status_filter:
        s_filter = status_filter.strip().lower()
        if s_filter == "active":
            base_stmt = base_stmt.where(User.is_active.is_(True))
        elif s_filter == "suspended":
            base_stmt = base_stmt.where(User.is_active.is_(False))

    if verified_filter:
        v_filter = verified_filter.strip().lower()
        if v_filter == "verified":
            base_stmt = base_stmt.where(User.email_verified.is_(True))
        elif v_filter == "unverified":
            base_stmt = base_stmt.where(User.email_verified.is_(False))

    total = db.scalar(select(func.count()).select_from(base_stmt.subquery())) or 0

    stmt = base_stmt.order_by(User.created_at.desc(), User.id.desc())

    if cursor:
        try:
            cursor_dt = datetime.fromisoformat(cursor)
            stmt = stmt.where(User.created_at < cursor_dt)
        except ValueError:
            pass

    users = db.scalars(stmt.limit(limit + 1)).all()
    has_more = len(users) > limit
    returned_users = users[:limit]
    next_cursor = returned_users[-1].created_at.isoformat() if has_more and returned_users else None

    user_ids = [u.id for u in returned_users]
    now = datetime.now(timezone.utc)

    active_sessions_count_map: dict[uuid.UUID, int] = {}
    total_rides_map: dict[uuid.UUID, int] = {}
    last_sync_map: dict[uuid.UUID, datetime] = {}
    latest_version_map: dict[uuid.UUID, str | None] = {}

    if user_ids:
        session_counts = db.execute(
            select(Session.user_id, func.count(Session.id))
            .where(
                Session.user_id.in_(user_ids),
                Session.revoked_at.is_(None),
                Session.expires_at > now,
            )
            .group_by(Session.user_id)
        ).all()
        for uid, count in session_counts:
            active_sessions_count_map[uid] = count

        ride_counts = db.execute(
            select(Ride.user_id, func.count(Ride.id))
            .where(Ride.user_id.in_(user_ids))
            .group_by(Ride.user_id)
        ).all()
        for uid, count in ride_counts:
            total_rides_map[uid] = count

        last_rides = db.execute(
            select(Ride.user_id, func.max(Ride.created_at))
            .where(Ride.user_id.in_(user_ids))
            .group_by(Ride.user_id)
        ).all()
        for uid, dt in last_rides:
            if dt:
                last_sync_map[uid] = dt

        last_telemetry = db.execute(
            select(AuditEvent.actor_user_id, func.max(AuditEvent.created_at))
            .where(
                AuditEvent.actor_user_id.in_(user_ids),
                AuditEvent.event_type == "telemetry.ingested",
            )
            .group_by(AuditEvent.actor_user_id)
        ).all()
        for uid, dt in last_telemetry:
            if dt and (uid not in last_sync_map or dt > last_sync_map[uid]):
                last_sync_map[uid] = dt

        for uid in user_ids:
            ev = db.scalar(
                select(AuditEvent)
                .where(
                    or_(AuditEvent.actor_user_id == uid, AuditEvent.target_user_id == uid),
                )
                .order_by(AuditEvent.created_at.desc())
                .limit(1)
            )
            if ev and ev.metadata_json and isinstance(ev.metadata_json, dict) and ev.metadata_json.get("app_version"):
                latest_version_map[uid] = str(ev.metadata_json["app_version"])
            elif ev and ev.user_agent and "MotorSignal/" in ev.user_agent:
                latest_version_map[uid] = ev.user_agent.split("MotorSignal/")[1].split()[0]
            else:
                sess = db.scalar(
                    select(Session)
                    .where(Session.user_id == uid)
                    .order_by(Session.created_at.desc())
                    .limit(1)
                )
                if sess and sess.device_name:
                    latest_version_map[uid] = sess.device_name
                else:
                    latest_version_map[uid] = None

    items = [
        {
            "id": str(u.id),
            "email": u.email,
            "email_verified": u.email_verified,
            "is_active": u.is_active,
            "role": u.role,
            "created_at": u.created_at.isoformat(),
            "latest_app_version": latest_version_map.get(u.id),
            "last_sync_at": last_sync_map.get(u.id).isoformat() if last_sync_map.get(u.id) else None,
            "active_sessions_count": active_sessions_count_map.get(u.id, 0),
            "total_rides": total_rides_map.get(u.id, 0),
        }
        for u in returned_users
    ]

    return {
        "items": items,
        "next_cursor": next_cursor,
        "total": total,
    }


@app.get("/api/v1/ops/users/{user_id}")
def ops_get_user(
    user_id: str,
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    try:
        u_uuid = uuid.UUID(user_id)
    except ValueError:
        raise HTTPException(status_code=404, detail="User not found")

    user = db.get(User, u_uuid)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")

    now = datetime.now(timezone.utc)

    sessions = db.scalars(
        select(Session)
        .where(Session.user_id == user.id)
        .order_by(Session.created_at.desc())
    ).all()
    sessions_payload = [
        {
            "id": str(s.id),
            "device_name": s.device_name,
            "created_at": s.created_at.isoformat(),
            "last_used_at": s.last_used_at.isoformat() if s.last_used_at else None,
            "is_revoked": s.revoked_at is not None or s.expires_at <= now,
        }
        for s in sessions
    ]

    subscription = db.scalar(
        select(Subscription).where(Subscription.user_id == user.id)
    )
    sub_payload = None
    if subscription:
        sub_payload = {
            "plan": subscription.plan,
            "status": subscription.status,
            "expires_at": subscription.current_period_end.isoformat() if subscription.current_period_end else None,
        }

    total_rides = db.scalar(
        select(func.count(Ride.id)).where(Ride.user_id == user.id)
    ) or 0
    last_ride_dt = db.scalar(
        select(func.max(Ride.created_at)).where(Ride.user_id == user.id)
    )
    ride_summary = {
        "total_rides": total_rides,
        "last_ride_at": last_ride_dt.isoformat() if last_ride_dt else None,
    }

    audit_events = db.scalars(
        select(AuditEvent)
        .where(or_(AuditEvent.actor_user_id == user.id, AuditEvent.target_user_id == user.id))
        .order_by(AuditEvent.created_at.desc())
        .limit(10)
    ).all()

    ops_audit_events = db.scalars(
        select(OpsAuditEvent)
        .where(OpsAuditEvent.target_resource == f"user:{user.id}")
        .order_by(OpsAuditEvent.timestamp.desc())
        .limit(10)
    ).all()

    combined_audit = []
    for ae in audit_events:
        combined_audit.append({
            "action": ae.event_type,
            "timestamp": ae.created_at.isoformat(),
            "ip": ae.ip_address,
            "_raw_time": ae.created_at,
        })
    for oae in ops_audit_events:
        combined_audit.append({
            "action": oae.action,
            "timestamp": oae.timestamp.isoformat(),
            "ip": None,
            "_raw_time": oae.timestamp,
        })

    combined_audit.sort(key=lambda x: x["_raw_time"], reverse=True)
    recent_audit = [
        {"action": a["action"], "timestamp": a["timestamp"], "ip": a["ip"]}
        for a in combined_audit[:10]
    ]

    return {
        "id": str(user.id),
        "email": user.email,
        "email_verified": user.email_verified,
        "is_active": user.is_active,
        "role": user.role,
        "created_at": user.created_at.isoformat(),
        "updated_at": user.updated_at.isoformat(),
        "sessions": sessions_payload,
        "subscription": sub_payload,
        "ride_summary": ride_summary,
        "recent_audit": recent_audit,
    }


@app.post("/api/v1/ops/users/{user_id}/status")
def ops_update_user_status(
    user_id: str,
    request: OpsUserStatusUpdate,
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    operator, _ = auth
    try:
        u_uuid = uuid.UUID(user_id)
    except ValueError:
        raise HTTPException(status_code=404, detail="User not found")

    user = db.get(User, u_uuid)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")

    now = datetime.now(timezone.utc)
    user.is_active = request.is_active
    user.updated_at = now

    if not request.is_active:
        db.query(Session).filter(
            Session.user_id == user.id,
            Session.revoked_at.is_(None),
        ).update({Session.revoked_at: now}, synchronize_session=False)

    ops_audit(
        db=db,
        action="user.suspend" if not request.is_active else "user.reinstate",
        actor_identity=operator.identity,
        actor_role=operator.role,
        target_resource=f"user:{user.id}",
        reason_code=request.reason,
        result="success",
    )
    db.commit()
    db.refresh(user)

    return {
        "id": str(user.id),
        "is_active": user.is_active,
        "status": "active" if user.is_active else "suspended",
    }


@app.post("/api/v1/ops/users/{user_id}/revoke-sessions")
def ops_revoke_user_sessions(
    user_id: str,
    request: OpsUserActionRequest,
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    operator, _ = auth
    try:
        u_uuid = uuid.UUID(user_id)
    except ValueError:
        raise HTTPException(status_code=404, detail="User not found")

    user = db.get(User, u_uuid)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")

    now = datetime.now(timezone.utc)
    db.query(Session).filter(
        Session.user_id == user.id,
        Session.revoked_at.is_(None),
    ).update({Session.revoked_at: now}, synchronize_session=False)

    ops_audit(
        db=db,
        action="user.sessions_revoked",
        actor_identity=operator.identity,
        actor_role=operator.role,
        target_resource=f"user:{user.id}",
        reason_code=request.reason,
        result="success",
    )
    db.commit()

    return {
        "id": str(user.id),
        "status": "revoked",
    }


@app.post("/api/v1/ops/users/{user_id}/reset-password")
def ops_reset_user_password(
    user_id: str,
    request: OpsUserActionRequest,
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    operator, _ = auth
    try:
        u_uuid = uuid.UUID(user_id)
    except ValueError:
        raise HTTPException(status_code=404, detail="User not found")

    user = db.get(User, u_uuid)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")

    raw, token_hash = create_one_time_token()
    now = datetime.now(timezone.utc)
    expires_at = now + timedelta(hours=24)

    db.add(
        PasswordResetToken(
            user_id=user.id,
            token_hash=token_hash,
            expires_at=expires_at,
        )
    )

    ops_audit(
        db=db,
        action="user.password_reset_initiated",
        actor_identity=operator.identity,
        actor_role=operator.role,
        target_resource=f"user:{user.id}",
        reason_code=request.reason,
        result="success",
    )
    db.commit()

    web_base = os.getenv("WEB_BASE_URL", "https://motorsignal.app")
    send_email(
        user.email,
        "Reset your MotorSignal password",
        f"Reset your password: {web_base}/forgot-password?token={raw}",
    )

    # Never expose passwords or raw tokens
    return {
        "status": "dispatched",
        "expires_at": expires_at.isoformat(),
    }


class OpsUserCreateRequest(BaseModel):
    email: str
    display_name: str | None = None
    password: str | None = None
    role: str = "user"
    email_verified: bool = True
    reason: str


@app.post("/api/v1/ops/users", status_code=201)
def ops_create_user(
    request: OpsUserCreateRequest,
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    operator, _ = auth
    reason_clean = request.reason.strip()
    if len(reason_clean) < 3:
        raise HTTPException(status_code=422, detail="Audit reason must be at least 3 characters")

    try:
        validated = validate_email(request.email, check_deliverability=False)
        email = normalize_email(validated.normalized)
    except EmailNotValidError:
        raise HTTPException(status_code=400, detail="Invalid email address")

    existing = db.scalar(select(User).where(User.email == email))
    if existing is not None:
        raise HTTPException(status_code=409, detail="User account with this email already exists")

    role_clean = request.role.strip().lower()
    if role_clean in ("operator", "owner") and operator.role != "owner":
        raise HTTPException(status_code=403, detail="Only owner can assign administrative roles")
    if role_clean not in ("user", "operator", "owner"):
        role_clean = "user"

    raw_password = request.password.strip() if request.password and request.password.strip() else secrets.token_urlsafe(16)
    hashed_pwd = hash_password(raw_password)

    new_user = User(
        email=email,
        password_hash=hashed_pwd,
        email_verified=request.email_verified,
        role=role_clean,
        is_active=True,
    )
    db.add(new_user)
    db.flush()

    if role_clean in ("operator", "owner"):
        op_entry = db.scalar(select(Operator).where(Operator.identity == email))
        if op_entry is None:
            op_entry = Operator(
                identity=email,
                role=role_clean,
                status="active",
            )
            db.add(op_entry)
        else:
            op_entry.role = role_clean
            op_entry.status = "active"

    ops_audit(
        db=db,
        action="user.created",
        actor_identity=operator.identity,
        actor_role=operator.role,
        target_resource=f"user:{new_user.id}",
        reason_code=reason_clean,
        result="success",
        metadata={
            "email": new_user.email,
            "role": new_user.role,
            "email_verified": new_user.email_verified,
        },
    )
    db.commit()
    db.refresh(new_user)

    return {
        "id": str(new_user.id),
        "email": new_user.email,
        "display_name": request.display_name,
        "role": new_user.role,
        "is_active": new_user.is_active,
        "email_verified": new_user.email_verified,
        "is_email_verified": new_user.email_verified,
        "created_at": new_user.created_at.isoformat(),
        "updated_at": new_user.updated_at.isoformat(),
        "active_sessions_count": 0,
        "total_rides_count": 0,
        "last_sync_at": None,
        "latest_app_version": None,
    }


# ============================================================
# Operations (ops) - Section 3: Support Center & Diagnostics
# ============================================================

_support_case_seq = 100


def generate_support_case_reference(db: DbSession | None = None) -> str:
    global _support_case_seq
    year = datetime.now(timezone.utc).year
    if db is not None:
        try:
            with db.begin_nested():
                max_ref = db.scalar(
                    select(func.max(OpsSupportTicket.case_reference)).where(
                        OpsSupportTicket.case_reference.like(f"SUP-{year}-%")
                    )
                )
                max_suffix = 100
                if max_ref and max_ref.startswith(f"SUP-{year}-"):
                    try:
                        max_suffix = max(max_suffix, int(max_ref.split("-")[-1]))
                    except (ValueError, IndexError):
                        pass

                val = db.execute(text("SELECT nextval('ops.support_case_seq')")).scalar()
                if val is not None:
                    if int(val) <= max_suffix:
                        val = db.execute(
                            text(f"SELECT setval('ops.support_case_seq', {max_suffix + 1}, true)")
                        ).scalar()
                    return f"SUP-{year}-{int(val):06d}"
        except Exception:
            pass

        try:
            with db.begin_nested():
                max_ref = db.scalar(
                    select(func.max(OpsSupportTicket.case_reference)).where(
                        OpsSupportTicket.case_reference.like(f"SUP-{year}-%")
                    )
                )
                if max_ref and max_ref.startswith(f"SUP-{year}-"):
                    suffix = int(max_ref.split("-")[-1])
                    if suffix >= _support_case_seq:
                        _support_case_seq = suffix
        except Exception:
            pass
    _support_case_seq += 1
    return f"SUP-{year}-{_support_case_seq:06d}"


SUPPORT_TICKETS: dict[str, dict] = {
    "e1a11111-2222-3333-4444-555555555551": {
        "id": "e1a11111-2222-3333-4444-555555555551",
        "case_reference": "SUP-2026-000101",
        "ticket_type": "bug_report",
        "reporter_email": "rider.alex@example.com",
        "subject": "Lean angle reading freeze during high-bank turn",
        "description": "During a canyon ride yesterday, the lean angle widget stuck at 28 degrees right and stopped updating until I restarted the app. Telemetry was still logging speed and GPS coordinates.",
        "severity": "high",
        "status": "open",
        "created_at": "2026-09-14T18:22:15Z",
        "updated_at": "2026-09-14T19:00:00Z",
        "resolved_at": None,
        "notes": [
            {
                "id": "n1111111-0000-0000-0000-000000000001",
                "author_identity": "ops-lead@motorsignal.com",
                "body": "Reproduced in lab with high vibration profile. Escalating to firmware/sensor fusion team.",
                "created_at": "2026-09-14T19:00:00Z",
            }
        ],
        "diagnostic_bundle": {
            "app_version": "2.4.1 (build 184)",
            "os_version": "Android 14 (API 34)",
            "device_model": "Google Pixel 8 Pro",
            "stack_trace": "SensorFusionException: IMU sample buffer overflow at SensorFusionService.processGyro (sensor_fusion_service.dart:182)\n  at TelemetryPipeline.ingest (telemetry_service.dart:94)",
            "sensor_status": "IMU: DEGRADED (sample rate jitter > 45ms), GPS: OK (3D Fix, 11 satellites), BLE: CONNECTED",
            "logs_snippet": "[14:22:01.102] [SensorFusion] Buffer warning: 82 pending frames\n[14:22:01.152] [SensorFusion] Gyro acquisition hitch: 52ms > 50ms ceiling\n[14:22:01.204] [SensorFusion] Fallback complementary filter engaged",
            "submitted_at": "2026-09-14T18:22:15Z",
        },
    },
    "e1a11111-2222-3333-4444-555555555552": {
        "id": "e1a11111-2222-3333-4444-555555555552",
        "case_reference": "SUP-2026-000102",
        "ticket_type": "contact_request",
        "reporter_email": "marcus.v@motoclub-pnw.org",
        "subject": "Club Fleet Telemetry Export & Multi-rider Group Inquiries",
        "description": "We run a 40-rider club in Oregon and want to know if MotorSignal Pro supports group ride telemetry aggregation or team dashboard exports.",
        "severity": "medium",
        "status": "in_progress",
        "created_at": "2026-09-13T10:15:00Z",
        "updated_at": "2026-09-13T11:20:00Z",
        "resolved_at": None,
        "notes": [
            {
                "id": "n1111111-0000-0000-0000-000000000002",
                "author_identity": "operator@motorsignal.com",
                "body": "Replied via email explaining Pro group ride features and export formats.",
                "created_at": "2026-09-13T11:20:00Z",
            }
        ],
        "diagnostic_bundle": None,
    },
    "e1a11111-2222-3333-4444-555555555553": {
        "id": "e1a11111-2222-3333-4444-555555555553",
        "case_reference": "SUP-2026-000103",
        "ticket_type": "telemetry_issue",
        "reporter_email": "trackday_sam@fastlap.net",
        "subject": "0-60 timer false trigger while idling on paddock stand",
        "description": "Kickstand was down on rear paddock stand while warming up engine. The 0-60 timer armed and triggered a 1.2s run due to vibration spike.",
        "severity": "critical",
        "status": "open",
        "created_at": "2026-09-15T07:15:00Z",
        "updated_at": "2026-09-15T07:15:00Z",
        "resolved_at": None,
        "notes": [],
        "diagnostic_bundle": {
            "app_version": "2.4.2-beta.3",
            "os_version": "iOS 18.0.1",
            "device_model": "iPhone 16 Pro",
            "stack_trace": None,
            "sensor_status": "IMU: OK, GPS: STATIONARY (Speed 0.2 mph), KICKSTAND_ISOLATION: INACTIVE",
            "logs_snippet": "[09:12:00.040] [RideStateMachine] Speed 0.18 mph < 3.0 mph gate\n[09:12:00.090] [AccelTrigger] Peak longitudinal G: 1.42g\n[09:12:00.100] [PerfTimer] Run triggered in paddock",
            "submitted_at": "2026-09-15T07:15:00Z",
        },
    },
    "e1a11111-2222-3333-4444-555555555554": {
        "id": "e1a11111-2222-3333-4444-555555555554",
        "case_reference": "SUP-2026-000104",
        "ticket_type": "general",
        "reporter_email": "elena.r@example.com",
        "subject": "Account email change verification link expired",
        "description": "Requested email change link but clicked it 30 hours later and it said token expired.",
        "severity": "low",
        "status": "resolved",
        "created_at": "2026-09-14T14:30:00Z",
        "updated_at": "2026-09-14T16:00:00Z",
        "resolved_at": "2026-09-14T16:00:00Z",
        "notes": [
            {
                "id": "n1111111-0000-0000-0000-000000000003",
                "author_identity": "operator@motorsignal.com",
                "body": "Re-issued email change verification token from user management.",
                "created_at": "2026-09-14T15:58:00Z",
            }
        ],
        "diagnostic_bundle": None,
    },
}


def _seed_support_tickets_if_needed(db: DbSession):
    try:
        count = db.scalar(select(func.count()).select_from(OpsSupportTicket))
        if count == 0:
            for t_dict in SUPPORT_TICKETS.values():
                t_id_uuid = uuid.UUID(t_dict["id"])
                created_dt = datetime.fromisoformat(t_dict["created_at"].replace("Z", "+00:00"))
                updated_dt = datetime.fromisoformat(t_dict["updated_at"].replace("Z", "+00:00"))
                resolved_dt = datetime.fromisoformat(t_dict["resolved_at"].replace("Z", "+00:00")) if t_dict.get("resolved_at") else None
                t_obj = OpsSupportTicket(
                    id=t_id_uuid,
                    case_reference=t_dict["case_reference"],
                    ticket_type=t_dict["ticket_type"],
                    reporter_email=t_dict["reporter_email"],
                    subject=t_dict["subject"],
                    description=t_dict["description"],
                    severity=t_dict["severity"],
                    status=t_dict["status"],
                    created_at=created_dt,
                    updated_at=updated_dt,
                    resolved_at=resolved_dt,
                    diagnostic_bundle=t_dict.get("diagnostic_bundle"),
                )
                db.add(t_obj)
                for n_dict in t_dict.get("notes", []):
                    n_id = n_dict.get("id", "")
                    try:
                        n_uuid = uuid.UUID(n_id)
                    except Exception:
                        n_uuid = uuid.uuid4()
                    n_obj = OpsSupportTicketNote(
                        id=n_uuid,
                        ticket_id=t_id_uuid,
                        author_identity=n_dict["author_identity"],
                        body=n_dict["body"],
                        created_at=datetime.fromisoformat(n_dict["created_at"].replace("Z", "+00:00")),
                    )
                    db.add(n_obj)
            db.commit()
            try:
                with db.begin_nested():
                    db.execute(text("SELECT setval('ops.support_case_seq', 104, true)"))
            except Exception:
                pass
    except Exception as e:
        db.rollback()
        logger.warning(f"Support ticket DB seed skipped/failed: {e}")


def find_support_ticket(ticket_id: str, db: DbSession | None = None) -> dict | None:
    clean_id = ticket_id.strip()
    if db is not None:
        try:
            _seed_support_tickets_if_needed(db)
            t_obj = None
            try:
                u = uuid.UUID(clean_id)
                t_obj = db.scalar(select(OpsSupportTicket).where(OpsSupportTicket.id == u))
            except ValueError:
                pass
            if t_obj is None:
                t_obj = db.scalar(select(OpsSupportTicket).where(OpsSupportTicket.case_reference == clean_id))
            if t_obj is not None:
                notes = [
                    {
                        "id": str(n.id),
                        "author_identity": n.author_identity,
                        "body": n.body,
                        "created_at": n.created_at.isoformat(),
                    }
                    for n in t_obj.notes
                ]
                d = {
                    "id": str(t_obj.id),
                    "case_reference": t_obj.case_reference,
                    "ticket_type": t_obj.ticket_type,
                    "reporter_email": t_obj.reporter_email,
                    "subject": t_obj.subject,
                    "description": t_obj.description,
                    "severity": t_obj.severity,
                    "status": t_obj.status,
                    "created_at": t_obj.created_at.isoformat(),
                    "updated_at": t_obj.updated_at.isoformat(),
                    "resolved_at": t_obj.resolved_at.isoformat() if t_obj.resolved_at else None,
                    "diagnostic_bundle": t_obj.diagnostic_bundle,
                    "notes": notes,
                }
                SUPPORT_TICKETS[str(t_obj.id)] = d
                return d
        except Exception as e:
            logger.warning(f"Error reading ticket from DB: {e}")

    # Fallback to in-memory SUPPORT_TICKETS
    if clean_id in SUPPORT_TICKETS:
        return SUPPORT_TICKETS[clean_id]
    for t in SUPPORT_TICKETS.values():
        if t["case_reference"] == clean_id or t["id"] == clean_id:
            return t
    return None


@app.get("/api/v1/ops/support/tickets")
def ops_list_support_tickets(
    type_filter: str | None = None,
    status_filter: str | None = None,
    q: str | None = None,
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    _seed_support_tickets_if_needed(db)
    try:
        stmt = select(OpsSupportTicket).order_by(OpsSupportTicket.created_at.desc())
        if type_filter and type_filter.strip() and type_filter.strip().lower() != "all":
            stmt = stmt.where(OpsSupportTicket.ticket_type == type_filter.strip().lower())
        if status_filter and status_filter.strip() and status_filter.strip().lower() != "all":
            stmt = stmt.where(OpsSupportTicket.status == status_filter.strip().lower())
        if q and q.strip():
            query_term = f"%{q.strip().lower()}%"
            stmt = stmt.where(
                or_(
                    OpsSupportTicket.case_reference.ilike(query_term),
                    OpsSupportTicket.reporter_email.ilike(query_term),
                    OpsSupportTicket.subject.ilike(query_term),
                    OpsSupportTicket.description.ilike(query_term),
                )
            )
        db_tickets = db.scalars(stmt).all()
        if db_tickets:
            items = [
                {
                    "id": str(t.id),
                    "case_reference": t.case_reference,
                    "ticket_type": t.ticket_type,
                    "reporter_email": t.reporter_email,
                    "subject": t.subject,
                    "severity": t.severity,
                    "status": t.status,
                    "created_at": t.created_at.isoformat(),
                    "updated_at": t.updated_at.isoformat(),
                    "resolved_at": t.resolved_at.isoformat() if t.resolved_at else None,
                    "has_diagnostics": t.diagnostic_bundle is not None,
                    "notes_count": len(t.notes),
                }
                for t in db_tickets
            ]
            return {"items": items, "total": len(items)}
    except Exception as e:
        logger.warning(f"Error querying tickets from DB: {e}")

    tickets = list(SUPPORT_TICKETS.values())
    if type_filter and type_filter.strip() and type_filter.strip().lower() != "all":
        tf = type_filter.strip().lower()
        tickets = [t for t in tickets if t["ticket_type"] == tf]

    if status_filter and status_filter.strip() and status_filter.strip().lower() != "all":
        sf = status_filter.strip().lower()
        tickets = [t for t in tickets if t["status"] == sf]

    if q and q.strip():
        query = q.strip().lower()
        tickets = [
            t
            for t in tickets
            if query in t["case_reference"].lower()
            or query in t["reporter_email"].lower()
            or query in t["subject"].lower()
            or query in t["description"].lower()
        ]

    tickets.sort(key=lambda t: t["created_at"], reverse=True)
    items = [
        {
            "id": t["id"],
            "case_reference": t["case_reference"],
            "ticket_type": t["ticket_type"],
            "reporter_email": t["reporter_email"],
            "subject": t["subject"],
            "severity": t["severity"],
            "status": t["status"],
            "created_at": t["created_at"],
            "updated_at": t["updated_at"],
            "resolved_at": t.get("resolved_at"),
            "has_diagnostics": t.get("diagnostic_bundle") is not None,
            "notes_count": len(t.get("notes", [])),
        }
        for t in tickets
    ]
    return {"items": items, "total": len(items)}


@app.get("/api/v1/ops/support/tickets/{ticket_id}")
def ops_get_support_ticket(
    ticket_id: str,
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    ticket = find_support_ticket(ticket_id, db=db)
    if ticket is None:
        raise HTTPException(status_code=404, detail="Support ticket not found")

    return {
        "id": ticket["id"],
        "case_reference": ticket["case_reference"],
        "ticket_type": ticket["ticket_type"],
        "reporter_email": ticket["reporter_email"],
        "subject": ticket["subject"],
        "description": ticket["description"],
        "severity": ticket["severity"],
        "status": ticket["status"],
        "created_at": ticket["created_at"],
        "updated_at": ticket["updated_at"],
        "resolved_at": ticket.get("resolved_at"),
        "has_diagnostics": ticket.get("diagnostic_bundle") is not None,
        "notes": ticket.get("notes", []),
    }



def _run_ops_action_in_background(
    action_id: uuid.UUID,
    operator_identity: str,
    operator_role: str,
    bind: Any = None,
    helper_override: str | None = None,
) -> None:
    if not os.environ.get("OPS_ACTION_MOCK_EXECUTION"):
        # Grace period (200ms) ensuring the HTTP response is fully flushed to Next.js and client browser
        time.sleep(0.2)
    try:
        session_bind = bind or engine
        with DbSession(session_bind) as db:
            action = db.get(OpsAction, action_id)
            if not action:
                logger.error("Background ops action %s not found", action_id)
                return

            action = ops_actions_service.execute_action_sync(action, db, helper_override=helper_override)

            try:
                ops_audit(
                    db=db,
                    action=f"ops.action_{action.state}",
                    actor_identity=operator_identity,
                    actor_role=operator_role,
                    target_resource=f"action:{action.id}",
                    reason_code=(action.failure_code or "")[:64] if action.state != "succeeded" else None,
                    result="success" if action.state == "succeeded" else "failed",
                    metadata={
                        "action_type": action.action_type,
                        "failure_code": action.failure_code,
                        "deployed_revision": action.deployed_revision,
                    },
                )
                db.commit()
            except Exception as e:
                logger.warning(f"Error recording action completion audit: {e}")
                try:
                    db.rollback()
                except Exception:
                    pass
    except Exception as exc:
        logger.exception("Unexpected error in background ops action execution: %s", exc)


@app.post("/api/v1/ops/actions", response_model=OpsActionRead)
def ops_create_action(
    req: OpsActionCreate,
    background_tasks: BackgroundTasks,
    db: DbSession = Depends(get_db),
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("owner")),
):
    operator, session = auth

    confirm = (req.confirmation_text or req.confirmation or "").strip()
    if req.action_type in ("reboot_server", "restart_api"):
        expected = "REBOOT MOTORSIGNAL SERVER" if req.action_type == "reboot_server" else "RESTART API"
        if confirm != expected:
            raise HTTPException(status_code=422, detail="Invalid confirmation text.")

    # Mutual exclusion check: Only one active mutation at a time
    active = ops_actions_service.get_active_action(db)
    if active:
        raise HTTPException(
            status_code=409,
            detail=f"Another operations action ({active.action_type}) is currently active.",
        )

    # Idempotency lookup
    if req.idempotency_key:
        try:
            existing = db.scalars(
                select(OpsAction).where(
                    OpsAction.requested_by_id == operator.id,
                    OpsAction.action_type == req.action_type,
                    OpsAction.idempotency_key == req.idempotency_key,
                )
            ).first()
            if existing:
                return existing
        except Exception as e:
            logger.warning(f"Error checking idempotent ops action: {e}")
            try:
                db.rollback()
            except Exception:
                pass

    # Check helper availability unless mocked in tests
    if not ops_actions_service.is_helper_configured() and not os.environ.get("OPS_ACTION_MOCK_EXECUTION"):
        try:
            ops_audit(
                db=db,
                action="action_rejected",
                actor_identity=operator.identity,
                actor_role=operator.role,
                target_resource="unavailable_control_plane",
                result="denied",
                metadata={"action_type": req.action_type, "status": "rejected_unavailable"},
            )
            db.commit()
        except Exception as e:
            logger.warning(f"Error recording action rejection audit: {e}")
            try:
                db.rollback()
            except Exception:
                pass
        raise HTTPException(
            status_code=503,
            detail="Action control plane is currently unconfigured. Missing ops-action-helper.sh.",
        )

    # Create new action record
    now_utc = datetime.now(timezone.utc)
    try:
        action = OpsAction(
            id=uuid.uuid4(),
            action_type=req.action_type,
            requested_by_id=operator.id,
            request_timestamp=now_utc,
            start_timestamp=now_utc,
            state="running",
            idempotency_key=req.idempotency_key,
        )
        db.add(action)
        db.commit()
    except Exception as e:
        logger.warning(f"Error persisting ops action: {e}")
        try:
            db.rollback()
        except Exception:
            pass
        raise HTTPException(
            status_code=503,
            detail="Operations action storage is currently unavailable. Run database migrations.",
        )

    try:
        ops_audit(
            db=db,
            action="ops.action_requested",
            actor_identity=operator.identity,
            actor_role=operator.role,
            target_resource=f"action:{action.id}",
            metadata={"action_type": action.action_type},
        )
        db.commit()
    except Exception as e:
        logger.warning(f"Error recording ops.action_requested audit: {e}")
        try:
            db.rollback()
        except Exception:
            pass

    # Launch execution asynchronously in background
    background_tasks.add_task(
        _run_ops_action_in_background,
        action.id,
        operator.identity,
        operator.role,
        bind=db.get_bind(),
    )

    return action

@app.get("/api/v1/ops/actions", response_model=list[OpsActionRead])
def ops_list_actions(
    db: DbSession = Depends(get_db),
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    limit: int = Query(default=50, ge=1, le=100),
    offset: int = Query(default=0, ge=0),
):
    try:
        stmt = (
            select(OpsAction)
            .order_by(OpsAction.request_timestamp.desc(), OpsAction.id.desc())
            .offset(offset)
            .limit(limit)
        )
        return list(db.scalars(stmt).all())
    except Exception as e:
        logger.warning(f"Error listing ops actions: {e}")
        try:
            db.rollback()
        except Exception:
            pass
        return []

@app.get("/api/v1/ops/actions/capabilities")
def ops_get_action_capabilities(
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    active = ops_actions_service.get_active_action(db)
    return {
        "supported_actions": list(ops_actions_service.ALLOWED_ACTION_TYPES),
        "helper_configured": ops_actions_service.is_helper_configured() or bool(os.environ.get("OPS_ACTION_MOCK_EXECUTION")),
        "active_action_id": str(active.id) if active else None,
        "active_action_type": active.action_type if active else None,
    }

@app.get("/api/v1/ops/actions/{action_id}", response_model=OpsActionRead)
def ops_get_action(
    action_id: uuid.UUID,
    db: DbSession = Depends(get_db),
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
):
    action = db.get(OpsAction, action_id)
    if not action:
        raise HTTPException(status_code=404, detail="Action not found")
    return action

@app.post("/api/v1/ops/support/tickets", status_code=201)
def ops_create_support_ticket(
    request: OpsSupportTicketCreate,
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    operator, _ = auth
    ticket_type = request.ticket_type.strip().lower()
    if ticket_type not in ("bug_report", "contact_request", "telemetry_issue", "general"):
        raise HTTPException(status_code=422, detail="Invalid ticket type")

    severity = request.severity.strip().lower()
    if severity not in ("low", "medium", "high", "critical"):
        raise HTTPException(status_code=422, detail="Invalid severity level")

    bundle = None
    if request.diagnostic_bundle:
        if isinstance(request.diagnostic_bundle, BaseModel):
            bundle = request.diagnostic_bundle.model_dump()
        elif isinstance(request.diagnostic_bundle, dict):
            bundle = request.diagnostic_bundle

    now_dt = datetime.now(timezone.utc)
    now_iso = now_dt.isoformat()

    # Attempt up to 3 times to create with unique case_reference and commit cleanly
    for attempt in range(3):
        t_id_uuid = uuid.uuid4()
        ticket_id = str(t_id_uuid)
        case_ref = generate_support_case_reference(db=db)

        new_ticket_obj = OpsSupportTicket(
            id=t_id_uuid,
            case_reference=case_ref,
            ticket_type=ticket_type,
            reporter_email=request.reporter_email.strip().lower(),
            subject=request.subject.strip(),
            description=request.description.strip(),
            severity=severity,
            status="open",
            created_at=now_dt,
            updated_at=now_dt,
            resolved_at=None,
            diagnostic_bundle=bundle,
        )
        db.add(new_ticket_obj)

        new_ticket_dict = {
            "id": ticket_id,
            "case_reference": case_ref,
            "ticket_type": ticket_type,
            "reporter_email": request.reporter_email.strip().lower(),
            "subject": request.subject.strip(),
            "description": request.description.strip(),
            "severity": severity,
            "status": "open",
            "created_at": now_iso,
            "updated_at": now_iso,
            "resolved_at": None,
            "notes": [],
            "diagnostic_bundle": bundle,
        }

        ops_audit(
            db=db,
            action="support.ticket_created",
            actor_identity=operator.identity,
            actor_role=operator.role,
            target_resource=f"ticket:{ticket_id}",
            case_reference=case_ref,
            result="success",
        )

        try:
            db.commit()
            SUPPORT_TICKETS[ticket_id] = new_ticket_dict

            return {
                "id": ticket_id,
                "case_reference": case_ref,
                "ticket_type": ticket_type,
                "reporter_email": new_ticket_dict["reporter_email"],
                "subject": new_ticket_dict["subject"],
                "description": new_ticket_dict["description"],
                "severity": severity,
                "status": "open",
                "created_at": now_iso,
                "updated_at": now_iso,
                "resolved_at": None,
                "has_diagnostics": bundle is not None,
                "notes": [],
                "notes_count": 0,
            }
        except IntegrityError as ie:
            db.rollback()
            logger.warning(f"IntegrityError creating ticket attempt {attempt + 1}: {ie}")
            if attempt == 2:
                raise HTTPException(status_code=500, detail="Failed to generate unique case reference for support ticket")
            continue
        except Exception as e:
            db.rollback()
            logger.error(f"Failed to create support ticket: {e}")
            raise HTTPException(status_code=500, detail="Database error while creating support ticket")


@app.post("/api/v1/ops/support/tickets/{ticket_id}/notes", status_code=201)
def ops_add_support_ticket_note(
    ticket_id: str,
    request: OpsSupportTicketNoteCreate,
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    operator, _ = auth
    ticket = find_support_ticket(ticket_id, db=db)
    if ticket is None:
        raise HTTPException(status_code=404, detail="Support ticket not found")

    body_text = request.body.strip()
    if not body_text:
        raise HTTPException(status_code=422, detail="Note body cannot be empty")

    now_dt = datetime.now(timezone.utc)
    now_iso = now_dt.isoformat()
    note_id_uuid = uuid.uuid4()
    note_id = str(note_id_uuid)

    try:
        clean_id = ticket_id.strip()
        u = None
        try:
            u = uuid.UUID(clean_id)
        except ValueError:
            pass
        t_obj = db.scalar(select(OpsSupportTicket).where((OpsSupportTicket.id == u) if u else (OpsSupportTicket.case_reference == clean_id)))
        if t_obj is not None:
            n_obj = OpsSupportTicketNote(
                id=note_id_uuid,
                ticket_id=t_obj.id,
                author_identity=operator.identity,
                body=body_text,
                created_at=now_dt,
            )
            db.add(n_obj)
            t_obj.updated_at = now_dt
            db.flush()
    except Exception as e:
        logger.warning(f"Error persisting note to DB: {e}")

    note = {
        "id": note_id,
        "author_identity": operator.identity,
        "body": body_text,
        "created_at": now_iso,
    }
    ticket.setdefault("notes", []).append(note)
    ticket["updated_at"] = now_iso

    ops_audit(
        db=db,
        action="support.note_added",
        actor_identity=operator.identity,
        actor_role=operator.role,
        target_resource=f"ticket:{ticket['id']}",
        case_reference=ticket["case_reference"],
        result="success",
    )
    db.commit()

    return note


@app.patch("/api/v1/ops/support/tickets/{ticket_id}")
def ops_update_support_ticket(
    ticket_id: str,
    patch: OpsSupportTicketPatch,
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    operator, _ = auth
    ticket = find_support_ticket(ticket_id, db=db)
    if ticket is None:
        raise HTTPException(status_code=404, detail="Support ticket not found")

    if patch.status is None and patch.severity is None:
        raise HTTPException(status_code=422, detail="Must provide status or severity to update")

    now_dt = datetime.now(timezone.utc)
    now_iso = now_dt.isoformat()

    try:
        clean_id = ticket_id.strip()
        u = None
        try:
            u = uuid.UUID(clean_id)
        except ValueError:
            pass
        t_obj = db.scalar(select(OpsSupportTicket).where((OpsSupportTicket.id == u) if u else (OpsSupportTicket.case_reference == clean_id)))
        if t_obj is not None:
            if patch.status is not None:
                st = patch.status.strip().lower()
                if st not in ("open", "in_progress", "resolved", "closed"):
                    raise HTTPException(status_code=422, detail="Invalid ticket status")
                t_obj.status = st
                if st in ("resolved", "closed") and not t_obj.resolved_at:
                    t_obj.resolved_at = now_dt
                elif st in ("open", "in_progress"):
                    t_obj.resolved_at = None
            if patch.severity is not None:
                sev = patch.severity.strip().lower()
                if sev not in ("low", "medium", "high", "critical"):
                    raise HTTPException(status_code=422, detail="Invalid severity level")
                t_obj.severity = sev
            t_obj.updated_at = now_dt
    except HTTPException:
        raise
    except Exception as e:
        logger.warning(f"Error updating ticket in DB: {e}")

    if patch.status is not None:
        st = patch.status.strip().lower()
        if st not in ("open", "in_progress", "resolved", "closed"):
            raise HTTPException(status_code=422, detail="Invalid ticket status")
        ticket["status"] = st
        if st in ("resolved", "closed"):
            if not ticket.get("resolved_at"):
                ticket["resolved_at"] = now_iso
        elif st in ("open", "in_progress"):
            ticket["resolved_at"] = None

    if patch.severity is not None:
        sev = patch.severity.strip().lower()
        if sev not in ("low", "medium", "high", "critical"):
            raise HTTPException(status_code=422, detail="Invalid severity level")
        ticket["severity"] = sev

    ticket["updated_at"] = now_iso

    ops_audit(
        db=db,
        action="support.ticket_updated",
        actor_identity=operator.identity,
        actor_role=operator.role,
        target_resource=f"ticket:{ticket['id']}",
        case_reference=ticket["case_reference"],
        result="success",
        metadata={"status": ticket["status"], "severity": ticket["severity"]},
    )
    db.commit()

    return {
        "id": ticket["id"],
        "case_reference": ticket["case_reference"],
        "ticket_type": ticket["ticket_type"],
        "reporter_email": ticket["reporter_email"],
        "subject": ticket["subject"],
        "description": ticket["description"],
        "severity": ticket["severity"],
        "status": ticket["status"],
        "created_at": ticket["created_at"],
        "updated_at": ticket["updated_at"],
        "resolved_at": ticket.get("resolved_at"),
        "has_diagnostics": ticket.get("diagnostic_bundle") is not None,
        "notes": ticket.get("notes", []),
    }


@app.get("/api/v1/ops/support/tickets/{ticket_id}/diagnostics")
def ops_get_support_ticket_diagnostics(
    ticket_id: str,
    reason: str = Query(..., min_length=3),
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    operator, _ = auth
    ticket = find_support_ticket(ticket_id, db=db)
    if ticket is None:
        raise HTTPException(status_code=404, detail="Support ticket not found")

    bundle = ticket.get("diagnostic_bundle")
    if bundle is None:
        raise HTTPException(status_code=404, detail="No diagnostic bundle found for this ticket")

    ops_audit(
        db=db,
        action="support.diagnostics_viewed",
        actor_identity=operator.identity,
        actor_role=operator.role,
        target_resource=f"ticket:{ticket['id']}",
        reason_code=reason.strip(),
        case_reference=ticket["case_reference"],
        result="success",
    )
    db.commit()

    return bundle


# ==============================================================================
# SECTION 4: WEBSITE CONTENT CMS (OPERATIONS)
# ==============================================================================

class OpsCmsUpdateRequest(BaseModel):
    action: Literal["draft", "publish"]
    content: dict
    change_summary: str = Field(default="Content updated")


class OpsCmsRollbackRequest(BaseModel):
    revision_id: str
    reason: str = Field(min_length=3)


CMS_DEFAULT_DOCUMENTS: dict[str, dict] = {
    "homepage_hero": {
        "slug": "homepage_hero",
        "title": "Homepage Hero",
        "description": "Main landing hero copy, value proposition headline, and primary call-to-action.",
        "published": {
            "headline": "Know what your ride is doing.",
            "subheadline": "MotorSignal turns every ride into clear, actionable insight: real-time lean angle, live GPS track telemetry on the road, post-session performance replay, and enterprise security for riders and clubs.",
            "cta_text": "Create your account",
            "cta_link": "/register",
            "badge_text": "⚡ Next-Gen Telemetry Engine 3.0 • Real-Time CAN-Bus & GPS",
        },
        "draft": None,
        "updated_at": "2026-09-12T12:00:00Z",
        "updated_by": "system",
        "revisions": [
            {
                "revision_id": "rev-hero-init-001",
                "created_at": "2026-09-12T12:00:00Z",
                "created_by": "system",
                "status": "published",
                "change_summary": "Initial baseline published content",
                "content": {
                    "headline": "Know what your ride is doing.",
                    "subheadline": "MotorSignal turns every ride into clear, actionable insight: real-time lean angle, live GPS track telemetry on the road, post-session performance replay, and enterprise security for riders and clubs.",
                    "cta_text": "Create your account",
                    "cta_link": "/register",
                    "badge_text": "⚡ Next-Gen Telemetry Engine 3.0 • Real-Time CAN-Bus & GPS",
                },
            }
        ],
    },
    "features": {
        "slug": "features",
        "title": "Key Features",
        "description": "High-level feature cards detailing 20Hz sensor fusion, automatic ride gate, and telemetry.",
        "published": {
            "items": [
                {
                    "title": "Live Cockpit & Handlebar HUD",
                    "description": "Real-time pitch, roll, yaw, throttle position, braking pressure, and predictive apex speeds delivered at 60fps to your handlebar display or phone mount without visual drag.",
                    "icon": "dashboard_customize",
                },
                {
                    "title": "Ride Intelligence & Session Replay",
                    "description": "Multi-layer track analysis with sector split times, deceleration zones, elevation profiles, and 3D GPS satellite trace replays after every canyon run or track session.",
                    "icon": "analytics",
                },
                {
                    "title": "Club Telemetry & Fleet Beacon",
                    "description": "Live pack radar tracking, automatic crash detection beacon with SOS emergency dispatch, and competitive leaderboards with verified telemetry data.",
                    "icon": "emergency",
                },
                {
                    "title": "Zero-Friction Hardware & Security",
                    "description": "Encrypted zero-knowledge ride logs, offline Bluetooth / CAN-bus hardware sync, and end-to-end authenticated APIs compatible with major manufacturers.",
                    "icon": "memory",
                },
            ]
        },
        "draft": None,
        "updated_at": "2026-09-12T12:00:00Z",
        "updated_by": "system",
        "revisions": [
            {
                "revision_id": "rev-features-init-001",
                "created_at": "2026-09-12T12:00:00Z",
                "created_by": "system",
                "status": "published",
                "change_summary": "Initial baseline published content",
                "content": {
                    "items": [
                        {
                            "title": "Live Cockpit & Handlebar HUD",
                            "description": "Real-time pitch, roll, yaw, throttle position, braking pressure, and predictive apex speeds delivered at 60fps to your handlebar display or phone mount without visual drag.",
                            "icon": "dashboard_customize",
                        },
                        {
                            "title": "Ride Intelligence & Session Replay",
                            "description": "Multi-layer track analysis with sector split times, deceleration zones, elevation profiles, and 3D GPS satellite trace replays after every canyon run or track session.",
                            "icon": "analytics",
                        },
                        {
                            "title": "Club Telemetry & Fleet Beacon",
                            "description": "Live pack radar tracking, automatic crash detection beacon with SOS emergency dispatch, and competitive leaderboards with verified telemetry data.",
                            "icon": "emergency",
                        },
                        {
                            "title": "Zero-Friction Hardware & Security",
                            "description": "Encrypted zero-knowledge ride logs, offline Bluetooth / CAN-bus hardware sync, and end-to-end authenticated APIs compatible with major manufacturers.",
                            "icon": "memory",
                        },
                    ]
                },
            }
        ],
    },
    "screenshots": {
        "slug": "screenshots",
        "title": "App Screenshots & Captions",
        "description": "App gallery visuals showcasing live dashboard and interactive turn analysis.",
        "published": {
            "items": [
                {
                    "title": "Live Dashboard",
                    "caption": "Real-time lean gauge and cockpit telemetry",
                    "image_url": "/screenshots/dashboard.png",
                },
                {
                    "title": "Ride Forensic Analysis",
                    "caption": "Interactive turn analysis with braking points",
                    "image_url": "/screenshots/analysis.png",
                },
            ]
        },
        "draft": None,
        "updated_at": "2026-09-12T12:00:00Z",
        "updated_by": "system",
        "revisions": [
            {
                "revision_id": "rev-screenshots-init-001",
                "created_at": "2026-09-12T12:00:00Z",
                "created_by": "system",
                "status": "published",
                "change_summary": "Initial baseline published content",
                "content": {
                    "items": [
                        {
                            "title": "Live Dashboard",
                            "caption": "Real-time lean gauge and cockpit telemetry",
                            "image_url": "/screenshots/dashboard.png",
                        },
                        {
                            "title": "Ride Forensic Analysis",
                            "caption": "Interactive turn analysis with braking points",
                            "image_url": "/screenshots/analysis.png",
                        },
                    ]
                },
            }
        ],
    },
    "faqs": {
        "slug": "faqs",
        "title": "Frequently Asked Questions",
        "description": "Hardware mounting guidelines and offline data recording details for riders.",
        "published": {
            "items": [
                {
                    "question": "What mounting setup is required?",
                    "answer": "Any rigid phone mount such as QuadLock or RAM Mount positioned in landscape or portrait.",
                },
                {
                    "question": "Does it work without cellular signal?",
                    "answer": "Yes, MotorSignal records 100% locally offline and syncs when connectivity returns.",
                },
            ]
        },
        "draft": None,
        "updated_at": "2026-09-12T12:00:00Z",
        "updated_by": "system",
        "revisions": [
            {
                "revision_id": "rev-faqs-init-001",
                "created_at": "2026-09-12T12:00:00Z",
                "created_by": "system",
                "status": "published",
                "change_summary": "Initial baseline published content",
                "content": {
                    "items": [
                        {
                            "question": "What mounting setup is required?",
                            "answer": "Any rigid phone mount such as QuadLock or RAM Mount positioned in landscape or portrait.",
                        },
                        {
                            "question": "Does it work without cellular signal?",
                            "answer": "Yes, MotorSignal records 100% locally offline and syncs when connectivity returns.",
                        },
                    ]
                },
            }
        ],
    },
    "download_links": {
        "slug": "download_links",
        "title": "Download & Store Links",
        "description": "Direct APK download URL, Google Play store link, and latest production version tag.",
        "published": {
            "android_url": "https://motorsignal.app/builds/motorsignal-release.apk",
            "testflight_url": "",
            "play_store_url": "https://play.google.com/store/apps/details?id=app.motorsignal",
            "version_name": "1.0.4",
            "build_number": 42,
        },
        "draft": None,
        "updated_at": "2026-09-12T12:00:00Z",
        "updated_by": "system",
        "revisions": [
            {
                "revision_id": "rev-downloads-init-001",
                "created_at": "2026-09-12T12:00:00Z",
                "created_by": "system",
                "status": "published",
                "change_summary": "Initial baseline published content",
                "content": {
                    "android_url": "https://motorsignal.app/builds/motorsignal-release.apk",
                    "testflight_url": "",
                    "play_store_url": "https://play.google.com/store/apps/details?id=app.motorsignal",
                    "version_name": "1.0.4",
                    "build_number": 42,
                },
            }
        ],
    },
    "changelog": {
        "slug": "changelog",
        "title": "Release Changelog",
        "description": "Public release notes, version dates, and acquisition refinement summaries.",
        "published": {
            "releases": [
                {
                    "version": "v1.0.4",
                    "release_date": "2026-09-12",
                    "title": "Operations Platform V1",
                    "notes": "Added real-time system health, telemetry diagnostics, and improved offline sync.",
                },
                {
                    "version": "v1.0.3",
                    "release_date": "2026-09-01",
                    "title": "20Hz Acquisition Refinements",
                    "notes": "Optimized sensor fusion filter latency.",
                },
            ]
        },
        "draft": None,
        "updated_at": "2026-09-12T12:00:00Z",
        "updated_by": "system",
        "revisions": [
            {
                "revision_id": "rev-changelog-init-001",
                "created_at": "2026-09-12T12:00:00Z",
                "created_by": "system",
                "status": "published",
                "change_summary": "Initial baseline published content",
                "content": {
                    "releases": [
                        {
                            "version": "v1.0.4",
                            "release_date": "2026-09-12",
                            "title": "Operations Platform V1",
                            "notes": "Added real-time system health, telemetry diagnostics, and improved offline sync.",
                        },
                        {
                            "version": "v1.0.3",
                            "release_date": "2026-09-01",
                            "title": "20Hz Acquisition Refinements",
                            "notes": "Optimized sensor fusion filter latency.",
                        },
                    ]
                },
            }
        ],
    },
    "policy_pages": {
        "slug": "policy_pages",
        "title": "Legal & Policy Pages",
        "description": "Rider telemetry data minimization privacy policy and terms of service terms.",
        "published": {
            "privacy_policy": "MotorSignal operates under strict data minimization. Rider telemetry is private and never sold.",
            "terms_of_service": "MotorSignal telemetry is for closed-course performance analysis and recreational touring.",
        },
        "draft": None,
        "updated_at": "2026-09-12T12:00:00Z",
        "updated_by": "system",
        "revisions": [
            {
                "revision_id": "rev-policies-init-001",
                "created_at": "2026-09-12T12:00:00Z",
                "created_by": "system",
                "status": "published",
                "change_summary": "Initial baseline published content",
                "content": {
                    "privacy_policy": "MotorSignal operates under strict data minimization. Rider telemetry is private and never sold.",
                    "terms_of_service": "MotorSignal telemetry is for closed-course performance analysis and recreational touring.",
                },
            }
        ],
    },
}

# Mutable in-memory store initialized from defaults
CMS_DOCUMENTS: dict[str, dict] = {
    k: {
        "slug": v["slug"],
        "title": v["title"],
        "description": v["description"],
        "published": dict(v["published"]) if isinstance(v["published"], dict) else v["published"],
        "draft": dict(v["draft"]) if isinstance(v["draft"], dict) else None,
        "updated_at": v["updated_at"],
        "updated_by": v["updated_by"],
        "revisions": list(v["revisions"]),
    }
    for k, v in CMS_DEFAULT_DOCUMENTS.items()
}


def _seed_cms_documents_if_needed(db: DbSession) -> None:
    try:
        existing_slugs = set(db.scalars(select(OpsCmsDocument.slug)).all())
        now_dt = datetime.now(timezone.utc)
        added_any = False
        for slug, doc in CMS_DEFAULT_DOCUMENTS.items():
            if slug not in existing_slugs:
                doc_obj = OpsCmsDocument(
                    slug=slug,
                    title=doc["title"],
                    description=doc.get("description", ""),
                    published_json=copy.deepcopy(doc["published"]),
                    draft_json=copy.deepcopy(doc.get("draft")) if doc.get("draft") else None,
                    updated_at=now_dt,
                    updated_by=doc.get("updated_by", "system"),
                )
                db.add(doc_obj)
                for rev in doc.get("revisions", []):
                    rev_obj = OpsCmsRevision(
                        revision_id=rev["revision_id"],
                        document_slug=slug,
                        status=rev["status"],
                        change_summary=rev["change_summary"],
                        content_json=copy.deepcopy(rev["content"]),
                        created_at=now_dt,
                        created_by=rev.get("created_by", "system"),
                    )
                    db.add(rev_obj)
                added_any = True
        if added_any:
            db.commit()
    except Exception as e:
        logger.warning(f"Error seeding CMS documents to DB: {e}")
        db.rollback()


@app.get("/api/v1/cms/documents/{slug}")
def get_public_cms_document(
    slug: str,
    db: DbSession = Depends(get_db),
):
    clean_slug = slug.strip()
    _seed_cms_documents_if_needed(db)
    try:
        doc_record = db.scalar(select(OpsCmsDocument).where(OpsCmsDocument.slug == clean_slug))
        if doc_record and doc_record.published_json:
            return {
                "slug": doc_record.slug,
                "title": doc_record.title,
                "published": doc_record.published_json,
                "updated_at": doc_record.updated_at.isoformat() if doc_record.updated_at else None,
            }
    except Exception as e:
        logger.warning(f"Error querying public CMS doc from DB: {e}")
    if clean_slug in CMS_DOCUMENTS:
        doc = CMS_DOCUMENTS[clean_slug]
        return {
            "slug": doc["slug"],
            "title": doc["title"],
            "published": doc["published"],
            "updated_at": doc["updated_at"],
        }
    raise HTTPException(status_code=404, detail="Document not found")


@app.get("/api/v1/ops/cms/documents")
def ops_get_cms_documents(
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    _seed_cms_documents_if_needed(db)
    try:
        db_docs = db.scalars(select(OpsCmsDocument).order_by(OpsCmsDocument.slug)).all()
        if db_docs:
            return [
                {
                    "slug": d.slug,
                    "title": d.title,
                    "description": d.description,
                    "has_draft": d.draft_json is not None,
                    "updated_at": d.updated_at.isoformat() if d.updated_at else datetime.now(timezone.utc).isoformat(),
                    "updated_by": d.updated_by,
                }
                for d in db_docs
            ]
    except Exception as e:
        logger.warning(f"Error fetching CMS documents from DB: {e}")

    summaries = []
    for doc in CMS_DOCUMENTS.values():
        summaries.append({
            "slug": doc["slug"],
            "title": doc["title"],
            "description": doc["description"],
            "has_draft": doc["draft"] is not None,
            "updated_at": doc["updated_at"],
            "updated_by": doc["updated_by"],
        })
    return summaries


@app.get("/api/v1/ops/cms/documents/{slug}")
def ops_get_cms_document_detail(
    slug: str,
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    clean_slug = slug.strip()
    _seed_cms_documents_if_needed(db)
    try:
        doc_record = db.scalar(select(OpsCmsDocument).where(OpsCmsDocument.slug == clean_slug))
        if doc_record:
            rev_records = db.scalars(
                select(OpsCmsRevision)
                .where(OpsCmsRevision.document_slug == clean_slug)
                .order_by(OpsCmsRevision.created_at.desc())
            ).all()
            revisions = [
                {
                    "revision_id": r.revision_id,
                    "created_at": r.created_at.isoformat() if r.created_at else "",
                    "created_by": r.created_by,
                    "status": r.status,
                    "change_summary": r.change_summary,
                    "content": r.content_json,
                }
                for r in rev_records
            ]
            return {
                "slug": doc_record.slug,
                "title": doc_record.title,
                "description": doc_record.description,
                "published": doc_record.published_json,
                "draft": doc_record.draft_json,
                "has_draft": doc_record.draft_json is not None,
                "updated_at": doc_record.updated_at.isoformat() if doc_record.updated_at else "",
                "updated_by": doc_record.updated_by,
                "revisions": revisions,
            }
    except Exception as e:
        logger.warning(f"Error fetching CMS detail from DB: {e}")

    if clean_slug not in CMS_DOCUMENTS:
        raise HTTPException(status_code=404, detail="CMS document not found")

    doc = CMS_DOCUMENTS[clean_slug]
    return {
        "slug": doc["slug"],
        "title": doc["title"],
        "description": doc["description"],
        "published": doc["published"],
        "draft": doc["draft"],
        "has_draft": doc["draft"] is not None,
        "updated_at": doc["updated_at"],
        "updated_by": doc["updated_by"],
        "revisions": doc["revisions"],
    }


@app.put("/api/v1/ops/cms/documents/{slug}")
def ops_update_cms_document(
    slug: str,
    body: OpsCmsUpdateRequest,
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    operator, _ = auth
    clean_slug = slug.strip()
    _seed_cms_documents_if_needed(db)
    if clean_slug not in CMS_DEFAULT_DOCUMENTS and clean_slug not in CMS_DOCUMENTS:
        doc_record = db.scalar(select(OpsCmsDocument).where(OpsCmsDocument.slug == clean_slug))
        if not doc_record:
            raise HTTPException(status_code=404, detail="CMS document not found")

    now_dt = datetime.now(timezone.utc)
    now_str = now_dt.isoformat()
    rev_id = f"rev-{clean_slug[:8]}-{uuid.uuid4().hex[:8]}"
    summary_text = body.change_summary.strip() or ("Draft saved" if body.action == "draft" else "Published live")

    try:
        doc_record = db.scalar(select(OpsCmsDocument).where(OpsCmsDocument.slug == clean_slug))
        if not doc_record:
            doc_def = CMS_DEFAULT_DOCUMENTS.get(clean_slug, CMS_DOCUMENTS.get(clean_slug, {}))
            doc_record = OpsCmsDocument(
                slug=clean_slug,
                title=doc_def.get("title", clean_slug.replace("_", " ").title()),
                description=doc_def.get("description", ""),
                published_json=copy.deepcopy(body.content) if body.action == "publish" else copy.deepcopy(doc_def.get("published", {})),
                draft_json=copy.deepcopy(body.content) if body.action == "draft" else None,
                updated_at=now_dt,
                updated_by=operator.identity,
            )
            db.add(doc_record)
            db.flush()
        else:
            doc_record.updated_at = now_dt
            doc_record.updated_by = operator.identity
            if body.action == "draft":
                doc_record.draft_json = copy.deepcopy(body.content)
                flag_modified(doc_record, "draft_json")
            elif body.action == "publish":
                doc_record.published_json = copy.deepcopy(body.content)
                doc_record.draft_json = None
                flag_modified(doc_record, "published_json")
                flag_modified(doc_record, "draft_json")

        rev_record = OpsCmsRevision(
            revision_id=rev_id,
            document_slug=clean_slug,
            created_at=now_dt,
            created_by=operator.identity,
            status="draft" if body.action == "draft" else "published",
            change_summary=summary_text,
            content_json=copy.deepcopy(body.content),
        )
        db.add(rev_record)
        db.flush()
    except Exception as e:
        logger.warning(f"Error persisting CMS update to DB: {e}")
        db.rollback()

    doc = CMS_DOCUMENTS.setdefault(clean_slug, {
        "slug": clean_slug,
        "title": clean_slug.replace("_", " ").title(),
        "description": "",
        "published": {},
        "draft": None,
        "updated_at": now_str,
        "updated_by": operator.identity,
        "revisions": [],
    })

    if body.action == "draft":
        doc["draft"] = body.content
        doc["updated_at"] = now_str
        doc["updated_by"] = operator.identity

        rev = {
            "revision_id": rev_id,
            "created_at": now_str,
            "created_by": operator.identity,
            "status": "draft",
            "change_summary": summary_text,
            "content": body.content,
        }
        doc["revisions"].insert(0, rev)

        ops_audit(
            db=db,
            action="cms.draft_saved",
            actor_identity=operator.identity,
            actor_role=operator.role,
            target_resource=f"cms:{clean_slug}",
            reason_code="content_draft_update",
            result="success",
            metadata={"slug": clean_slug, "revision_id": rev_id, "summary": summary_text},
        )
        db.commit()

    elif body.action == "publish":
        doc["published"] = body.content
        doc["draft"] = None
        doc["updated_at"] = now_str
        doc["updated_by"] = operator.identity

        rev = {
            "revision_id": rev_id,
            "created_at": now_str,
            "created_by": operator.identity,
            "status": "published",
            "change_summary": summary_text,
            "content": body.content,
        }
        doc["revisions"].insert(0, rev)

        ops_audit(
            db=db,
            action="cms.published",
            actor_identity=operator.identity,
            actor_role=operator.role,
            target_resource=f"cms:{clean_slug}",
            reason_code="content_published",
            result="success",
            metadata={"slug": clean_slug, "revision_id": rev_id, "summary": summary_text},
        )
        db.commit()

    return {
        "slug": doc["slug"],
        "title": doc["title"],
        "description": doc["description"],
        "published": doc["published"],
        "draft": doc["draft"],
        "has_draft": doc["draft"] is not None,
        "updated_at": doc["updated_at"],
        "updated_by": doc["updated_by"],
        "revisions": doc["revisions"],
    }


@app.post("/api/v1/ops/cms/documents/{slug}/rollback")
def ops_rollback_cms_document(
    slug: str,
    body: OpsCmsRollbackRequest,
    auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("operator", "owner")),
    db: DbSession = Depends(get_db),
):
    operator, _ = auth
    clean_slug = slug.strip()
    _seed_cms_documents_if_needed(db)

    now_dt = datetime.now(timezone.utc)
    now_str = now_dt.isoformat()
    rev_id = f"rev-{clean_slug[:8]}-{uuid.uuid4().hex[:8]}"

    target_content = None
    try:
        rev_record = db.scalar(
            select(OpsCmsRevision).where(
                OpsCmsRevision.document_slug == clean_slug,
                OpsCmsRevision.revision_id == body.revision_id.strip()
            )
        )
        if rev_record:
            target_content = copy.deepcopy(rev_record.content_json)
            doc_record = db.scalar(select(OpsCmsDocument).where(OpsCmsDocument.slug == clean_slug))
            if not doc_record:
                doc_def = CMS_DEFAULT_DOCUMENTS.get(clean_slug, CMS_DOCUMENTS.get(clean_slug, {}))
                doc_record = OpsCmsDocument(
                    slug=clean_slug,
                    title=doc_def.get("title", clean_slug.replace("_", " ").title()),
                    description=doc_def.get("description", ""),
                    published_json=target_content,
                    draft_json=None,
                    updated_at=now_dt,
                    updated_by=operator.identity,
                )
                db.add(doc_record)
                db.flush()
            else:
                doc_record.published_json = target_content
                doc_record.draft_json = None
                doc_record.updated_at = now_dt
                doc_record.updated_by = operator.identity
                flag_modified(doc_record, "published_json")
                flag_modified(doc_record, "draft_json")

            new_rev = OpsCmsRevision(
                revision_id=rev_id,
                document_slug=clean_slug,
                created_at=now_dt,
                created_by=operator.identity,
                status="published",
                change_summary=f"Rollback to {body.revision_id.strip()}: {body.reason.strip()}",
                content_json=target_content,
            )
            db.add(new_rev)
            db.flush()
    except Exception as e:
        logger.warning(f"Error rolling back CMS document in DB: {e}")
        db.rollback()

    if clean_slug not in CMS_DOCUMENTS:
        if not target_content:
            raise HTTPException(status_code=404, detail="CMS document not found")

    doc = CMS_DOCUMENTS.get(clean_slug)
    if doc:
        target_rev = next((r for r in doc["revisions"] if r["revision_id"] == body.revision_id.strip()), None)
        if not target_rev and not target_content:
            raise HTTPException(status_code=404, detail="Revision not found")
        restored_content = target_content or (dict(target_rev["content"]) if target_rev else {})

        doc["published"] = restored_content
        doc["draft"] = None
        doc["updated_at"] = now_str
        doc["updated_by"] = operator.identity

        rev = {
            "revision_id": rev_id,
            "created_at": now_str,
            "created_by": operator.identity,
            "status": "published",
            "change_summary": f"Rollback to {body.revision_id.strip()}: {body.reason.strip()}",
            "content": restored_content,
        }
        doc["revisions"].insert(0, rev)

    ops_audit(
        db=db,
        action="cms.revision_restored",
        actor_identity=operator.identity,
        actor_role=operator.role,
        target_resource=f"cms:{clean_slug}",
        reason_code=body.reason.strip(),
        result="success",
        metadata={
            "slug": clean_slug,
            "restored_revision_id": body.revision_id.strip(),
            "new_revision_id": rev_id,
        },
    )
    db.commit()

    return {
        "slug": doc["slug"] if doc else clean_slug,
        "title": doc["title"] if doc else clean_slug,
        "description": doc["description"] if doc else "",
        "published": doc["published"] if doc else target_content,
        "draft": doc["draft"] if doc else None,
        "has_draft": False,
        "updated_at": doc["updated_at"] if doc else now_str,
        "updated_by": doc["updated_by"] if doc else operator.identity,
        "revisions": doc["revisions"] if doc else [],
    }