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 [], }