""" Application Entry Point What: Boots the FastAPI application, mounts all routers, configures global middleware, and exposes framework-level utility endpoints such as health check and the SSE stream. How: The module creates the `FastAPI` app, configures rate limiting and CORS, imports the domain routers, and registers them with the application. It intentionally stays thin and leaves domain behavior to routers, services, and repositories. Usage: Used as the backend startup module when the FastAPI app is launched, whether by an ASGI server importing `app` or by running `python main.py` during local development. What it does: - Creates the FastAPI application instance - Configures CORS and rate-limit exception handling - Mounts all router groups - Exposes a health-check endpoint - Exposes the realtime SSE stream endpoint What it does NOT do: - Does not contain business logic - Does not execute domain-specific database work - Does not validate domain payloads itself - Does not replace routers or services as an orchestration layer """ import asyncio import contextlib import json import logging import os import secrets import time from pathlib import Path import jwt from fastapi import Depends, FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import Response, StreamingResponse from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded # Domain routers — each owns its own slice of the API surface from app.routers import auth, products, categories, loyalty, reservations, settings, orders, customer, rewards, cms, reports, seats, materials, notifications, feedback, stalls from app.routers.admin import audit_router, inventory_router, users_router, apk_router from app.routers.inventory_allocation import router as inventory_allocation_router from app.repositories.reservations_repo import ReservationRepository from app.repositories.settings_repo import SettingsRepository from app.repositories.seats_repo import SeatsRepository from app.realtime_manager import broker from app.auth import get_current_user, get_user_from_token, resolve_role_from_user logger = logging.getLogger(__name__) API_VERSION = "2.0.0" APK_MANIFEST_PATH = Path(__file__).resolve().parent / "static" / "apk" / "manifest.json" def _get_apk_release_summary() -> dict: try: manifest = json.loads(APK_MANIFEST_PATH.read_text(encoding="utf-8-sig")) except (OSError, json.JSONDecodeError): return {} return { "apk_version": manifest.get("version"), "apk_build_date": manifest.get("build_date"), "apk_filename": manifest.get("filename"), } """ Application setup: Initializes FastAPI app, configures rate limiting, and sets up exception handlers. """ limiter = Limiter(key_func=get_remote_address) app = FastAPI( title="Catsy Coffee API", description="Secure FastAPI bridge → Supabase backend", version=API_VERSION, ) app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) """ CORS configuration: - Reads exact allowed origins from ALLOWED_ORIGINS (comma-separated). - Reads optional regex from ALLOWED_ORIGIN_REGEX. - Development defaults allow localhost plus private-LAN Vite-style origins. - Production requires explicit ALLOWED_ORIGINS and does not allow private-LAN regex by default. """ APP_ENV = ( os.environ.get("APP_ENV", os.environ.get("ENVIRONMENT", "development")) .strip() .lower() ) IS_PRODUCTION = APP_ENV in {"prod", "production"} _default_allowed_origins = ( "" if IS_PRODUCTION else "http://localhost:5173,http://127.0.0.1:5173" ) _raw = os.environ.get("ALLOWED_ORIGINS", _default_allowed_origins) ALLOWED_ORIGINS = [o.strip() for o in _raw.split(",") if o.strip()] ALLOWED_ORIGIN_REGEX = os.environ.get( "ALLOWED_ORIGIN_REGEX", None if IS_PRODUCTION else r"^https?://(localhost|127\.0\.0\.1|10(?:\.\d{1,3}){3}|192\.168(?:\.\d{1,3}){2}|172\.(?:1[6-9]|2\d|3[0-1])(?:\.\d{1,3}){2})(?::\d+)?$", ) app.add_middleware( CORSMiddleware, allow_origins=ALLOWED_ORIGINS, allow_origin_regex=ALLOWED_ORIGIN_REGEX, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) """ Mount routers: Registers all API routers with the FastAPI application instance. Routers are grouped by domain and phase. """ app.include_router(auth.router) app.include_router(products.router) app.include_router(categories.router) app.include_router(loyalty.router) app.include_router(reservations.router) app.include_router(reservations.customer_router) app.include_router(settings.router) app.include_router(audit_router) app.include_router(inventory_router) app.include_router(users_router) """ Domain routers: Each router module owns its own slice of the API surface. Routers are imported and mounted below. """ app.include_router(apk_router) app.include_router(orders.router) app.include_router(reservations.public_router) app.include_router(reservations.admin_router) # Admin maintenance endpoints for reservations app.include_router(customer.customer_router) # Handles GET /api/customer/orders app.include_router(customer.staff_router) # Handles GET /api/staff/members?search= app.include_router(rewards.admin_router) # Admin CRUD endpoints for reward_items app.include_router(rewards.public_router) # Public endpoint for GET /api/rewards/active """ Phase 3 routers: Additional features (CMS, reports, seats, etc.) """ app.include_router(cms.admin_router) app.include_router(feedback.router) # Customer feedback endpoints app.include_router(feedback.admin_router) # Admin feedback moderation endpoints app.include_router(cms.public_router) app.include_router(reports.router) app.include_router(seats.router) app.include_router(seats.staff_router) app.include_router(notifications.router) """ Materials routers: Registers both the main materials router and the new recipe router. """ app.include_router(materials.router) app.include_router(materials.recipe_router) app.include_router(materials.staff_router) app.include_router(materials.staff_recipe_router) app.include_router(stalls.admin_router) app.include_router(stalls.staff_router) app.include_router(stalls.public_router) app.include_router(inventory_allocation_router) # Admin per-stall inventory allocation STREAM_TICKET_TTL_SECONDS = int(os.environ.get("STREAM_TICKET_TTL_SECONDS", "30")) STREAM_TICKET_SECRET = ( os.environ.get("SSE_STREAM_SECRET") or os.environ.get("SUPABASE_SERVICE_ROLE_KEY") or secrets.token_urlsafe(32) ) RESERVATION_MAINTENANCE_ENABLED = ( os.environ.get("RESERVATION_MAINTENANCE_ENABLED", "true").strip().lower() not in {"0", "false", "no", "off"} ) RESERVATION_MAINTENANCE_INTERVAL_SECONDS = int( os.environ.get("RESERVATION_MAINTENANCE_INTERVAL_SECONDS", "300") ) RESERVATION_MAINTENANCE_INITIAL_DELAY_SECONDS = int( os.environ.get("RESERVATION_MAINTENANCE_INITIAL_DELAY_SECONDS", "60") ) """ Utility endpoints: Health check and other non-domain-specific endpoints. """ @app.get("/", tags=["Health"]) def health_check(): return { "status": "✅ Catsy API is online", "version": API_VERSION, **_get_apk_release_summary(), } @app.head("/", tags=["Health"]) def health_check_head(): return Response(status_code=200) @app.options("/{rest_of_path:path}", tags=["CORS"]) async def preflight_handler(rest_of_path: str, request: Request): """Global OPTIONS catch-all so SlowAPI never intercepts CORS preflight requests.""" return {} @app.on_event("startup") async def startup_realtime_broker(): try: await broker.startup() except Exception: logger.exception( "Realtime broker startup failed; continuing with in-memory events." ) @app.on_event("shutdown") async def shutdown_realtime_broker(): await broker.shutdown() def _run_reservation_maintenance_once() -> dict: return reservations.run_reservation_maintenance( ReservationRepository(), SettingsRepository(), SeatsRepository(), ) async def _reservation_maintenance_loop() -> None: if RESERVATION_MAINTENANCE_INITIAL_DELAY_SECONDS > 0: await asyncio.sleep(RESERVATION_MAINTENANCE_INITIAL_DELAY_SECONDS) while True: try: result = await asyncio.to_thread(_run_reservation_maintenance_once) logger.info( "Reservation maintenance completed: dismissed_count=%s", result.get("dismissed_count"), ) except asyncio.CancelledError: raise except Exception: logger.exception("Reservation maintenance failed.") await asyncio.sleep(RESERVATION_MAINTENANCE_INTERVAL_SECONDS) @app.on_event("startup") async def startup_reservation_maintenance(): if not RESERVATION_MAINTENANCE_ENABLED: logger.info("Reservation maintenance scheduler disabled.") return if RESERVATION_MAINTENANCE_INTERVAL_SECONDS <= 0: logger.warning("Reservation maintenance scheduler disabled: interval must be positive.") return app.state.reservation_maintenance_task = asyncio.create_task( _reservation_maintenance_loop() ) @app.on_event("shutdown") async def shutdown_reservation_maintenance(): task = getattr(app.state, "reservation_maintenance_task", None) if task is None: return task.cancel() with contextlib.suppress(asyncio.CancelledError): await task """ Server-Sent Events (SSE): Provides a real-time event stream for UI updates. """ async def _event_generator( request: Request, *, user_id: str | None = None, role: str | None = None, last_event_id: int | None = None, ): """Heartbeat ping plus filtered broadcast events for connected clients.""" subscriber = None subscriber, backlog = broker.subscribe(user_id=user_id, role=role, last_event_id=last_event_id) try: for message in backlog: yield message while True: if await request.is_disconnected(): break try: message = await asyncio.wait_for(subscriber.queue.get(), timeout=15) yield message except asyncio.TimeoutError: yield ": ping\n\n" except asyncio.CancelledError: pass finally: if subscriber is not None: broker.unsubscribe(subscriber) @app.get("/api/events/stream", tags=["Realtime"]) @limiter.limit("30/minute") async def stream_events(request: Request): stream_ticket = request.query_params.get("stream_ticket") last_event_id_value = request.headers.get("Last-Event-ID") or request.query_params.get("last_event_id") last_event_id = None if last_event_id_value: try: last_event_id = int(last_event_id_value) except ValueError: last_event_id = None user_id = None role = None if stream_ticket: try: payload = jwt.decode(stream_ticket, STREAM_TICKET_SECRET, algorithms=["HS256"]) except jwt.InvalidTokenError as exc: raise HTTPException(status_code=401, detail="Invalid stream ticket") from exc if payload.get("type") != "sse_stream": raise HTTPException(status_code=401, detail="Invalid stream ticket") user_id = payload.get("sub") role = payload.get("role") elif request.query_params.get("access_token"): raise HTTPException(status_code=401, detail="Use a stream ticket for authenticated realtime access") return StreamingResponse( _event_generator(request, user_id=user_id, role=role, last_event_id=last_event_id), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", }, ) @app.post("/api/events/token", tags=["Realtime"]) @limiter.limit("60/minute") def create_stream_ticket(request: Request, user=Depends(get_current_user)): role = resolve_role_from_user(user) now = int(time.time()) token = jwt.encode( { "sub": str(user.id), "role": role, "type": "sse_stream", "iat": now, "exp": now + STREAM_TICKET_TTL_SECONDS, }, STREAM_TICKET_SECRET, algorithm="HS256", ) return { "stream_ticket": token, "expires_in": STREAM_TICKET_TTL_SECONDS, } """ Development entry point: Allows running the app with `python main.py` for local development. """ if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)