import uuid import logging import os import smtplib import secrets import hashlib from datetime import datetime, timedelta, timezone from email.message import EmailMessage from typing import Literal from email_validator import EmailNotValidError, validate_email from fastapi import Depends, FastAPI, Header, HTTPException, Query, Request, status from pydantic import BaseModel, ConfigDict, Field, field_validator from sqlalchemy import func, or_, select, text from sqlalchemy.orm import Session as DbSession 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, PasswordResetToken, Ride, Session, Subscription, TelemetryPoint, User, Vehicle, ) from infrastructure_metrics import get_infrastructure_metrics 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, ) @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) 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) 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(): now_iso = datetime.now(timezone.utc).isoformat() db_ok = False try: db_ok = check_database() except Exception: db_ok = False overall = "available" if db_ok 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": "__Host-motorsignal_ops token verification & authority", }, "audit_persistence": { "status": "available" if db_ok else "degraded", "description": "Immutable operational security activity persistence", }, }, } @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: db_ok = check_database() except Exception: db_ok = False return get_infrastructure_metrics(database_ok=db_ok) @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) total_users = db.scalar(select(func.count()).select_from(User)) or 0 new_users_7d = db.scalar(select(func.count()).select_from(User).where(User.created_at >= t_7d)) or 0 active_users_24h = db.scalar( select(func.count(func.distinct(Session.user_id))) .where(Session.last_used_at >= t_24h, Session.revoked_at.is_(None)) ) or 0 active_users_7d = db.scalar( select(func.count(func.distinct(Session.user_id))) .where(Session.last_used_at >= t_7d, Session.revoked_at.is_(None)) ) or 0 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_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 web_logins_24h = db.scalar( select(func.count()).select_from(AuditEvent) .where(AuditEvent.event_type == "auth.login", AuditEvent.created_at >= t_24h) ) or 0 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 open_incidents = db.scalars( select(OpsIncident) .where(OpsIncident.status.in_(["open", "investigating", "monitoring"])) .order_by(OpsIncident.created_at.desc()) ).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") recent_open = [ { "case_reference": inc.case_reference, "title": inc.title, "severity": inc.severity, "status": inc.status, "created_at": inc.created_at.isoformat(), "updated_at": inc.updated_at.isoformat(), } for inc in open_incidents[:5] ] error_events = db.scalars( select(OpsAuditEvent) .where(OpsAuditEvent.result.in_(["failure", "denied"])) .order_by(OpsAuditEvent.timestamp.desc()) .limit(5) ).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, } for e in error_events ] return { "generated_at": now.isoformat(), "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, }, "incidents": { "total_open": total_open, "critical_count": crit_count, "high_count": high_count, "medium_count": med_count, "low_count": low_count, "recent_open": recent_open, }, "errors": { "errors_24h": errors_24h, "denials_24h": denials_24h, "recent": recent_errors, }, "health": ops_get_health(), } @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, "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, "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}, ) 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, "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(), } # ============================================================ # Operations (ops) - Section 3: Support Center & Diagnostics # ============================================================ _support_case_seq = 100 def generate_support_case_reference() -> str: global _support_case_seq year = datetime.now(timezone.utc).year _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 find_support_ticket(ticket_id: str) -> dict | None: clean_id = ticket_id.strip() 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), ): 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) 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", []), } @app.post("/api/v1/ops/actions", response_model=OpsActionRead) def ops_create_action( req: OpsActionCreate, db: DbSession = Depends(get_db), auth: tuple[Operator, OpsSession] = Depends(require_ops_roles("owner")), ): operator, session = auth if req.action_type in ("reboot_server", "restart_api"): expected = "REBOOT MOTORSIGNAL SERVER" if req.action_type == "reboot_server" else "RESTART API" if req.confirmation_text != 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: 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 # Check helper availability unless mocked in tests if not ops_actions_service.is_helper_configured() and not os.environ.get("OPS_ACTION_MOCK_EXECUTION"): 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() raise HTTPException( status_code=503, detail="Action control plane is currently unconfigured. Missing ops-action-helper.sh.", ) # Create new action record action = OpsAction( id=uuid.uuid4(), action_type=req.action_type, requested_by_id=operator.id, request_timestamp=datetime.now(timezone.utc), state="queued", idempotency_key=req.idempotency_key, ) db.add(action) db.commit() 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() # Execute action via helper service action = ops_actions_service.execute_action_sync(action, db) ops_audit( db=db, action=f"ops.action_{action.state}", actor_identity=operator.identity, actor_role=operator.role, target_resource=f"action:{action.id}", 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() 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), ): return db.scalars(select(OpsAction).order_by(OpsAction.request_timestamp.desc(), OpsAction.id.desc()).offset(offset).limit(limit)).all() @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") ticket_id = str(uuid.uuid4()) case_ref = generate_support_case_reference() now_iso = datetime.now(timezone.utc).isoformat() 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 new_ticket = { "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, } SUPPORT_TICKETS[ticket_id] = new_ticket 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", ) db.commit() return { "id": new_ticket["id"], "case_reference": new_ticket["case_reference"], "ticket_type": new_ticket["ticket_type"], "reporter_email": new_ticket["reporter_email"], "subject": new_ticket["subject"], "description": new_ticket["description"], "severity": new_ticket["severity"], "status": new_ticket["status"], "created_at": new_ticket["created_at"], "updated_at": new_ticket["updated_at"], "resolved_at": new_ticket["resolved_at"], "has_diagnostics": new_ticket["diagnostic_bundle"] is not None, "notes": new_ticket["notes"], } @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) 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_iso = datetime.now(timezone.utc).isoformat() note_id = str(uuid.uuid4()) 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) 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_iso = datetime.now(timezone.utc).isoformat() 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) 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": "Precision Telemetry for Every Corner", "subheadline": "Professional-grade motorcycle lean angle, 20Hz dynamics, and track analysis directly on your smartphone.", "cta_text": "Download App", "cta_link": "/download", "badge_text": "Engineered for Riders", }, "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": "Precision Telemetry for Every Corner", "subheadline": "Professional-grade motorcycle lean angle, 20Hz dynamics, and track analysis directly on your smartphone.", "cta_text": "Download App", "cta_link": "/download", "badge_text": "Engineered for Riders", }, } ], }, "features": { "slug": "features", "title": "Key Features", "description": "High-level feature cards detailing 20Hz sensor fusion, automatic ride gate, and telemetry.", "published": { "items": [ { "title": "20Hz Lean & Dynamics", "description": "Sub-50ms authoritative sensor fusion derived lean angle.", "icon": "speed", }, { "title": "Automatic Ride Detection", "description": "Zero-touch start and stop with 3.0 MPH moving gate.", "icon": "auto_mode", }, { "title": "Track & Road Telemetry", "description": "Forensic high-rate GPS & IMU playback with corner metrics.", "icon": "timeline", }, ] }, "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": "20Hz Lean & Dynamics", "description": "Sub-50ms authoritative sensor fusion derived lean angle.", "icon": "speed", }, { "title": "Automatic Ride Detection", "description": "Zero-touch start and stop with 3.0 MPH moving gate.", "icon": "auto_mode", }, { "title": "Track & Road Telemetry", "description": "Forensic high-rate GPS & IMU playback with corner metrics.", "icon": "timeline", }, ] }, } ], }, "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() } @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), ): 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() 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() if clean_slug not in CMS_DOCUMENTS: raise HTTPException(status_code=404, detail="CMS document not found") doc = CMS_DOCUMENTS[clean_slug] now_str = datetime.now(timezone.utc).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") 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() if clean_slug not in CMS_DOCUMENTS: raise HTTPException(status_code=404, detail="CMS document not found") doc = CMS_DOCUMENTS[clean_slug] target_rev = next((r for r in doc["revisions"] if r["revision_id"] == body.revision_id.strip()), None) if not target_rev: raise HTTPException(status_code=404, detail="Revision not found") now_str = datetime.now(timezone.utc).isoformat() rev_id = f"rev-{clean_slug[:8]}-{uuid.uuid4().hex[:8]}" restored_content = dict(target_rev["content"]) 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"], "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"], }