kirche-einfach/backend/app/flows/api.py

"""Flows + approvals HTTP API.

Auth model: flow *management* (create/version/deploy/sync/import) is
admin-only. Browsing (list/get/input-schema) is public so the website
renders. Runs, run reads and approval reads require login and are scoped
to the session tenant. Flow *callbacks* (LFX has no session) carry a
shared X-Callback-Token instead.
"""

from __future__ import annotations

import json
import logging
import secrets
from collections.abc import AsyncIterator, Iterator
from pathlib import Path
from typing import Annotated
from uuid import UUID

import httpx
from fastapi import APIRouter, Depends, Header, HTTPException, Query
from fastapi.responses import StreamingResponse
from sqlmodel import Session

from app.auth.deps import get_current_user, require_admin
from app.db import get_engine, get_session
from app.models import User
from app.settings import settings

from . import input_schema as form_schema
from . import runner, store
from .deploy import import_from_langflow, pull_all_from_langflow
from .langflow_client import (
    LangflowClient,
    LangflowError,
    LangflowNotFound,
    LangflowUnavailable,
)
from .lfx_client import (
    LfxClient,
    LfxConflict,
    LfxError,
    LfxInvalidFlow,
    LfxNotFound,
    LfxUnavailable,
)
from .schemas import (
    ApprovalCreate,
    ApprovalRead,
    DecisionRead,
    DecisionRequest,
    FlowDefinitionCreate,
    FlowDefinitionRead,
    FlowImportRead,
    FlowImportRequest,
    FlowVersionCreate,
    FlowVersionRead,
    InputSchemaRead,
    RunRead,
    RunRequest,
    SyncReport,
)
from .sync import lfx_id_for, sync_from_directory

flows_router = APIRouter(prefix="/api/flows", tags=["flows"])
approvals_router = APIRouter(prefix="/api/approvals", tags=["approvals"])

logger = logging.getLogger(__name__)


def get_db() -> Iterator[Session]:
    yield from get_session()


def get_lfx() -> LfxClient:
    return LfxClient(settings.lfx_url, api_key=settings.lfx_api_key or None)


def get_langflow() -> LangflowClient:
    return LangflowClient(settings.langflow_url, api_key=settings.langflow_api_key or "")


def _read_definition(d) -> FlowDefinitionRead:
    return FlowDefinitionRead(
        id=d.id,
        name=d.name,
        description=d.description,
        status=d.status,
        source_path=d.source_path,
        continuation_flow_id=d.continuation_flow_id,
        created_at=d.created_at,
    )


def _read_version(v) -> FlowVersionRead:
    return FlowVersionRead(
        id=v.id,
        flow_id=v.flow_id,
        version_no=v.version_no,
        langflow_json=v.langflow_json,
        lfx_flow_id=v.lfx_flow_id,
        deployed_at=v.deployed_at,
        author=v.author,
        created_at=v.created_at,
        input_schema=v.input_schema,
    )


def _read_run(r) -> RunRead:
    return RunRead(
        id=r.id,
        version_id=r.version_id,
        tenant_id=r.tenant_id,
        status=r.status,
        input=r.input,
        output=r.output,
        session_id=r.session_id,
        created_at=r.created_at,
        finished_at=r.finished_at,
    )


def _read_task(t) -> ApprovalRead:
    return ApprovalRead(
        id=t.id,
        run_id=t.run_id,
        version_id=t.version_id,
        tenant_id=t.tenant_id,
        title=t.title,
        payload=t.payload,
        continuation_flow_id=t.continuation_flow_id,
        approver=t.approver,
        status=t.status,
        decision_note=t.decision_note,
        created_at=t.created_at,
        decided_at=t.decided_at,
    )


@flows_router.post("", response_model=FlowDefinitionRead, status_code=201)
def create_flow(
    data: FlowDefinitionCreate,
    session: Annotated[Session, Depends(get_db)],
    _admin: Annotated[User, Depends(require_admin)],
):
    try:
        return _read_definition(store.create_definition(session, data))
    except store.ConflictError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc
    except store.InvalidError as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc


@flows_router.get("", response_model=list[FlowDefinitionRead])
def list_flows(session: Annotated[Session, Depends(get_db)]):
    return [_read_definition(d) for d in store.list_definitions(session)]


