import asyncio
from datetime import UTC, datetime

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import func
from sqlmodel import Session, select

from fluksio.api.routes import flows as flows_route
from fluksio.core.config import settings
from fluksio.flow.events import event_bus
from fluksio.models import Run, RunArtifact, RunMetric, RunNode

PREFIX = f"{settings.API_V1_STR}/flows"

WORKING_NODE = """
def process():
    return {"reading": 21.5}
"""

BROKEN_NODE = """
def process():
    raise RuntimeError("boom")
"""


def a_flow(name: str = "demo") -> dict:
    return {
        "name": name,
        "title": "Demo",
        "nodes": [
            {
                "id": "sensor",
                "type": "python",
                "provides": [{"name": "reading", "dtype": "float"}],
            },
            {
                "id": "logger",
                "type": "python",
                "requires": [{"name": "reading", "dtype": "float"}],
            },
        ],
    }


def test_flows_require_authentication(client: TestClient) -> None:
    assert client.get(f"{PREFIX}/").status_code == 401


def test_save_then_read_flow(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    response = client.put(
        f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow()
    )
    assert response.status_code == 200

    response = client.get(f"{PREFIX}/demo", headers=superuser_token_headers)
    assert response.status_code == 200
    body = response.json()
    assert body["definition"]["title"] == "Demo"
    assert {node["id"] for node in body["definition"]["nodes"]} == {"sensor", "logger"}

    # The time beside a version carries its offset, so no client guesses a zone.
    listed = client.get(f"{PREFIX}/", headers=superuser_token_headers).json()["data"]
    summary = next(flow for flow in listed if flow["name"] == "demo")
    assert datetime.fromisoformat(summary["updated_at"]).tzinfo is not None


def test_name_mismatch_is_rejected(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    response = client.put(
        f"{PREFIX}/other", headers=superuser_token_headers, json=a_flow("demo")
    )
    assert response.status_code == 400


def test_broken_node_is_reported_and_siblings_stay_active(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow())
    client.put(
        f"{PREFIX}/demo/nodes/sensor/source",
        headers=superuser_token_headers,
        json={"code": WORKING_NODE},
    )

    response = client.put(
        f"{PREFIX}/demo/nodes/logger/source",
        headers=superuser_token_headers,
        json={"code": "def process(reading:\n"},
    )
    assert response.status_code == 200
    assert response.json()["status"] == "error"

    statuses = client.get(f"{PREFIX}/demo", headers=superuser_token_headers).json()
    by_id = {node["id"]: node for node in statuses["nodes"]}
    assert by_id["demo.sensor"]["status"] == "active"
    assert by_id["demo.logger"]["status"] == "error"


def test_running_a_flow_produces_values(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow())
    client.put(
        f"{PREFIX}/demo/nodes/sensor/source",
        headers=superuser_token_headers,
        json={"code": WORKING_NODE},
    )
    client.put(
        f"{PREFIX}/demo/nodes/logger/source",
        headers=superuser_token_headers,
        json={"code": "def process(reading):\n    return {}\n"},
    )

    response = client.post(
        f"{PREFIX}/demo/run", headers=superuser_token_headers, json={"inputs": {}}
    )
    assert response.status_code == 200
    assert response.json()["values"]["demo.reading"]["value"] == 21.5


def test_editing_does_not_deploy_until_published(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    saved = client.put(
        f"{PREFIX}/staged", headers=superuser_token_headers, json=a_flow("staged")
    ).json()
    assert saved["has_draft"] is True
    # Nothing is running yet, so the engine knows no nodes of this flow.
    assert (
        client.get(f"{PREFIX}/staged/state", headers=superuser_token_headers).json()[
            "nodes"
        ]
        == []
    )

    published = client.post(
        f"{PREFIX}/staged/publish",
        headers=superuser_token_headers,
        json={"version": saved["definition"]["version"]},
    )
    assert published.status_code == 200
    assert published.json()["has_draft"] is False
    assert {
        node["id"]
        for node in client.get(
            f"{PREFIX}/staged/state", headers=superuser_token_headers
        ).json()["nodes"]
    } == {"staged.sensor", "staged.logger"}

    client.delete(f"{PREFIX}/staged", headers=superuser_token_headers)


def test_a_stale_save_is_refused(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    saved = client.put(
        f"{PREFIX}/contested", headers=superuser_token_headers, json=a_flow("contested")
    ).json()
    stale = saved["definition"]

    client.put(
        f"{PREFIX}/contested",
        headers=superuser_token_headers,
        json={**stale, "title": "Mine"},
    )

    response = client.put(
        f"{PREFIX}/contested",
        headers=superuser_token_headers,
        json={**stale, "title": "Theirs"},
    )
    assert response.status_code == 409
    assert response.json()["detail"]["current_version"] == stale["version"] + 1

    client.delete(f"{PREFIX}/contested", headers=superuser_token_headers)


def test_discarding_a_draft_restores_what_is_running(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    saved = client.put(
        f"{PREFIX}/reverted", headers=superuser_token_headers, json=a_flow("reverted")
    ).json()
    client.post(
        f"{PREFIX}/reverted/publish",
        headers=superuser_token_headers,
        json={"version": saved["definition"]["version"]},
    )
    published = client.get(f"{PREFIX}/reverted", headers=superuser_token_headers).json()

    client.put(
        f"{PREFIX}/reverted",
        headers=superuser_token_headers,
        json={**published["definition"], "title": "Scratch that"},
    )

    response = client.post(
        f"{PREFIX}/reverted/discard-draft", headers=superuser_token_headers
    )
    assert response.status_code == 200
    assert response.json()["has_draft"] is False
    assert response.json()["definition"]["title"] == "Demo"

    client.delete(f"{PREFIX}/reverted", headers=superuser_token_headers)


def test_stopping_a_flow_takes_it_off_the_engine(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    saved = client.put(
        f"{PREFIX}/halted", headers=superuser_token_headers, json=a_flow("halted")
    ).json()
    client.post(
        f"{PREFIX}/halted/publish",
        headers=superuser_token_headers,
        json={"version": saved["definition"]["version"]},
    )

    stopped = client.post(f"{PREFIX}/halted/stop", headers=superuser_token_headers)
    assert stopped.status_code == 200
    assert stopped.json()["enabled"] is False
    # Nothing of it is loaded, so there is nothing to run.
    assert (
        client.post(
            f"{PREFIX}/halted/run", headers=superuser_token_headers, json={"inputs": {}}
        ).status_code
        == 409
    )

    started = client.post(f"{PREFIX}/halted/start", headers=superuser_token_headers)
    assert started.json()["enabled"] is True
    assert (
        client.post(
            f"{PREFIX}/halted/run", headers=superuser_token_headers, json={"inputs": {}}
        ).status_code
        == 200
    )

    client.delete(f"{PREFIX}/halted", headers=superuser_token_headers)


def test_pausing_is_reported_back(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    saved = client.put(
        f"{PREFIX}/held", headers=superuser_token_headers, json=a_flow("held")
    ).json()
    client.post(
        f"{PREFIX}/held/publish",
        headers=superuser_token_headers,
        json={"version": saved["definition"]["version"]},
    )

    assert (
        client.post(f"{PREFIX}/held/pause", headers=superuser_token_headers).status_code
        == 200
    )
    assert (
        client.get(f"{PREFIX}/held", headers=superuser_token_headers).json()["paused"]
        is True
    )

    client.post(f"{PREFIX}/held/resume", headers=superuser_token_headers)
    assert (
        client.get(f"{PREFIX}/held", headers=superuser_token_headers).json()["paused"]
        is False
    )

    client.delete(f"{PREFIX}/held", headers=superuser_token_headers)


def test_unconnected_input_is_surfaced(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    flow = a_flow()
    flow["nodes"][0]["provides"] = []  # nothing produces "reading" any more
    client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=flow)

    issues = client.post(
        f"{PREFIX}/demo/validate", headers=superuser_token_headers
    ).json()["issues"]
    assert any(issue["code"] == "unconnected_input" for issue in issues)


def test_delete_flow(
    client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None:
    client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow())
    # Finished: a flow with a live run refuses to be deleted, which is its own
    # test below.
    db.add(Run(id="run-1", flow="demo", status="ok", created_at=datetime.now(UTC)))
    db.add(RunNode(run_id="run-1", node="sensor"))
    db.add(RunMetric(run_id="run-1", name="loss", step=-1))
    db.add(RunArtifact(run_id="run-1", name="model.pt"))
    db.commit()

    assert (
        client.delete(f"{PREFIX}/demo", headers=superuser_token_headers).status_code
        == 200
    )
    assert (
        client.get(f"{PREFIX}/demo", headers=superuser_token_headers).status_code == 404
    )

    # The run outlives the flow: a run record is a record of what ran, and
    # `DELETE /runs?flow=` is the only thing that clears one.
    db.expire_all()
    for model in (Run, RunNode, RunMetric, RunArtifact):
        assert db.exec(select(func.count()).select_from(model)).one() == 1
    assert (
        client.get(
            f"{settings.API_V1_STR}/runs/run-1", headers=superuser_token_headers
        ).status_code
        == 200
    )


def test_delete_flow_is_refused_while_a_run_is_live(
    client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None:
    client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow())
    db.add(
        Run(id="run-live", flow="demo", status="running", created_at=datetime.now(UTC))
    )
    db.commit()

    response = client.delete(f"{PREFIX}/demo", headers=superuser_token_headers)
    assert response.status_code == 409
    assert "run-live" in response.json()["detail"]
    # Its driver is still writing node rows against that id, so nothing may go.
    assert (
        client.get(f"{PREFIX}/demo", headers=superuser_token_headers).status_code == 200
    )


def test_node_types_are_listed(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    types = client.get(f"{PREFIX}/node-types", headers=superuser_token_headers).json()
    by_type = {entry["type"]: entry for entry in types}
    assert by_type["python"]["has_source"] is True
    assert "properties" in by_type["mqtt"]["params_schema"]


def test_rename_flow(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow())

    response = client.post(
        f"{PREFIX}/demo/rename",
        headers=superuser_token_headers,
        json={"new_name": "demo_renamed"},
    )
    assert response.status_code == 200
    assert response.json()["definition"]["name"] == "demo_renamed"

    assert (
        client.get(f"{PREFIX}/demo", headers=superuser_token_headers).status_code == 404
    )
    assert (
        client.get(
            f"{PREFIX}/demo_renamed", headers=superuser_token_headers
        ).status_code
        == 200
    )

    # Put it back so the tests that follow find the flow they expect.
    client.post(
        f"{PREFIX}/demo_renamed/rename",
        headers=superuser_token_headers,
        json={"new_name": "demo"},
    )


def test_rename_onto_a_taken_name_is_refused(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow())
    client.put(
        f"{PREFIX}/occupied", headers=superuser_token_headers, json=a_flow("occupied")
    )

    response = client.post(
        f"{PREFIX}/demo/rename",
        headers=superuser_token_headers,
        json={"new_name": "occupied"},
    )
    assert response.status_code == 409

    client.delete(f"{PREFIX}/occupied", headers=superuser_token_headers)


def test_rename_rejects_an_invalid_name(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow())

    response = client.post(
        f"{PREFIX}/demo/rename",
        headers=superuser_token_headers,
        json={"new_name": "Not A Flow Name"},
    )
    assert response.status_code == 400


def test_a_node_that_raises_answers_with_its_error(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """A manual run fails the way a scheduled one does, not as a 500."""
    saved = client.put(
        f"{PREFIX}/failing", headers=superuser_token_headers, json=a_flow("failing")
    ).json()
    client.put(
        f"{PREFIX}/failing/nodes/sensor/source",
        headers=superuser_token_headers,
        json={"code": BROKEN_NODE},
    )
    client.post(
        f"{PREFIX}/failing/publish",
        headers=superuser_token_headers,
        json={"version": saved["definition"]["version"]},
    )

    response = client.post(
        f"{PREFIX}/failing/nodes/sensor/trigger",
        headers=superuser_token_headers,
        json={"values": {}},
    )

    assert response.status_code == 400
    assert "boom" in response.json()["detail"]

    client.delete(f"{PREFIX}/failing", headers=superuser_token_headers)


def test_acknowledging_a_failure_is_announced_on_the_bus(
    client: TestClient,
    superuser_token_headers: dict[str, str],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Another browser holds the same marker; the event is what clears it."""
    saved = client.put(
        f"{PREFIX}/acked", headers=superuser_token_headers, json=a_flow("acked")
    ).json()
    client.post(
        f"{PREFIX}/acked/publish",
        headers=superuser_token_headers,
        json={"version": saved["definition"]["version"]},
    )

    published: list[dict] = []
    monkeypatch.setattr(event_bus, "publish", published.append)
    response = client.post(
        f"{PREFIX}/acked/nodes/sensor/acknowledge", headers=superuser_token_headers
    )

    assert response.status_code == 200
    event = next(e for e in published if e["type"] == "node_error_acknowledged")
    assert event["node"] == "acked.sensor"

    client.delete(f"{PREFIX}/acked", headers=superuser_token_headers)


def test_renaming_a_flow_leaves_nothing_under_the_old_name(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    saved = client.put(
        f"{PREFIX}/movable", headers=superuser_token_headers, json=a_flow("movable")
    ).json()
    client.put(
        f"{PREFIX}/movable/nodes/sensor/source",
        headers=superuser_token_headers,
        json={"code": WORKING_NODE},
    )
    client.post(
        f"{PREFIX}/movable/publish",
        headers=superuser_token_headers,
        json={"version": saved["definition"]["version"]},
    )
    client.post(
        f"{PREFIX}/movable/run", headers=superuser_token_headers, json={"inputs": {}}
    )

    state = client.app.state.flow_controller.state
    assert [key for key in state.keys() if "movable." in key]

    client.post(
        f"{PREFIX}/movable/rename",
        headers=superuser_token_headers,
        json={"new_name": "moved"},
    )

    # The values repopulate under the new name on the next run; what the old
    # name left behind would sit there for good.
    assert [key for key in state.keys() if "movable." in key] == []

    client.delete(f"{PREFIX}/moved", headers=superuser_token_headers)


@pytest.mark.anyio
async def test_the_live_socket_ends_when_its_client_goes(
    client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    """A departing client ends the stream instead of restarting the read.

    `receive_text` raises for good once the peer has gone, so reading it again
    spun this loop at full tilt: the event loop stalled, `/health` answered
    503 and one queue getter was orphaned per turn.

    It is also what made `fluksio serve` unstoppable. A handler that never
    returns is an ASGI task uvicorn never sees finish, and its graceful
    shutdown waits on those with no timeout — so SIGTERM logged "Waiting for
    background tasks to complete" and nothing else, leaving `serve.pid` behind
    because only SIGKILL ended the process. This returning is what ends that.
    """
    monkeypatch.setattr(
        flows_route, "user_from_token", lambda session, token, **_: object()
    )

    class Peer:
        """One frame, then gone — the way a closed tab leaves."""

        def __init__(self) -> None:
            self.app = client.app
            self.headers: dict[str, str] = {}
            self.frames = ['{"type": "media", "names": []}']
            self.sent: list[str] = []

        async def accept(self) -> None:
            return None

        async def send_text(self, text: str) -> None:
            self.sent.append(text)

        async def receive(self) -> dict[str, object]:
            if self.frames:
                return {"type": "websocket.receive", "text": self.frames.pop(0)}
            return {"type": "websocket.disconnect", "code": 1006}

    peer = Peer()
    subscribed = len(event_bus._subscribers)

    await asyncio.wait_for(flows_route.flow_events(peer, token="not-a-jwt"), timeout=5)

    # The snapshot went out, and the frame in between did not end the stream:
    # reaching the disconnect at all means the second read happened.
    assert peer.sent
    assert peer.frames == []
    assert len(event_bus._subscribers) == subscribed
    # A turn of the loop for the cancellations the handler asked for.
    await asyncio.sleep(0.05)
    parked = [t for t in asyncio.all_tasks() if "Queue.get" in repr(t.get_coro())]
    assert parked == []


def test_saving_a_header_moves_the_node_ports_in_the_draft(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    saved = client.put(
        f"{PREFIX}/header", headers=superuser_token_headers, json=a_flow("header")
    ).json()

    response = client.put(
        f"{PREFIX}/header/nodes/logger/source",
        headers=superuser_token_headers,
        json={"code": "def process(reading, level, gain=2):\n    return {}\n"},
    ).json()

    assert [spec["port"] for spec in response["node"]["requires"]] == [
        "reading",
        "level",
    ]
    assert response["node"]["params"] == {"gain": 2}
    draft = client.get(f"{PREFIX}/header", headers=superuser_token_headers).json()
    assert draft["definition"]["version"] == response["version"]
    assert response["version"] == saved["definition"]["version"] + 1
    logger = next(n for n in draft["definition"]["nodes"] if n["id"] == "logger")
    assert logger["params"] == {"gain": 2}
    # The editor's autosave: a parameter typed away keeps its port for now.
    paused = client.put(
        f"{PREFIX}/header/nodes/logger/source?drop=false",
        headers=superuser_token_headers,
        json={"code": "def process(reading, gain=2):\n    return {}\n"},
    ).json()
    assert paused["node"] is None

    client.delete(f"{PREFIX}/header", headers=superuser_token_headers)


def test_a_shared_node_in_use_is_retired_only_with_detach(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    client.put(
        f"{PREFIX}/shared", headers=superuser_token_headers, json=a_flow("shared")
    )
    client.put(
        f"{PREFIX}/shared/nodes/sensor/source",
        headers=superuser_token_headers,
        json={"code": WORKING_NODE},
    )
    client.post(
        f"{PREFIX}/shared/nodes/sensor/share",
        headers=superuser_token_headers,
        json={"lib_name": "reading"},
    )

    url = f"{PREFIX}/library/reading"
    assert client.delete(url, headers=superuser_token_headers).status_code == 409
    response = client.delete(f"{url}?detach=true", headers=superuser_token_headers)
    assert response.status_code == 200

    assert client.get(f"{PREFIX}/library", headers=superuser_token_headers).json() == []
    source = client.get(
        f"{PREFIX}/shared/nodes/sensor/source", headers=superuser_token_headers
    )
    assert source.json()["code"] == WORKING_NODE

    client.delete(f"{PREFIX}/shared", headers=superuser_token_headers)


def test_a_shared_node_only_the_published_flow_uses_is_kept(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """An unshare in the draft leaves the running flow on the source."""
    client.put(
        f"{PREFIX}/unshared", headers=superuser_token_headers, json=a_flow("unshared")
    )
    shared = client.post(
        f"{PREFIX}/unshared/nodes/sensor/share",
        headers=superuser_token_headers,
        json={"lib_name": "published_reading"},
    ).json()
    client.post(
        f"{PREFIX}/unshared/publish",
        headers=superuser_token_headers,
        json={"version": shared["definition"]["version"]},
    )
    client.post(
        f"{PREFIX}/unshared/nodes/sensor/unshare", headers=superuser_token_headers
    )

    url = f"{PREFIX}/library/published_reading"
    response = client.delete(url, headers=superuser_token_headers)
    assert response.status_code == 409
    assert "unshared.sensor" in response.json()["detail"]

    client.delete(f"{url}?detach=true", headers=superuser_token_headers)
    client.delete(f"{PREFIX}/unshared", headers=superuser_token_headers)


def test_a_shared_source_write_is_audited(
    client: TestClient,
    superuser_token_headers: dict[str, str],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """It goes live at once, so it is recorded like a publish."""
    client.put(
        f"{PREFIX}/audited", headers=superuser_token_headers, json=a_flow("audited")
    )
    client.post(
        f"{PREFIX}/audited/nodes/sensor/share",
        headers=superuser_token_headers,
        json={"lib_name": "audited_reading"},
    )

    published: list[dict] = []
    monkeypatch.setattr(event_bus, "publish", published.append)
    client.put(
        f"{PREFIX}/audited/nodes/sensor/source",
        headers=superuser_token_headers,
        json={"code": WORKING_NODE},
    )

    (audit,) = [e for e in published if e["type"] == "audit"]
    assert audit["action"] == "edited shared node 'audited_reading'"
    assert audit["flow"] == "audited"

    client.delete(
        f"{PREFIX}/library/audited_reading?detach=true",
        headers=superuser_token_headers,
    )
    client.delete(f"{PREFIX}/audited", headers=superuser_token_headers)