main.py

from __future__ import annotations

import time
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from uuid import uuid4

from fastapi import FastAPI, HTTPException, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse, ORJSONResponse
from starlette.middleware.base import RequestResponseEndpoint
from starlette.middleware.cors import CORSMiddleware
from starlette.responses import Response

from app.ai import api as ai
from app.analytics import api as analytics
from app.brand import api as brand
from app.api import (
    api_keys,
    audio,
    generation,
    health,
    image,
    media,
    probe,
    social,
    templates,
    video,
    whisper,
    ytdlp,
)
from app.container import build_container
from app.copilot import api as copilot
from app.core.config import Settings, get_settings
from app.core.exceptions import MediaAPIError
from app.core.logger import configure_logging, get_logger, request_id_context
from app.core.response import ErrorBody, ErrorResponse
from app.mcp.server import create_mcp_server
from app.projects import api as projects
from app.security.middleware import APIKeyAuthenticationMiddleware
from app.social.workers.scheduler import SocialSchedulerWorker
from app.templates import marketplace_api
from app.workers.cleanup_worker import CleanupWorker

configure_logging()
logger = get_logger(__name__)
try:
    __import__("orjson")

    DefaultJSONResponse = ORJSONResponse
except ImportError:  # pragma: no cover - production requirements always install orjson
    DefaultJSONResponse = JSONResponse
try:
    import psutil as _psutil
except ImportError:  # pragma: no cover - production requirements always install psutil
    _psutil = None  # type: ignore[assignment]


