import asyncio
from datetime import UTC, datetime

import pytest
from fastapi import WebSocketDisconnect
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"}


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.
    """
    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.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_text(self) -> str:
            if self.frames:
                return self.frames.pop(0)
            raise WebSocketDisconnect(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 == []