app/backend/fluksio/main.py

import asyncio
import contextlib
from collections.abc import AsyncIterator
from contextlib import AbstractAsyncContextManager, asynccontextmanager

import sentry_sdk
from fastapi import FastAPI
from fastapi.concurrency import run_in_threadpool
from fastapi.responses import JSONResponse
from fastapi.routing import APIRoute
from fluksio_worker.worker_main import ARTIFACT_DIR_ENV
from starlette.middleware.cors import CORSMiddleware

from fluksio.api.main import api_router
from fluksio.api.routes.alerts import read_config as read_alerts_config
from fluksio.cloud import config as cloud_config
from fluksio.core import security
from fluksio.core.config import settings
from fluksio.core.db import engine as db_engine
from fluksio.core.db import prepare
from fluksio.flow import logs, modules
from fluksio.flow.alerts import AlertManager
from fluksio.flow.artifacts import ArtifactStore
from fluksio.flow.controller import FlowController
from fluksio.flow.dashboards import DashboardStore
from fluksio.flow.events import event_bus
from fluksio.flow.executor import ExecutionService
from fluksio.flow.metrics import MetricsCollector
from fluksio.flow.nodes.http import close_shared_client
from fluksio.flow.pipeline import ValueSource
from fluksio.flow.plugins import load_plugins
from fluksio.flow.queue import MemoryWorkQueue, RedisWorkQueue, WorkQueue
from fluksio.flow.remote import RemoteWorkerHub
from fluksio.flow.runs import RUN_STATE_TTL, RunService
from fluksio.flow.secrets import init_secrets
from fluksio.flow.state import MemoryState, RedisState, StateBackend
from fluksio.flow.store import FlowStore
from fluksio.flow.watchdog import LoopWatchdog
from fluksio.flow.workers import PythonWorkerPool


def custom_generate_unique_id(route: APIRoute) -> str:
    return f"{route.tags[0]}-{route.name}"


if settings.SENTRY_DSN and settings.ENVIRONMENT != "local":
    # `enable_tracing` was removed in sentry-sdk 2.x; this is what it meant.
    sentry_sdk.init(dsn=str(settings.SENTRY_DSN), traces_sample_rate=1.0)


def _state_backend() -> StateBackend:
    if settings.REDIS_HOST:
        return RedisState(host=settings.REDIS_HOST, port=settings.REDIS_PORT)
    return MemoryState()


def _work_queue(namespace: str = "queue") -> WorkQueue:
    """Redis makes queued work survive the process; memory does not pretend to."""
    if settings.REDIS_HOST:
        return RedisWorkQueue(
            host=settings.REDIS_HOST, port=settings.REDIS_PORT, namespace=namespace
        )
    return MemoryWorkQueue()


def _run_state(namespace: str) -> StateBackend:
    """A state backend of a run's own, which is what isolates it.

    It expires: a finished run's messages are read out into its result, and
    what is left is only worth keeping while someone might look at it.
    """
    if settings.REDIS_HOST:
        return RedisState(
            host=settings.REDIS_HOST,
            port=settings.REDIS_PORT,
            namespace=namespace,
            ttl=RUN_STATE_TTL,
        )
    return MemoryState()