@flows_router.post("/sync", response_model=SyncReport)
async def sync_flows(
    session: Annotated[Session, Depends(get_db)],
    lfx: Annotated[LfxClient, Depends(get_lfx)],
    langflow: Annotated[LangflowClient, Depends(get_langflow)],
    _admin: Annotated[User, Depends(require_admin)],
):
    """Manual full pass: Langflow -> backend/flows/*.json -> DB + LFX."""
    try:
        pulled_result = await pull_all_from_langflow(langflow, Path(settings.flows_dir))
    except LangflowUnavailable as exc:
        logger.warning("langflow unavailable: %s", exc)
        raise HTTPException(status_code=503, detail="authoring server unavailable") from exc
    except LangflowError as exc:
        logger.warning("langflow error: %s", exc)
        raise HTTPException(status_code=502, detail="authoring server error") from exc
    try:
        report = await sync_from_directory(session, lfx, Path(settings.flows_dir))
    except FileNotFoundError as exc:
        raise HTTPException(status_code=500, detail=str(exc)) from exc
    report.pulled = pulled_result["pulled"]
    report.errors = {**pulled_result["errors"], **report.errors}
    return report


@flows_router.get("/{flow_id}", response_model=FlowDefinitionRead)
def get_flow(flow_id: UUID, session: Annotated[Session, Depends(get_db)]):
    try:
        return _read_definition(store.get_definition(session, flow_id))
    except store.NotFoundError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc


@flows_router.post("/{flow_id}/versions", response_model=FlowVersionRead, status_code=201)
def create_version(
    flow_id: UUID,
    data: FlowVersionCreate,
    session: Annotated[Session, Depends(get_db)],
    _admin: Annotated[User, Depends(require_admin)],
):
    try:
        return _read_version(store.create_version(session, flow_id, data))
    except store.NotFoundError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except store.InvalidError as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc


@flows_router.get("/{flow_id}/versions", response_model=list[FlowVersionRead])
def list_versions(
    flow_id: UUID,
    session: Annotated[Session, Depends(get_db)],
    _admin: Annotated[User, Depends(require_admin)],
):
    try:
        return [_read_version(v) for v in store.list_versions(session, flow_id)]
    except store.NotFoundError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc


@flows_router.post("/{flow_id}/versions/{version_no}/deploy", response_model=FlowVersionRead)
async def deploy_version(
    flow_id: UUID,
    version_no: int,
    session: Annotated[Session, Depends(get_db)],
    lfx: Annotated[LfxClient, Depends(get_lfx)],
    _admin: Annotated[User, Depends(require_admin)],
):
    try:
        version = store.get_version_by_no(session, flow_id, version_no)
    except store.NotFoundError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    try:
        # Deterministic LFX id: same (definition, version) always maps to the
        # same registration, no matter which path deployed it.
        uploaded = await lfx.upload_flow(
            version.langflow_json,
            flow_id=lfx_id_for(flow_id, version_no),
            replace=True,
        )
    except LfxInvalidFlow as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc
    except LfxConflict as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc
    except LfxUnavailable as exc:
        logger.warning("lfx unavailable: %s", exc)
        raise HTTPException(status_code=503, detail="flow runtime unavailable") from exc
    except LfxError as exc:
        logger.warning("lfx error: %s", exc)
        raise HTTPException(status_code=502, detail="flow runtime error") from exc
    return _read_version(store.mark_deployed(session, version, uploaded.id))


def _resolve_version(session: Session, flow_id: UUID, version_no: int | None):
    if version_no is not None:
        return store.get_version_by_no(session, flow_id, version_no)
    return store.get_latest_deployed_version(session, flow_id)


@flows_router.post("/{flow_id}/run", response_model=RunRead)
async def run_flow(
    flow_id: UUID,
    data: RunRequest,
    session: Annotated[Session, Depends(get_db)],
    lfx: Annotated[LfxClient, Depends(get_lfx)],
    user: Annotated[User, Depends(get_current_user)],
    stream: bool = Query(default=False),
):
    # Tenant, callback URL and continuation always come from the server:
    # the session, the settings, and the admin-authored definition.
    tenant_id = user.tenant_id
    try:
        definition = store.get_definition(session, flow_id)
        version = _resolve_version(session, flow_id, data.version_no)
    except store.NotFoundError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except store.InvalidError as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc
    if stream:
        return await _run_stream(session, lfx, version, data, user, definition)
    try:
        run = await runner.execute_run(
            session,
            lfx,
            version,
            payload=data.input,
            tenant_id=tenant_id,
            session_id=data.session_id,
            callback_base_url=settings.callback_base_url,
            continuation_flow_id=definition.continuation_flow_id,
            callback_token=settings.callback_token,
        )
    except LfxNotFound as exc:
        raise HTTPException(status_code=502, detail=f"LFX drift: {exc}") from exc
    except LfxUnavailable as exc:
        logger.warning("lfx unavailable: %s", exc)
        raise HTTPException(status_code=503, detail="flow runtime unavailable") from exc
    except LfxError as exc:
        logger.warning("lfx error: %s", exc)
        raise HTTPException(status_code=502, detail="flow runtime error") from exc
    return _read_run(run)


