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

"""Flows + approvals HTTP API."""

from __future__ import annotations

import json
from collections.abc import AsyncIterator, Iterator
from pathlib import Path
from uuid import UUID

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

from app.db import get_engine, get_session
from app.settings import settings

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

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


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


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: Session = Depends(get_db)):
    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: 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: Session = Depends(get_db),
    lfx: LfxClient = Depends(get_lfx),
):
    """Reconcile backend/flows/*.json into the DB + LFX (add/update/delete)."""
    try:
        return await sync_from_directory(session, lfx, Path(settings.flows_dir))
    except FileNotFoundError as exc:
        raise HTTPException(status_code=500, detail=str(exc)) from exc


@flows_router.get("/{flow_id}", response_model=FlowDefinitionRead)
def get_flow(flow_id: UUID, session: 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: Session = Depends(get_db)
):
    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: Session = Depends(get_db)):
    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: Session = Depends(get_db),
    lfx: LfxClient = Depends(get_lfx),
):
    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:
        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 _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,
    stream: bool = Query(default=False),
    session: Session = Depends(get_db),
    lfx: LfxClient = Depends(get_lfx),
):
    try:
        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)
    try:
        run = await runner.execute_run(
            session,
            lfx,
            version,
            payload=data.input,
            tenant_id=data.tenant_id,
            session_id=data.session_id,
            callback_base_url=data.callback_base_url or settings.callback_base_url,
            continuation_flow_id=data.continuation_flow_id,
        )
    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 _read_run(run)


async def _run_stream(session: Session, lfx: LfxClient, version, data: RunRequest):
    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=data.tenant_id,
        session_id=data.session_id,
    )
    envelope = runner.build_envelope(
        payload=data.input,
        run_id=run.id,
        callback_base_url=data.callback_base_url or settings.callback_base_url,
        continuation_flow_id=data.continuation_flow_id,
    )
    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:
                async with 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,
    tenant_id: str | None = Query(default=None),
    session: Session = Depends(get_db),
):
    try:
        return [_read_run(r) for r in store.list_runs(session, flow_id, 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: LangflowClient = Depends(get_langflow)):
    """Browse flows on the Langflow authoring server (admin import source)."""
    try:
        return await langflow.list_flows()
    except LangflowUnavailable as exc:
        raise HTTPException(status_code=503, detail=str(exc)) from exc
    except LangflowError as exc:
        raise HTTPException(status_code=502, detail=str(exc)) from exc


@flows_router.post("/import", response_model=FlowImportRead, status_code=201)
async def import_flow(
    data: FlowImportRequest,
    session: Session = Depends(get_db),
    langflow: LangflowClient = Depends(get_langflow),
    lfx: LfxClient = Depends(get_lfx),
):
    """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:
        raise HTTPException(status_code=503, detail=str(exc)) from exc
    except LangflowError as exc:
        raise HTTPException(status_code=502, 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
    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:
        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 FlowImportRead(
        definition=_read_definition(definition), version=_read_version(version)
    )


@flows_router.get("/runs/{run_id}", response_model=RunRead)
def get_run(
    run_id: UUID,
    tenant_id: str | None = Query(default=None),
    session: Session = Depends(get_db),
):
    """Read a run. With tenant_id, cross-tenant reads 404 (v1 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 tenant_id is not None and run.tenant_id != tenant_id:
        raise HTTPException(status_code=404, detail=f"run {run_id} not found")
    return _read_run(run)


@approvals_router.post("", response_model=ApprovalRead, status_code=201)
def create_approval(data: ApprovalCreate, session: Session = Depends(get_db)):
    """Flow callback contract: report a terminal approval state for a run."""
    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(
    tenant_id: str | None = Query(default=None),
    session: Session = Depends(get_db),
):
    return [_read_task(t) for t in store.list_approval_tasks(session, tenant_id)]


@approvals_router.post("/{task_id}/decision", response_model=DecisionRead)
async def decide_approval(
    task_id: UUID,
    data: DecisionRequest,
    session: Session = Depends(get_db),
    lfx: LfxClient = Depends(get_lfx),
):
    """Decide an open task. Approve starts the continuation flow (if any)."""
    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=data.approver,
            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=data.callback_base_url or settings.callback_base_url,
            )
        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,
    )