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

"""Reconciliation: backend/flows/*.json (source of truth) -> DB + LFX.

The ONLY place that derives registry state from the canonical directory.
Triggers (CLI, POST /api/flows/sync, embedded watcher) call
sync_from_directory(); single-document callers (e.g. /import) use
sync_document(). Everything is async (the LFX client is async) and guarded
by a module lock so concurrent triggers serialize on SQLite's single writer.
"""

from __future__ import annotations

import asyncio
import json
import logging
import threading
import uuid
from collections.abc import Callable
from pathlib import Path
from uuid import UUID

from sqlmodel import Session
from watchfiles import DefaultFilter, awatch

from app.db import get_engine
from app.models import FlowDefinition, FlowVersion
from app.settings import settings

from . import export, store
from .lfx_client import LfxClient, LfxError
from .schemas import FlowDefinitionCreate, FlowVersionCreate, SyncReport

logger = logging.getLogger(__name__)

# Serializes all writers in this process (sync engine + manual API writes).
sync_lock = threading.Lock()


def lfx_id_for(definition_id: UUID, version_no: int) -> str:
    """Deterministic LFX id (a real UUID: LFX validates the format).

    Same (definition, version) always maps to the same id, so re-syncs and
    crash recovery overwrite instead of orphaning LFX registrations.
    """
    return str(uuid.uuid5(uuid.NAMESPACE_URL, f"kirche-einfach/{definition_id}/v{version_no}"))


def _latest_version(session: Session, flow_id: UUID) -> FlowVersion | None:
    versions = store.list_versions(session, flow_id)
    return versions[-1] if versions else None


async def sync_document(
    session: Session,
    lfx: LfxClient,
    doc: dict,
    *,
    source_filename: str,
    author: str = "",
    deploy: bool = True,
) -> tuple[FlowDefinition, FlowVersion | None, str]:
    """Reconcile one parsed document. Returns (definition, version, action).

    Actions: "added" (new definition + version), "updated" (new version),
    "skipped" (content unchanged). Raises store.ConflictError on name
    clashes, store.InvalidError on invalid JSON, LfxError when deploy fails.
    With deploy=False a version is recorded without touching LFX.
    """
    store.validate_flow_json(doc)
    name = doc.get("name") or Path(source_filename).stem
    with sync_lock:
        definition = store.get_definition_by_source_path(session, source_filename)
        if definition is None:
            definition = store.create_definition(session, FlowDefinitionCreate(name=name))
            definition.source_path = source_filename
            session.add(definition)
            session.commit()
            session.refresh(definition)
            action = "added"
        else:
            action = "updated"
            if definition.name != name:
                store.set_definition_name(session, definition, name)

        latest = _latest_version(session, definition.id)
        if latest is not None and export.content_hash(latest.langflow_json) == export.content_hash(
            doc
        ):
            return definition, latest, "skipped"

        if not deploy:
            version = store.create_version(
                session,
                definition.id,
                FlowVersionCreate(langflow_json=doc, author=author),
            )
            return definition, version, action

        version_no = store.next_version_no(session, definition.id)
        lfx_id = lfx_id_for(definition.id, version_no)
        uploaded = await lfx.upload_flow(doc, flow_id=lfx_id, replace=True)
        version = store.create_version(
            session,
            definition.id,
            FlowVersionCreate(langflow_json=doc, author=author),
        )
        if version.version_no != version_no:  # pragma: no cover - lock prevents this
            raise store.ConflictError(f"concurrent version write for flow {definition.id}")
        version = store.mark_deployed(session, version, uploaded.id)
        if latest is not None and latest.lfx_flow_id and latest.lfx_flow_id != uploaded.id:
            try:
                await lfx.delete_flow(latest.lfx_flow_id)
            except LfxError as exc:
                logger.warning(
                    "LFX unregister %s failed (continuing): %s",
                    latest.lfx_flow_id,
                    exc,
                )
        return definition, version, action