async def _run_stream(
    session: Session, lfx: LfxClient, version, data: RunRequest, user: User, definition
):
    if not version.lfx_flow_id:
        raise HTTPException(status_code=422, detail=f"version {version.id} is not deployed to LFX")
    run = store.create_run(
        session,
        version,
        input=data.input,
        tenant_id=user.tenant_id,
        session_id=data.session_id,
    )
    envelope = runner.build_envelope(
        payload=data.input,
        run_id=run.id,
        callback_base_url=settings.callback_base_url,
        continuation_flow_id=definition.continuation_flow_id,
        callback_token=settings.callback_token,
        fields=form_schema.schema_field_names(
            version.input_schema or form_schema.extract_form_schema(version.langflow_json)
        ),
    )
    run_id = run.id
    lfx_flow_id = version.lfx_flow_id
    session_id = data.session_id or str(run.id)

    async def event_gen() -> AsyncIterator[str]:
        yield f"event: run\ndata: {json.dumps({'run_id': str(run_id)})}\n\n"
        try:
            async with (
                httpx.AsyncClient(
                    base_url=lfx.base_url, timeout=lfx.timeout, headers=lfx._headers()
                ) as client,
                client.stream(
                    "POST",
                    f"/flows/{lfx_flow_id}/stream",
                    json={"input_value": envelope, "session_id": session_id},
                ) as response,
            ):
                if response.status_code == 404:
                    raise LfxNotFound(f"flow {lfx_flow_id} unknown to LFX")
                response.raise_for_status()
                async for chunk in response.aiter_text():
                    yield chunk
        except (LfxNotFound, httpx.HTTPError) as exc:
            yield f"event: error\ndata: {json.dumps({'error': str(exc)})}\n\n"
        # Finalize from a fresh session: the request session may be closed.
        with Session(get_engine()) as fresh:
            try:
                fresh_run = store.get_run(fresh, run_id)
                status = "pending_approval" if runner._task_exists(fresh, run_id) else "completed"
                store.set_run_result(fresh, fresh_run, status, {"streamed": True})
            except store.NotFoundError:
                status = "failed"
        yield f"event: run_finished\ndata: {json.dumps({'run_id': str(run_id), 'status': status})}\n\n"

    return StreamingResponse(event_gen(), media_type="text/event-stream")


@flows_router.get("/{flow_id}/runs", response_model=list[RunRead])
def list_runs(
    flow_id: UUID,
    session: Annotated[Session, Depends(get_db)],
    user: Annotated[User, Depends(get_current_user)],
):
    # Strict tenant scoping: the session decides, query params cannot widen.
    try:
        return [_read_run(r) for r in store.list_runs(session, flow_id, user.tenant_id)]
    except store.NotFoundError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc


@flows_router.get("/langflow/available")
async def list_langflow_flows(
    langflow: Annotated[LangflowClient, Depends(get_langflow)],
    _admin: Annotated[User, Depends(require_admin)],
):
    """Browse flows on the Langflow authoring server (admin import source)."""
    try:
        return await langflow.list_flows()
    except LangflowUnavailable as exc:
        logger.warning("langflow unavailable: %s", exc)
        raise HTTPException(status_code=503, detail="authoring server unavailable") from exc
    except LangflowError as exc:
        logger.warning("langflow error: %s", exc)
        raise HTTPException(status_code=502, detail="authoring server error") from exc


@flows_router.post("/import", response_model=FlowImportRead, status_code=201)
async def import_flow(
    data: FlowImportRequest,
    session: Annotated[Session, Depends(get_db)],
    langflow: Annotated[LangflowClient, Depends(get_langflow)],
    lfx: Annotated[LfxClient, Depends(get_lfx)],
    _admin: Annotated[User, Depends(require_admin)],
):
    """Import a flow from Langflow: export -> version -> (deploy to LFX)."""
    try:
        definition, version = await import_from_langflow(
            session,
            langflow,
            lfx,
            data.langflow_flow_id,
            name=data.name,
            author=data.author,
            deploy=data.deploy,
        )
    except LangflowNotFound as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except LangflowUnavailable as exc:
        logger.warning("langflow unavailable: %s", exc)
        raise HTTPException(status_code=503, detail="authoring server unavailable") from exc
    except LangflowError as exc:
        logger.warning("langflow error: %s", exc)
        raise HTTPException(status_code=502, detail="authoring server error") from exc
    except store.ConflictError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc
    except store.InvalidError as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc
    except LfxInvalidFlow as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc
    except LfxConflict as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc
    except LfxUnavailable as exc:
        logger.warning("lfx unavailable: %s", exc)
        raise HTTPException(status_code=503, detail="flow runtime unavailable") from exc
    except LfxError as exc:
        logger.warning("lfx error: %s", exc)
        raise HTTPException(status_code=502, detail="flow runtime error") from exc
    return FlowImportRead(definition=_read_definition(definition), version=_read_version(version))


