"""Sync engine tests: backend/flows/*.json -> DB + LFX (LFX real, DB sqlite)."""

import asyncio
import json
import shutil
import time
from pathlib import Path

import pytest

from app.flows import store, sync
from app.flows.lfx_client import LfxNotFound
from app.flows.schemas import SyncReport
from app.flows.sync import lfx_id_for
from app.settings import settings

from .conftest import FIXTURES_DIR, langflow_export_doc, load_flow_fixture


def _seed_dir(tmp_path: Path, name: str = "simple_rule.json") -> Path:
    shutil.copy(FIXTURES_DIR / name, tmp_path / name)
    return tmp_path


async def test_sync_adds_new_flow(session, lfx_server, lfx_client, tmp_path):
    _seed_dir(tmp_path)
    report = await sync.sync_from_directory(session, lfx_client, tmp_path)
    assert report.added == ["simple_rule"]
    assert report.errors == {}
    definition = store.get_definition_by_source_path(session, "simple_rule.json")
    assert definition is not None
    assert definition.name == "simple_rule"
    versions = store.list_versions(session, definition.id)
    assert [v.version_no for v in versions] == [1]
    assert versions[0].lfx_flow_id == lfx_id_for(definition.id, 1)
    assert versions[0].deployed_at is not None


async def test_sync_skips_ignored_files(session, lfx_server, lfx_client, tmp_path):
    _seed_dir(tmp_path)
    (tmp_path / ".flowsignore").write_text("simple_rule\n", encoding="utf-8")
    report = await sync.sync_from_directory(session, lfx_client, tmp_path)
    assert report.added == [] and report.errors == {}
    assert store.get_definition_by_source_path(session, "simple_rule.json") is None


async def test_sync_skips_unchanged(session, lfx_server, lfx_client, tmp_path):
    _seed_dir(tmp_path)
    first = await sync.sync_from_directory(session, lfx_client, tmp_path)
    assert first.added == ["simple_rule"]
    second = await sync.sync_from_directory(session, lfx_client, tmp_path)
    assert second.skipped == ["simple_rule"]
    assert second.added == second.updated == []
    definition = store.get_definition_by_source_path(session, "simple_rule.json")
    assert len(store.list_versions(session, definition.id)) == 1


async def test_sync_updates_on_change_and_replaces_lfx(session, lfx_server, lfx_client, tmp_path):
    _seed_dir(tmp_path)
    await sync.sync_from_directory(session, lfx_client, tmp_path)
    definition = store.get_definition_by_source_path(session, "simple_rule.json")
    old_lfx_id = lfx_id_for(definition.id, 1)

    doc = load_flow_fixture("simple_rule.json")
    doc["description"] = "geaendert"
    (tmp_path / "simple_rule.json").write_text(json.dumps(doc))

    report = await sync.sync_from_directory(session, lfx_client, tmp_path)
    assert report.updated == ["simple_rule"]
    versions = store.list_versions(session, definition.id)
    assert [v.version_no for v in versions] == [1, 2]
    assert versions[1].lfx_flow_id == lfx_id_for(definition.id, 2)
    # Superseded LFX registration is gone.
    with pytest.raises(LfxNotFound):
        await lfx_client.flow_info(old_lfx_id)


async def test_sync_deletes_missing_file_with_cascade(session, lfx_server, lfx_client, tmp_path):
    _seed_dir(tmp_path)
    await sync.sync_from_directory(session, lfx_client, tmp_path)
    definition = store.get_definition_by_source_path(session, "simple_rule.json")
    version = store.list_versions(session, definition.id)[0]
    run = store.create_run(session, version, input={"a": 1}, tenant_id="default")
    lfx_id = lfx_id_for(definition.id, 1)

    (tmp_path / "simple_rule.json").unlink()
    report = await sync.sync_from_directory(session, lfx_client, tmp_path)
    assert report.deleted == ["simple_rule"]
    assert store.get_definition_by_source_path(session, "simple_rule.json") is None
    with pytest.raises(store.NotFoundError):
        store.get_run(session, run.id)
    with pytest.raises(LfxNotFound):
        await lfx_client.flow_info(lfx_id)


async def test_sync_reports_invalid_json(session, lfx_server, lfx_client, tmp_path):
    (tmp_path / "kaputt.json").write_text("{kein json")
    report = await sync.sync_from_directory(session, lfx_client, tmp_path)
    assert "kaputt.json" in report.errors
    assert report.added == []
    assert store.list_definitions(session) == []


