app/api/server.py

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()