async def delete_definition_tree(
    session: Session, lfx: LfxClient, definition: FlowDefinition
) -> dict[str, int]:
    """Remove a definition whose canonical file is gone.

    LFX registrations go first (best-effort, fail-closed for new runs), the
    DB subtree (tasks, runs, versions, definition) after. Deterministic LFX
    ids make any crash in between self-healing on the next sync.
    """
    for version in store.list_versions(session, definition.id):
        if version.lfx_flow_id:
            try:
                await lfx.delete_flow(version.lfx_flow_id)
            except LfxError as exc:
                logger.warning(
                    "LFX unregister %s failed (continuing): %s",
                    version.lfx_flow_id,
                    exc,
                )
    with sync_lock:
        return store.delete_definition_cascade(session, definition)


async def sync_from_directory(
    session: Session, lfx: LfxClient, flows_dir: str | Path
) -> SyncReport:
    """Reconcile the whole canonical directory. Per-file problems land in
    report.errors (never raised); a missing directory raises FileNotFoundError
    so misconfiguration fails loudly instead of wiping the registry.
    """
    flows_dir = Path(flows_dir)
    report = SyncReport()
    if not flows_dir.is_dir():
        raise FileNotFoundError(f"flows directory missing: {flows_dir}")
    files = {path.name: path for path in export.iter_flow_files(flows_dir)}
    for filename, path in files.items():
        try:
            doc = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, ValueError) as exc:
            report.errors[filename] = f"unreadable: {exc}"
            continue
        if not isinstance(doc, dict):
            report.errors[filename] = "top-level JSON must be an object"
            continue
        try:
            _, _, action = await sync_document(session, lfx, doc, source_filename=filename)
        except (store.InvalidError, store.ConflictError, store.NotFoundError, LfxError) as exc:
            report.errors[filename] = str(exc)
            continue
        getattr(report, action).append(path.stem)
    known = {
        definition.source_path
        for definition in store.list_definitions(session)
        if definition.source_path
    }
    for stale in sorted(known - set(files)):
        definition = store.get_definition_by_source_path(session, stale)
        if definition is None:  # pragma: no cover - defensive
            continue
        try:
            await delete_definition_tree(session, lfx, definition)
        except store.NotFoundError as exc:
            report.errors[stale] = str(exc)
            continue
        report.deleted.append(Path(stale).stem)
    return report


def default_session_factory() -> Callable[[], Session]:
    return lambda: Session(get_engine())


def default_lfx_factory() -> Callable[[], LfxClient]:
    return lambda: LfxClient(settings.lfx_url, api_key=settings.lfx_api_key or None)


def run_sync_blocking(
    flows_dir: str | Path,
    *,
    session_factory: Callable[[], Session] | None = None,
    lfx_factory: Callable[[], LfxClient] | None = None,
) -> SyncReport:
    """Blocking one-shot sync for worker threads (lifespan, watcher)."""
    session = (session_factory or default_session_factory())()
    try:
        lfx = (lfx_factory or default_lfx_factory())()
        return asyncio.run(sync_from_directory(session, lfx, Path(flows_dir)))
    finally:
        session.close()


class _JsonFilter(DefaultFilter):
    def __call__(self, change, path: str) -> bool:  # type: ignore[override]
        return super().__call__(change, path) and path.endswith(".json")


async def watch_forever(
    stop: asyncio.Event,
    flows_dir: str | Path,
    run_sync: Callable[[], SyncReport] | None = None,
) -> None:
    """Run until `stop` is set: sync on every batch of .json changes.

    All failures are logged, never raised — a broken file must not kill the
    watcher. `run_sync` is injectable for tests.
    """
    flows_dir = Path(flows_dir)
    run = run_sync or (lambda: run_sync_blocking(flows_dir))
    while not stop.is_set() and not flows_dir.is_dir():
        logger.error("flows dir missing, retrying in 30s: %s", flows_dir)
        try:
            await asyncio.wait_for(stop.wait(), timeout=30)
        except TimeoutError:
            pass
    if stop.is_set():
        return
    logger.info("watching %s", flows_dir)
    async for _ in awatch(flows_dir, watch_filter=_JsonFilter(), debounce=1000, step=100):
        if stop.is_set():
            break
        try:
            report = await asyncio.to_thread(run)
        except FileNotFoundError as exc:
            logger.error("flow sync failed: %s", exc)
            continue
        except Exception:
            logger.exception("flow sync failed")
            continue
        logger.info(
            "flow sync: +%d ~%d -%d skipped=%d errors=%d",
            len(report.added),
            len(report.updated),
            len(report.deleted),
            len(report.skipped),
            len(report.errors),
        )