def _mcp_sessions() -> AbstractAsyncContextManager[None]:
    """The MCP session manager's run scope, or nothing when MCP is off."""
    if not settings.MCP_ENABLED:
        return contextlib.nullcontext()
    from fluksio.mcp.server import mcp as mcp_server

    return mcp_server.session_manager.run()


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    """Start the flow engine alongside the API."""
    # The schema and the first superuser, before anything reads either. It is
    # idempotent, so a deployment that ran this from its own prestart step
    # pays a version check for it and nothing else.
    await run_in_threadpool(prepare, db_engine)
    event_bus.bind(asyncio.get_running_loop())
    # Node code is user code, and `print` is how it says things.
    logs.install()
    init_secrets(settings.SECRETS_FILE, settings.SECRET_KEY)
    # Connectors register their node types before any flow is built with them.
    load_plugins()

    alerts = AlertManager(event_bus, config=read_alerts_config())
    execution = ExecutionService(
        queue=_work_queue(),
        max_workers=settings.FLOW_MAX_WORKERS,
        events=event_bus,
    )
    store = FlowStore(settings.FLOWS_DIR)
    # The packages node code imports, before anything tries to import them.
    await run_in_threadpool(modules.reconcile, store)
    # Beside the flows rather than in them: an artifact is what a run produced,
    # not something anyone wrote, so it has no business in the git repository.
    artifacts = ArtifactStore(settings.FLOWS_DIR.parent / "artifacts")
    app.state.artifact_store = artifacts
    pool = PythonWorkerPool(
        python=modules.venv_python(),
        size=settings.FLOW_MAX_WORKERS,
        events=event_bus,
        # A worker in this container writes to the store directly; a remote one
        # is given a URL instead. Node code calls the same two functions.
        env={ARTIFACT_DIR_ENV: str(artifacts.root)},
    )
    pool.start()
    app.state.worker_pool = pool
    worker_hub = RemoteWorkerHub()
    app.state.worker_hub = worker_hub
    controller = FlowController(
        store=store,
        state=_state_backend(),
        events=event_bus,
        max_workers=settings.FLOW_MAX_WORKERS,
        fastapi_app=app,
        execution=execution,
        alerts=alerts,
        workers=pool,
        remote=worker_hub,
    )
    app.state.flow_controller = controller
    # A "dashboard" alert channel puts its alert into the graph. Bound here
    # rather than passed in: the manager is built before the controller is.
    alerts.publish = lambda name, value: controller.publish_message(
        name, value, ValueSource(kind="api", id="alerts", label="Alerts")
    )
    dashboards = DashboardStore(controller.store)
    app.state.dashboard_store = dashboards
    controller.dashboards = dashboards
    # Charts need a deeper series than the default; tell the engine
    # before it starts recording.
    controller.set_history_limits(dashboards.history_requirements())
    # Runs read from a stream of their own: a burst of sweep runs must not
    # stand between the automations and their work, and a run that takes an
    # hour must not be judged by the cascade reaper's timings.
    run_service = RunService(
        controller=controller,
        queue=_work_queue("run"),
        state_factory=_run_state,
    )
    app.state.run_service = run_service
    watchdog = LoopWatchdog(event_bus)
    app.state.watchdog = watchdog
    watchdog_task = asyncio.create_task(watchdog.run(), name="loop-watchdog")
    alerts_task = asyncio.create_task(alerts.run(), name="alert-manager")
    metrics_task = asyncio.create_task(
        MetricsCollector(event_bus).run(), name="metrics-collector"
    )
    await controller.start()
    run_service.start()
    # Optional, and off unless someone enrolled this installation: the
    # connector dials the portal, nothing dials in.
    cloud_task: asyncio.Task[None] | None = None
    app.state.cloud_connector = None
    app.state.cloud_task = None
    if cloud_config.exists():
        from fluksio.cloud.connector import CloudConnector

        connector = CloudConnector(app)
        app.state.cloud_connector = connector
        cloud_task = asyncio.create_task(
            connector.serve_forever(), name="cloud-connector"
        )
        app.state.cloud_task = cloud_task
    try:
        # A mounted sub-app gets no lifespan of its own, so the MCP session
        # manager is entered here; without it every /mcp request fails.
        async with _mcp_sessions():
            yield
    finally:
        watchdog_task.cancel()
        alerts_task.cancel()
        metrics_task.cancel()
        # Re-read from app.state: enrolling at runtime replaces this.
        running_cloud = getattr(app.state, "cloud_task", None) or cloud_task
        if running_cloud is not None:
            running_cloud.cancel()
        await run_in_threadpool(run_service.stop)
        await controller.stop()
        pool.stop()
        close_shared_client()
        if settings.MCP_ENABLED:
            from fluksio.mcp.http import aclose

            await aclose()


# The schema enumerates every endpoint this installation serves, including the
# paths trigger nodes mount at runtime. That is exactly what a developer wants
# and exactly what an internet-facing deployment should not hand out, so it
# follows the environment — the same rule the portal's backend uses. The
# generated client is built from a local run, not from the deployed host.
_docs_enabled = settings.ENVIRONMENT != "production"

app = FastAPI(
    title=settings.PROJECT_NAME,
    openapi_url=f"{settings.API_V1_STR}/openapi.json" if _docs_enabled else None,
    docs_url="/docs" if _docs_enabled else None,
    redoc_url="/redoc" if _docs_enabled else None,
    generate_unique_id_function=custom_generate_unique_id,
    lifespan=lifespan,
)

# Set all CORS enabled origins
if settings.all_cors_origins:
    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.all_cors_origins,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

app.include_router(api_router, prefix=settings.API_V1_STR)


# Tagged because the operation-id builder reads the first tag; the route
# itself stays out of the schema.
@app.get(
    "/.well-known/oauth-authorization-server",
    include_in_schema=False,
    tags=["oauth"],
)
def oauth_authorization_server() -> JSONResponse:
    """RFC 8414 metadata, so an agent can find its way in unaided.

    The authorization endpoint is the dashboard rather than the API: approving
    a client needs a signed-in human, and the browser session lives there.
    """
    issuer = settings.oauth_issuer
    return JSONResponse(
        content={
            "issuer": issuer,
            "authorization_endpoint": (
                f"{settings.FRONTEND_HOST.rstrip('/')}/oauth/authorize"
            ),
            "token_endpoint": f"{issuer}{settings.API_V1_STR}/oauth/token",
            "registration_endpoint": f"{issuer}{settings.API_V1_STR}/oauth/register",
            "jwks_uri": f"{issuer}/.well-known/jwks.json",
            "response_types_supported": ["code"],
            "grant_types_supported": ["authorization_code", "refresh_token"],
            "code_challenge_methods_supported": ["S256"],
            "token_endpoint_auth_methods_supported": ["none"],
            "scopes_supported": [security.MCP_SCOPE],
        },
        headers={"Cache-Control": "public, max-age=3600"},
    )


@app.get("/.well-known/jwks.json", include_in_schema=False, tags=["oauth"])
def jwks() -> JSONResponse:
    """The public half of the MCP signing key."""
    return JSONResponse(
        content=security.public_jwks(),
        headers={"Cache-Control": "public, max-age=3600"},
    )


# Mounted last, and at the root: the SDK serves both /mcp and the protected
# resource metadata that has to sit beside it, so mounting under /mcp would put
# that metadata somewhere no client looks for it.
if settings.MCP_ENABLED:
    from fluksio.mcp.http import build_http_app

    app.mount("/", build_http_app(app))