from __future__ import annotations import asyncio from contextlib import asynccontextmanager from fastapi import Depends, FastAPI, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.gzip import GZipMiddleware from fastapi.responses import JSONResponse from slowapi import Limiter from slowapi.errors import RateLimitExceeded from slowapi.util import get_remote_address from app.api.v1.router import api_v1_router from app.api.v1.system import is_maintenance from app.config import get_settings from app.core.auth.deps import init_auth_db from app.core.database import pool_manager from app.core.logger import get_logger from app.core.redis_client import close_redis, create_redis_client from app.core.scripts import load_scripts from app.services.embeddings_service import EmbeddingService from app.services.scheduler_service import SchedulerService from app.services.vector_store_service import VectorStoreService from app.utils.http_utils import SharedAsyncClient _logger = get_logger(__name__) _settings = get_settings() # Public /api/v1/ paths that do not require authentication _PUBLIC_API_PREFIXES = ( "/api/v1/auth/register", "/api/v1/auth/login", "/api/v1/auth/refresh", "/api/v1/auth/forgot-password", "/api/v1/auth/reset-password", "/api/v1/url-shortener/", ) # GET-only paths that must not require authentication: Google redirects the # user's browser here after authorization, and a redirect cannot attach the API key. _PUBLIC_GET_PATHS = frozenset({"/api/v1/google/oauth/callback"}) def _is_public_path(path: str, method: str = "GET") -> bool: if path in _PUBLIC_GET_PATHS and method == "GET": return True return path.startswith(_PUBLIC_API_PREFIXES) _embedding_service: EmbeddingService = EmbeddingService() _vector_store_service: VectorStoreService = VectorStoreService(_embedding_service) _scheduler_service: SchedulerService = SchedulerService() _ping_client = SharedAsyncClient(timeout=30.0) # Global per-client-IP rate limit applied to every /api/v1 route. # Configure via RATE_LIMIT_PER_MINUTE (0 or empty disables). limiter = Limiter(key_func=get_remote_address, storage_uri="memory://", headers_enabled=False) async def _api_rate_limit(request: Request) -> None: """Apply the global per-client-IP rate limit to all /api/v1 routes.""" return None _RATE_LIMIT_PER_MINUTE = _settings.rate_limit_per_minute if _RATE_LIMIT_PER_MINUTE and _RATE_LIMIT_PER_MINUTE > 0: _api_rate_limit = limiter.limit(f"{_RATE_LIMIT_PER_MINUTE}/minute")(_api_rate_limit) async def _self_ping(): health_url = _settings.self_ping_url while True: try: client = await _ping_client.get() response = await client.get(health_url) if response.status_code == 200: _logger.info("Self-ping successful: %s", health_url) else: _logger.warning("Self-ping returned: %s - %s", health_url, response.status_code) except Exception as exc: _logger.error("Self-ping error: %s", exc) await asyncio.sleep(900) @asynccontextmanager async def lifespan(app: FastAPI): if not _settings.supabase_url or not _settings.supabase_service_role_key: _logger.error( "Supabase not configured! Set SUPABASE_URL and " "SUPABASE_SERVICE_ROLE_KEY environment variables." ) else: _logger.info("Initializing Supabase databases...") await init_auth_db() _logger.info("Authentication database initialized via Supabase") _logger.info("Initializing vector store database...") from app.core.vector_store.deps import init_vector_store_db await init_vector_store_db() await _vector_store_service.init_db() _logger.info( "Vector store database initialized with %d stores", len(_vector_store_service.list_stores()), ) _logger.info("Initializing embedding service (loading 384-dim model)...") loop = asyncio.get_running_loop() await loop.run_in_executor(None, _embedding_service.load_model, 384) _logger.info("Embedding service initialized with dims: %s", _embedding_service.loaded_dimensions) _logger.info("Vector store service initialized with %d existing stores", len(_vector_store_service.list_stores())) redis = create_redis_client(_settings.redis_url) if _settings.redis_url else None scripts = await load_scripts(redis) if redis else {} app.state.redis = redis app.state.scripts = scripts if redis: _logger.info("Redis and Lua scripts initialized") else: _logger.warning("Redis not configured, running in degraded mode") asyncio.create_task(_self_ping()) if _settings.supabase_upload_enabled: try: from app.services.media_storage_service import get_storage_service storage = await get_storage_service() bucket = await storage.ensure_bucket(_settings.supabase_storage_bucket) _logger.info("Supabase Storage bucket ensured: %s", bucket) except Exception as exc: _logger.error("Failed to ensure Supabase Storage bucket at startup: %s", exc) await _scheduler_service.start() _logger.info("Scheduler service started") yield _logger.info("Shutting down...") await _scheduler_service.shutdown() await close_redis(redis) await _vector_store_service.close_all() await pool_manager.close_all() from app.api.v1.google_oauth import close_oauth_service await close_oauth_service() from app.api.v1.google_maps import close_maps_service await close_maps_service() from app.api.v1.gcs import close_gcs_service await close_gcs_service() from app.api.v1.gmail import close_gmail_service await close_gmail_service() from app.api.v1.sheets import close_sheets_service await close_sheets_service() from app.services.media_storage_service import close_storage_service await close_storage_service() from app.utils.http_utils import close_shared_aiohttp_sessions await close_shared_aiohttp_sessions() await _ping_client.close() from app.services.supabase import get_supabase_client client = get_supabase_client() if client: await client.close() _logger.info("Supabase client closed") def create_application() -> FastAPI: app = FastAPI( title=_settings.app_name, description="AgentDeck-Backend", version=_settings.app_version, docs_url="/docs", redoc_url="/redoc", openapi_tags=[ {"name": "Convert", "description": "Single-file and single-URL conversion"}, {"name": "Batch", "description": "Bulk conversion of files and URLs"}, {"name": "System", "description": "Health, info, and supported formats"}, {"name": "Embeddings", "description": "Text embedding generation using transformer models"}, {"name": "Verify", "description": "Phone number and identity verification"}, {"name": "Vector Stores", "description": "Create, manage, and search vector stores for RAG"}, {"name": "URL Shortener", "description": "Create and manage short URLs with analytics"}, {"name": "Media-to-Media Conversion", "description": "PDF-to-image and image-to-image conversion with local or Supabase Storage output"}, ], lifespan=lifespan, ) app.add_middleware(GZipMiddleware, minimum_size=1000) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) app.state.limiter = limiter @app.exception_handler(RateLimitExceeded) async def _rate_limit_exceeded_handler(request: Request, exc: RateLimitExceeded): retry_after = getattr(exc, "retry_after", 60) or 60 return JSONResponse( status_code=429, content={"success": False, "detail": "Rate limit exceeded. Please retry later."}, headers={"Retry-After": str(int(retry_after))}, ) @app.middleware("http") async def maintenance_middleware(request: Request, call_next): if is_maintenance(): method = request.method path = request.url.path if method not in ("GET", "HEAD", "OPTIONS"): if not path.startswith("/api/v1/maintenance"): if method == "POST" and path.startswith("/api/v1/backup"): pass else: from starlette.responses import JSONResponse return JSONResponse( status_code=503, content={ "success": False, "detail": "Service is under maintenance. No write operations allowed.", }, ) return await call_next(request) @app.middleware("http") async def auth_middleware(request: Request, call_next): path = request.url.path if path.startswith("/api/v1/") and not _is_public_path(path, request.method): auth_header = request.headers.get("Authorization", "") if not auth_header.startswith("Bearer "): from starlette.responses import JSONResponse return JSONResponse( status_code=401, content={"success": False, "detail": "Missing Authorization header"}, ) token = auth_header.removeprefix("Bearer ") if token != _settings.api_key: from starlette.responses import JSONResponse return JSONResponse( status_code=401, content={"success": False, "detail": "Invalid API key"}, ) return await call_next(request) app.include_router( api_v1_router, prefix="/api/v1", dependencies=[Depends(_api_rate_limit)], ) @app.get("/", include_in_schema=False) async def root(request: Request): from collections import defaultdict routes_by_tag: dict[str, list[dict]] = defaultdict(list) for route in app.routes: if not hasattr(route, "methods") or not hasattr(route, "path"): continue if route.path in ("/", "/health", "/ping", "/openapi.json", "/docs", "/redoc", "/docs/oauth2-redirect"): continue tags = getattr(route, "tags", None) or ["default"] for tag in tags: routes_by_tag[tag].append({ "method": list(route.methods - {"HEAD", "OPTIONS"}), "path": route.path, "summary": getattr(route, "summary", ""), }) return { "name": _settings.app_name, "version": _settings.app_version, "docs": { "swagger": str(request.base_url) + "docs", "redoc": str(request.base_url) + "redoc", }, } @app.get("/health", include_in_schema=False) async def root_health(): store_count = len(_vector_store_service.list_stores()) doc_count = await _vector_store_service.get_total_document_count() return { "success": True, "app_name": _settings.app_name, "version": _settings.app_version, "embedding_dimension": _settings.embedding_dimension, "vector_store_count": store_count, "total_documents": doc_count, "model_loaded": _embedding_service.is_loaded(384), } @app.get("/ping", include_in_schema=False) async def ping(): return {"name": f"{_settings.app_name}", "version": _settings.app_version} return app app = create_application()