def create_app(settings: Settings | None = None) -> FastAPI:
    active_settings = settings or get_settings()
    active_settings.ensure_directories()
    container = build_container(active_settings)
    cleanup_worker = CleanupWorker(container.cleanup, active_settings.cleanup_interval_seconds)
    social_worker = SocialSchedulerWorker(
        container.social, active_settings.social_scheduler_interval_seconds
    )
    mcp_server = create_mcp_server(container)
    mcp_http_app = mcp_server.streamable_http_app()

    @asynccontextmanager
    async def lifespan(application: FastAPI) -> AsyncIterator[None]:
        application.state.container = container
        application.state.mcp_server = mcp_server
        await container.security_database.initialize()
        if not await container.security_database.schema_ready():
            missing = ", ".join(await container.security_database.missing_schema_objects())
            raise RuntimeError(
                "Security schema is unavailable; apply app/security/migrations/ "
                "and app/projects/migrations/. "
                f"Missing: {missing}"
            )
        await container.security_database.verify_execution_boundary(
            expected_role=active_settings.security_database_role,
            enforce_rls=active_settings.security_enforce_rls,
        )
        await container.generation.initialize()
        await container.api_keys.ensure_bootstrap_admin()
        await container.tenants.ensure_all_api_key_principals()
        await container.social.initialize()
        await container.analytics.initialize(container.social.ready)
        await container.social.adopt_legacy_workspaces(await container.tenants.list_principals())
        async with mcp_server.session_manager.run():
            await cleanup_worker.start()
            await container.generation_worker.start()
            await container.render_worker.start()
            if active_settings.social_enabled and active_settings.social_worker_enabled:
                await social_worker.start()
                await container.analytics_worker.start()
            logger.info(
                "media API started",
                extra={"version": active_settings.app_version, "port": active_settings.port},
            )
            try:
                yield
            finally:
                await social_worker.stop()
                await container.analytics_worker.stop()
                await container.generation_worker.stop()
                await container.render_worker.stop()
                await cleanup_worker.stop()
                await container.generation.close()
                await container.social.close()
                await container.security_database.close()
                logger.info("media API stopped")

    application = FastAPI(
        title=active_settings.app_name,
        version=active_settings.app_version,
        description=(
            "CPU-optimized FFmpeg, yt-dlp, FFprobe, faster-whisper, MCP, and YAML template API."
        ),
        default_response_class=DefaultJSONResponse,
        lifespan=lifespan,
        docs_url="/docs",
        redoc_url="/redoc",
        openapi_url="/openapi.json",
    )
    # State is also assigned eagerly so ASGI test clients without lifespan support work.
    application.state.container = container
    application.state.mcp_server = mcp_server

    # Added before request_context so the request-ID/logging middleware remains outermost.
    application.add_middleware(
        APIKeyAuthenticationMiddleware,
        settings=active_settings,
        api_keys=container.api_keys,
        rate_limiter=container.rate_limiter,
        audit=container.audit,
    )
    if active_settings.allowed_cors_origins:
        application.add_middleware(
            CORSMiddleware,
            allow_origins=list(active_settings.allowed_cors_origins),
            allow_credentials=True,
            allow_methods=["GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
            allow_headers=[
                "Authorization",
                "Content-Type",
                "Idempotency-Key",
                "X-Request-ID",
                "X-MediaRouter-Human-Role",
            ],
            expose_headers=["Content-Disposition", "X-Request-ID"],
        )

    @application.middleware("http")
    async def request_context(request: Request, call_next: RequestResponseEndpoint) -> Response:
        request_id = str(uuid4())
        request.state.request_id = request_id
        request.state.operation = f"{request.method} {request.url.path}"
        token = request_id_context.set(request_id)
        started = time.monotonic()
        process = _psutil.Process() if _psutil else None
        status_code = 500
        try:
            response = await call_next(request)
            status_code = response.status_code
            response.headers["X-Request-ID"] = request_id
            return response
        finally:
            elapsed = round(time.monotonic() - started, 4)
            logger.info(
                "request completed",
                extra={
                    "operation": request.state.operation,
                    "method": request.method,
                    "path": request.url.path,
                    "status_code": status_code,
                    "duration": elapsed,
                    "cpu_percent": _psutil.cpu_percent(interval=None) if _psutil else None,
                    "memory_bytes": process.memory_info().rss if process else None,
                },
            )
            await container.cleanup.complete(request_id)
            request_id_context.reset(token)

    @application.exception_handler(MediaAPIError)
    async def media_error_handler(request: Request, exc: MediaAPIError) -> JSONResponse:
        logger.warning(
            "request failed",
            extra={"error_code": exc.code, "operation": request.state.operation},
        )
        body = ErrorResponse(
            request_id=request.state.request_id,
            error=ErrorBody(code=exc.code, message=exc.message, details=exc.details),
        )
        return DefaultJSONResponse(body.model_dump(), status_code=exc.status_code)

    @application.exception_handler(RequestValidationError)
    async def validation_error_handler(
        request: Request, exc: RequestValidationError
    ) -> JSONResponse:
        body = ErrorResponse(
            request_id=request.state.request_id,
            error=ErrorBody(
                code="VALIDATION_ERROR",
                message="Request validation failed",
                details=exc.errors(include_url=False, include_context=False),
            ),
        )
        return DefaultJSONResponse(body.model_dump(), status_code=422)

    @application.exception_handler(HTTPException)
    async def http_error_handler(request: Request, exc: HTTPException) -> JSONResponse:
        body = ErrorResponse(
            request_id=request.state.request_id,
            error=ErrorBody(code="HTTP_ERROR", message=str(exc.detail)),
        )
        return DefaultJSONResponse(body.model_dump(), status_code=exc.status_code)

    @application.exception_handler(Exception)
    async def unexpected_error_handler(request: Request, exc: Exception) -> JSONResponse:
        logger.exception("unhandled request error", extra={"operation": request.state.operation})
        body = ErrorResponse(
            request_id=request.state.request_id,
            error=ErrorBody(
                code="INTERNAL_ERROR",
                message="An unexpected internal error occurred",
            ),
        )
        return DefaultJSONResponse(body.model_dump(), status_code=500)

    application.include_router(health.router)
    application.include_router(api_keys.router)
    application.include_router(media.router)
    application.include_router(video.router)
    application.include_router(audio.router)
    application.include_router(image.router)
    application.include_router(probe.router)
    application.include_router(ytdlp.router)
    application.include_router(whisper.router)
    application.include_router(marketplace_api.router)
    # Register the static marketplace prefix before the legacy dynamic
    # /v1/templates/{template_id} route so "catalog" cannot be captured as a
    # YAML workflow template ID.
    application.include_router(templates.router)
    application.include_router(ai.router)
    application.include_router(copilot.router)
    application.include_router(generation.router)
    application.include_router(projects.router)
    application.include_router(social.router)
    application.include_router(analytics.router)
    application.include_router(brand.router)
    application.mount("/mcp", mcp_http_app, name="mcp")

    @application.get("/", tags=["public"], include_in_schema=False)
    async def root() -> dict[str, str]:
        return {
            "name": active_settings.app_name,
            "version": active_settings.app_version,
            "status": "healthy",
            "docs": "/docs",
        }

    @application.get("/version", tags=["public"], include_in_schema=False)
    async def version() -> dict[str, str]:
        return {"version": active_settings.app_version}

    original_openapi = application.openapi

    def secure_openapi() -> dict[str, object]:
        if application.openapi_schema:
            return application.openapi_schema  # type: ignore[return-value]
        schema = original_openapi()
        components = schema.setdefault("components", {})
        components.setdefault("securitySchemes", {})["APIKeyBearer"] = {
            "type": "http",
            "scheme": "bearer",
            "bearerFormat": "mp_live_…",
            "description": "MediaRouter API key",
        }
        schema["security"] = [{"APIKeyBearer": []}]
        # The provider callback cannot send an API key. It is authenticated by
        # the random, expiring, single-use OAuth state stored server-side.
        for path in ("/health", "/v1/social/accounts/{provider}/callback"):
            for operation in schema.get("paths", {}).get(path, {}).values():
                if isinstance(operation, dict):
                    operation["security"] = []
        application.openapi_schema = schema
        return schema

    application.openapi = secure_openapi  # type: ignore[method-assign]
    return application


app = create_app()