async def test_sync_missing_dir_raises(session, lfx_server, lfx_client, tmp_path):
    with pytest.raises(FileNotFoundError):
        await sync.sync_from_directory(session, lfx_client, tmp_path / "gibts-nicht")


async def test_sync_rename_updates_definition_name(session, lfx_server, lfx_client, tmp_path):
    _seed_dir(tmp_path)
    await sync.sync_from_directory(session, lfx_client, tmp_path)
    doc = load_flow_fixture("simple_rule.json")
    doc["name"] = "umbenannt"
    (tmp_path / "simple_rule.json").write_text(json.dumps(doc))
    report = await sync.sync_from_directory(session, lfx_client, tmp_path)
    assert report.updated == ["simple_rule"]
    definition = store.get_definition_by_source_path(session, "simple_rule.json")
    assert definition.name == "umbenannt"


async def test_sync_endpoint_returns_report(
    client, lfx_server, tmp_path, monkeypatch, login, override_langflow
):
    await login()
    override_langflow()  # empty Langflow: file-only pass
    monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
    _seed_dir(tmp_path)
    response = await client.post("/api/flows/sync")
    assert response.status_code == 200, response.text
    body = response.json()
    assert body["added"] == ["simple_rule"]
    assert body["pulled"] == []
    assert body["errors"] == {}

    response = await client.post("/api/flows/sync")
    assert response.json()["skipped"] == ["simple_rule"]


@pytest.mark.parametrize(
    ("seed", "flow_name", "flow_id", "extra_flows", "extra_docs", "expected_file"),
    [
        (
            None,
            "Gehaltserhöhung",
            "22222222-3333-4444-8555-666666666666",
            [],
            {},
            "Gehaltserhöhung.json",
        ),
        (
            # Seeded from simple_rule.json: updates in place instead of
            # duplicating as simple-rule.json (slug drift).
            "simple_rule.json",
            "simple_rule",
            "44444444-5555-4666-8777-888888888888",
            [],
            {},
            "simple_rule.json",
        ),
        (
            None,
            "Gut",
            "33333333-4444-4555-8666-777777777777",
            [{"id": "kaputt-id", "name": "Kaputt"}],
            {},
            "Gut.json",
        ),
    ],
    ids=["pull-new", "pull-reuse", "pull-error-continues"],
)
async def test_sync_endpoint_pull_matrix(
    client,
    lfx_server,
    tmp_path,
    monkeypatch,
    login,
    override_langflow,
    seed,
    flow_name,
    flow_id,
    extra_flows,
    extra_docs,
    expected_file,
):
    await login()
    monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
    if seed is not None:
        _seed_dir(tmp_path, seed)
    override_langflow(
        flows=[{"id": flow_id, "name": flow_name}, *extra_flows],
        docs={flow_id: langflow_export_doc(flow_name, flow_id), **extra_docs},
    )
    response = await client.post("/api/flows/sync")
    assert response.status_code == 200, response.text
    body = response.json()
    assert body["pulled"] == [flow_name]
    assert body["added"] == [flow_name]
    assert (tmp_path / expected_file).is_file()
    if flow_name == "Gut":
        assert "kaputt-id" in body["errors"]
    else:
        assert body["errors"] == {}
        assert not (tmp_path / "simple-rule.json").exists()

    # Second pass: same content pulls again but syncs nothing new.
    response = await client.post("/api/flows/sync")
    assert response.status_code == 200, response.text
    body = response.json()
    assert body["pulled"] == [flow_name]
    assert body["skipped"] == [flow_name]


@pytest.mark.parametrize(
    ("list_status", "expected_status"),
    [(0, 503), (500, 502)],
)
async def test_sync_endpoint_langflow_failures(
    client,
    lfx_server,
    tmp_path,
    monkeypatch,
    login,
    override_langflow,
    list_status,
    expected_status,
):
    """Listing failures map to 503/502 (parametrized)."""
    await login()
    monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
    override_langflow(flows=[], list_status=list_status)
    response = await client.post("/api/flows/sync")
    assert response.status_code == expected_status


async def test_watch_forever_triggers_on_file_change(tmp_path):
    import contextlib

    calls: list[str] = []
    stop = asyncio.Event()

    def run():
        calls.append("sync")
        return SyncReport()

    task = asyncio.create_task(sync.watch_forever(stop, tmp_path, run_sync=run))
    try:
        await asyncio.sleep(0.5)  # let the watcher arm
        (tmp_path / "neu.json").write_text("{}")
        deadline = time.time() + 15
        while not calls and time.time() < deadline:
            await asyncio.sleep(0.1)
        assert calls == ["sync"]
    finally:
        stop.set()
        task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await task