@flows_router.get("/{flow_id}/input-schema", response_model=InputSchemaRead)
def get_input_schema(
    flow_id: UUID,
    session: Annotated[Session, Depends(get_db)],
):
    """Form schema of the latest deployed version (404 when none)."""
    try:
        version = store.get_latest_deployed_version(session, flow_id)
    except (store.NotFoundError, store.InvalidError) as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    cached = version.input_schema or form_schema.extract_form_schema(version.langflow_json)
    return InputSchemaRead(
        flow_id=flow_id,
        version_no=version.version_no,
        json_schema=cached.get("json_schema", {}),
        ui_schema=cached.get("ui_schema", {}),
        fallback=cached.get("fallback"),
    )


@flows_router.get("/runs/{run_id}", response_model=RunRead)
def get_run(
    run_id: UUID,
    session: Annotated[Session, Depends(get_db)],
    user: Annotated[User, Depends(get_current_user)],
):
    """Read a run. Other tenants' runs 404 (strict session scoping)."""
    try:
        run = store.get_run(session, run_id)
    except store.NotFoundError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    if run.tenant_id != user.tenant_id:
        raise HTTPException(status_code=404, detail=f"run {run_id} not found")
    return _read_run(run)


CALLBACK_TOKEN_HEADER = "X-Callback-Token"


def _check_callback_token(token: str | None) -> None:
    """LFX-served flows prove they carry our run envelope (no session there)."""
    expected = settings.callback_token
    if not expected:
        raise HTTPException(status_code=503, detail="flow callbacks are disabled")
    if not token or not secrets.compare_digest(token, expected):
        raise HTTPException(status_code=401, detail="invalid callback token")


@approvals_router.post("", response_model=ApprovalRead, status_code=201)
def create_approval(
    data: ApprovalCreate,
    session: Annotated[Session, Depends(get_db)],
    callback_token: Annotated[str | None, Header(alias=CALLBACK_TOKEN_HEADER)] = None,
):
    """Flow callback contract: report a terminal approval state for a run."""
    _check_callback_token(callback_token)
    try:
        return _read_task(store.create_approval_task(session, data))
    except store.NotFoundError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except store.ConflictError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc
    except store.InvalidError as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc


@approvals_router.get("", response_model=list[ApprovalRead])
def list_approvals(
    session: Annotated[Session, Depends(get_db)],
    user: Annotated[User, Depends(get_current_user)],
):
    return [_read_task(t) for t in store.list_approval_tasks(session, user.tenant_id)]


@approvals_router.post("/{task_id}/decision", response_model=DecisionRead)
async def decide_approval(
    task_id: UUID,
    data: DecisionRequest,
    session: Annotated[Session, Depends(get_db)],
    lfx: Annotated[LfxClient, Depends(get_lfx)],
    user: Annotated[User, Depends(get_current_user)],
):
    """Decide an open task. Approve starts the continuation flow (if any).

    Identity comes from the session: the approver is the logged-in user,
    and the continuation is the task's own (never client-chosen).
    """
    try:
        task = store.get_task(session, task_id)
    except store.NotFoundError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc

    # Resolve the continuation first: bad references fail before any state change.
    continuation_version = None
    if data.decision == "approve" and task.continuation_flow_id is not None:
        try:
            continuation_version = store.get_latest_deployed_version(
                session, task.continuation_flow_id
            )
        except (store.NotFoundError, store.InvalidError) as exc:
            raise HTTPException(status_code=422, detail=str(exc)) from exc
    try:
        task = store.decide_task(
            session,
            task,
            decision="approved" if data.decision == "approve" else "rejected",
            approver=user.email,
            note=data.note,
        )
    except store.ConflictError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc

    # Note: the decision is committed before the continuation runs. If LFX is
    # down, the task stays decided and the continuation can be started
    # manually from the preserved task payload.
    continuation_run = None
    if continuation_version is not None:
        payload = task.payload
        if isinstance(payload.get("facts"), dict):
            payload = payload["facts"]
        try:
            continuation_run = await runner.execute_run(
                session,
                lfx,
                continuation_version,
                payload=payload,
                tenant_id=task.tenant_id,
                session_id=f"continuation-of-{task.run_id}",
                callback_base_url=settings.callback_base_url,
                callback_token=settings.callback_token,
            )
        except LfxNotFound as exc:
            raise HTTPException(status_code=502, detail=f"LFX drift: {exc}") from exc
        except LfxUnavailable as exc:
            raise HTTPException(status_code=503, detail=str(exc)) from exc
        except LfxError as exc:
            raise HTTPException(status_code=502, detail=str(exc)) from exc
    return DecisionRead(
        task=_read_task(task),
        continuation_run=_read_run(continuation_run) if continuation_run else None,
    )