"""Flows over HTTP by id, with keys apart from them.

A flow made through the API gets an opaque id unless a superuser pins one, so
most tests here run on both: a pinned ``x_<key>`` and an issued id. A path
that quietly read the key where it meant the id passes for a flow whose id is
its key, and nowhere else.
"""

import uuid
from collections.abc import Callable
from datetime import UTC, datetime
from typing import Any

import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, delete, select

from fluksio.core.config import settings
from fluksio.flow.dashboards import DashboardDef, WidgetDef
from fluksio.flow.store import ISSUED_ID
from fluksio.flow.tombstones import names_of
from fluksio.models import FlowTombstone, Run
from tests.api.routes.test_rename import _until

FLOWS = f"{settings.API_V1_STR}/flows"
DASHBOARDS = f"{settings.API_V1_STR}/dashboards"

SENSOR = "def process():\n    return {'reading': 21.5}\n"
DOUBLE = "def process(reading):\n    return {'doubled': reading * 2}\n"


def a_flow(key: str, mode: str = "live") -> dict[str, Any]:
    return {
        "name": key,
        "mode": mode,
        "nodes": [
            {
                "id": "sensor",
                "type": "python",
                "provides": [{"name": "reading", "dtype": "float"}],
            },
            {
                "id": "double",
                "type": "python",
                "requires": [{"name": "reading", "dtype": "float"}],
                "provides": [{"name": "doubled", "dtype": "float"}],
            },
            {
                "id": "hook",
                "type": "http",
                "params": {"url": "/tick", "method": "POST"},
                "provides": [{"name": "tick", "dtype": "float"}],
            },
        ],
        "outputs": ["doubled"] if mode == "batch" else [],
    }


def a_key() -> str:
    return f"ids_{uuid.uuid4().hex[:8]}"


Make = Callable[..., dict[str, Any]]


@pytest.fixture(params=["pinned", "issued"])
def make(
    request: pytest.FixtureRequest,
    client: TestClient,
    superuser_token_headers: dict[str, str],
) -> Make:
    """Create a flow, pinned to ``x_<key>`` or under an issued id; its detail."""
    made: list[str] = []

    def create(key: str, **document: Any) -> dict[str, Any]:
        body = {**a_flow(key), **document}
        if request.param == "pinned":
            body["id"] = f"x_{key}"
        response = client.post(f"{FLOWS}/", headers=superuser_token_headers, json=body)
        assert response.status_code == 201, response.text
        detail: dict[str, Any] = response.json()
        made.append(detail["definition"]["id"])
        return detail

    yield create  # type: ignore[misc]
    for flow_id in made:
        client.delete(f"{FLOWS}/{flow_id}", headers=superuser_token_headers)


def test_a_flow_is_made_found_and_saved_by_its_id(
    make: Make, client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    key = a_key()
    definition = make(key)["definition"]
    flow_id = definition["id"]
    assert flow_id != key and definition["name"] == key
    assert flow_id == f"x_{key}" or ISSUED_ID.fullmatch(flow_id)

    read = client.get(f"{FLOWS}/{flow_id}", headers=superuser_token_headers)
    assert read.json()["definition"]["name"] == key
    # By its key it is not there: only a client resolves a key.
    assert client.get(
        f"{FLOWS}/{key}", headers=superuser_token_headers
    ).status_code == (404)
    (listed,) = client.get(
        f"{FLOWS}/?name={key}", headers=superuser_token_headers
    ).json()["data"]
    assert (listed["id"], listed["name"]) == (flow_id, key)
    assert (
        client.get(f"{FLOWS}/?name={key}_no", headers=superuser_token_headers).json()[
            "data"
        ]
        == []
    )

    saved = client.put(
        f"{FLOWS}/{flow_id}",
        headers=superuser_token_headers,
        json={**definition, "title": "Retitled"},
    )
    assert saved.status_code == 200, saved.text
    assert saved.json()["definition"]["title"] == "Retitled"
    renamed = client.put(
        f"{FLOWS}/{flow_id}",
        headers=superuser_token_headers,
        json={**saved.json()["definition"], "name": f"{key}_b"},
    )
    assert renamed.status_code == 409

    copy = client.post(
        f"{FLOWS}/{flow_id}/duplicate", headers=superuser_token_headers
    ).json()["definition"]
    assert copy["name"] == f"{key}_copy" and ISSUED_ID.fullmatch(copy["id"])
    client.delete(f"{FLOWS}/{copy['id']}", headers=superuser_token_headers)


def test_a_flow_runs_under_its_key_and_is_reached_by_its_id(
    make: Make, client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """Publish, a message, the nodes running, the state under the key."""
    key = a_key()
    flow_id = make(key)["definition"]["id"]
    for node, code in (("sensor", SENSOR), ("double", DOUBLE)):
        saved = client.put(
            f"{FLOWS}/{flow_id}/nodes/{node}/source",
            headers=superuser_token_headers,
            json={"code": code},
        )
        assert saved.json()["id"] == f"{key}.{node}"
    version = client.get(f"{FLOWS}/{flow_id}", headers=superuser_token_headers).json()[
        "definition"
    ]["version"]
    published = client.post(
        f"{FLOWS}/{flow_id}/publish",
        headers=superuser_token_headers,
        json={"version": version},
    )
    assert published.status_code == 200, published.text
    assert {n["id"]: n["status"] for n in published.json()["nodes"]} == {
        f"{key}.sensor": "active",
        f"{key}.double": "active",
        f"{key}.hook": "active",
    }

    ran = client.post(
        f"{FLOWS}/{flow_id}/run", headers=superuser_token_headers, json={"inputs": {}}
    )
    assert ran.json()["values"][f"{key}.doubled"]["value"] == 43.0
    triggered = client.post(
        f"{FLOWS}/{flow_id}/nodes/double/trigger",
        headers=superuser_token_headers,
        json={"values": {f"{key}.reading": 1.0}},
    )
    assert triggered.json()["values"][f"{key}.doubled"]["value"] == 2.0
    history = client.get(
        f"{FLOWS}/{flow_id}/history/doubled", headers=superuser_token_headers
    ).json()
    assert history["message"] == f"{key}.doubled" and history["points"]

    # The webhook is under the id, which a later key change leaves alone.
    hooked = client.post(f"/hooks/{flow_id}/tick", json={"tick": 7.0})
    assert hooked.status_code == 200, hooked.text
    assert client.post(f"/hooks/{key}/tick", json={"tick": 8.0}).status_code == 404

    def values() -> dict[str, Any]:
        state = client.get(f"{FLOWS}/{flow_id}/state", headers=superuser_token_headers)
        return dict(state.json()["values"])

    # Once the cascade it queued has run.
    _until(lambda: f"{key}.tick" in values())
    assert values()[f"{key}.tick"]["value"] == 7.0

    paused = client.post(f"{FLOWS}/{flow_id}/pause", headers=superuser_token_headers)
    assert paused.status_code == 200
    assert client.get(f"{FLOWS}/{flow_id}", headers=superuser_token_headers).json()[
        "paused"
    ]
    client.post(f"{FLOWS}/{flow_id}/resume", headers=superuser_token_headers)
    stopped = client.post(f"{FLOWS}/{flow_id}/stop", headers=superuser_token_headers)
    assert stopped.json()["enabled"] is False
    assert (
        client.post(
            f"{FLOWS}/{flow_id}/run",
            headers=superuser_token_headers,
            json={"inputs": {}},
        ).status_code
        == 409
    )
    started = client.post(f"{FLOWS}/{flow_id}/start", headers=superuser_token_headers)
    assert started.json()["enabled"] is True
    acknowledged = client.post(
        f"{FLOWS}/{flow_id}/nodes/double/acknowledge", headers=superuser_token_headers
    )
    assert acknowledged.status_code == 200


def test_a_results_dashboard_is_made_from_a_flow_id_under_the_flows_key(
    make: Make, client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    key = a_key()
    flow_id = make(key, mode="batch")["definition"]["id"]
    client.post(
        f"{FLOWS}/{flow_id}/publish",
        headers=superuser_token_headers,
        json={"version": 1},
    )

    made = client.post(
        f"{DASHBOARDS}/from-flow/{flow_id}", headers=superuser_token_headers
    )
    assert made.status_code == 201, made.text
    board = made.json()
    assert board["name"] == f"{key}_results" and ISSUED_ID.fullmatch(board["id"])
    assert board["has_draft"] is False
    again = client.post(
        f"{DASHBOARDS}/from-flow/{flow_id}", headers=superuser_token_headers
    )
    assert again.status_code == 409
    by_key = client.post(
        f"{DASHBOARDS}/from-flow/{key}", headers=superuser_token_headers
    )
    assert by_key.status_code == 404
    client.delete(f"{DASHBOARDS}/{board['id']}", headers=superuser_token_headers)


def test_what_opens_a_flow_or_dashboard_names_it_by_id(
    make: Make, client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """A node goes by ``<key>.<node>``, which is what the canvas matches. Where
    a response leads somewhere, it says the id to go by beside it."""
    headers = superuser_token_headers
    key, reader_key = a_key(), a_key()
    flow_id = make(key)["definition"]["id"]
    for node, code in (("sensor", SENSOR), ("double", DOUBLE)):
        client.put(
            f"{FLOWS}/{flow_id}/nodes/{node}/source",
            headers=headers,
            json={"code": code},
        )
    client.post(
        f"{FLOWS}/{flow_id}/publish",
        headers=headers,
        json={
            "version": client.get(f"{FLOWS}/{flow_id}", headers=headers).json()[
                "definition"
            ]["version"]
        },
    )
    listen = {
        "id": "listen",
        "type": "python",
        "requires": [{"name": f"{key}.reading", "dtype": "float"}],
    }
    reader_id = make(reader_key, nodes=[listen])["definition"]["id"]
    board = client.app.state.dashboard_store.write(  # type: ignore[attr-defined]
        DashboardDef(
            id=f"d_{reader_key}",
            name=reader_key,
            widgets=[
                WidgetDef(
                    id="shown",
                    type="stat",
                    config={"message": f"{key}.reading", "dtype": "float"},
                )
            ],
        )
    )
    try:
        far = client.get(f"{FLOWS}/{reader_id}", headers=headers).json()["endpoints"]
        assert [(e["id"], e["owner_id"]) for e in far] == [
            (f"flow:{key}.sensor", flow_id)
        ]
        near = client.get(f"{FLOWS}/{flow_id}", headers=headers).json()["endpoints"]
        assert (f"dashboard:{board.id}:shown", board.id) in [
            (e["id"], e["owner_id"]) for e in near
        ]

        graph = client.get(f"{FLOWS}/graph", headers=headers).json()
        (sensor,) = [n for n in graph["nodes"] if f"{key}.sensor" in n["members"]]
        assert (sensor["flows"], sensor["flow_ids"]) == ([key], [flow_id])
    finally:
        client.delete(f"{DASHBOARDS}/{board.id}", headers=headers)


def test_who_may_create_what(
    client: TestClient,
    superuser_token_headers: dict[str, str],
    normal_user_token_headers: dict[str, str],
) -> None:
    key = a_key()
    plain = {"name": key, "nodes": [{"id": "hold", "type": "delay"}]}

    # Anyone makes a flow under an issued id, as anyone could save one before.
    made = client.post(f"{FLOWS}/", headers=normal_user_token_headers, json=plain)
    assert made.status_code == 201, made.text
    flow_id = made.json()["definition"]["id"]
    assert ISSUED_ID.fullmatch(flow_id)
    # Code stays a superuser's to put there, and so does choosing an id.
    coded = client.post(
        f"{FLOWS}/",
        headers=normal_user_token_headers,
        json={**plain, "name": f"{key}_code", "nodes": [{"id": "fn"}]},
    )
    assert coded.status_code == 403
    pinned = client.post(
        f"{FLOWS}/",
        headers=normal_user_token_headers,
        json={**plain, "name": f"{key}_pin", "id": f"x_{key}"},
    )
    assert pinned.status_code == 403

    def pin(flow: str, name: str) -> int:
        return client.post(
            f"{FLOWS}/",
            headers=superuser_token_headers,
            json={**plain, "name": name, "id": flow},
        ).status_code

    assert pin("Not-An-Id", f"{key}_a") == 422
    assert pin("graph", f"{key}_b") == 422  # behind `GET /flows/graph`
    assert pin(flow_id, f"{key}_c") == 409  # a live flow's
    assert pin(f"x_{key}", key) == 409  # the key is taken
    assert pin(f"x_{key}", "f_k3q7m2xa") == 400  # shaped like an issued id

    client.delete(f"{FLOWS}/{flow_id}", headers=superuser_token_headers)


def test_a_deleted_flow_keeps_its_name_and_frees_its_key(
    make: Make,
    client: TestClient,
    superuser_token_headers: dict[str, str],
    db: Session,
) -> None:
    key = a_key()
    old = make(key)["definition"]["id"]
    db.add(Run(id=f"run-{key}", flow_id=old, status="ok", created_at=datetime.now(UTC)))
    db.commit()
    controller = client.app.state.flow_controller  # type: ignore[attr-defined]
    controller.state.set(f"{key}.doubled", 1.0)

    deleted = client.delete(f"{FLOWS}/{old}", headers=superuser_token_headers)
    assert deleted.status_code == 200
    assert db.get(FlowTombstone, old).name == key
    assert names_of([old], controller.store) == {old: key}
    # What the old flow left under its key went with it.
    assert controller.key_residue(key) == []

    # Something still writing there keeps the key from a new flow.
    controller.state.set(f"{key}.stray", 1.0)
    refused = client.post(
        f"{FLOWS}/", headers=superuser_token_headers, json=a_flow(key)
    )
    assert refused.status_code == 409 and f"{key}.stray" in refused.json()["detail"]
    controller.forget_flow(key)

    # Under an id of its own, not the one the runs are filed under.
    made = client.post(f"{FLOWS}/", headers=superuser_token_headers, json=a_flow(key))
    assert made.status_code == 201, made.text
    new = made.json()["definition"]["id"]
    assert new != old
    assert db.exec(select(Run).where(Run.flow_id == new)).first() is None
    client.delete(f"{FLOWS}/{new}", headers=superuser_token_headers)
    db.exec(delete(Run).where(Run.id == f"run-{key}"))  # type: ignore[call-overload]
    db.commit()


def test_a_pinned_restore_takes_its_history_back(
    client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None:
    key = a_key()
    pin = f"x_{key}"
    plain = {"name": key, "nodes": []}
    client.post(f"{FLOWS}/", headers=superuser_token_headers, json={**plain, "id": pin})
    client.delete(f"{FLOWS}/{pin}", headers=superuser_token_headers)

    other = client.post(
        f"{FLOWS}/",
        headers=superuser_token_headers,
        json={**plain, "name": f"{key}_b", "id": pin},
    )
    assert other.status_code == 409 and other.json()["key"] == key
    restored = client.post(
        f"{FLOWS}/", headers=superuser_token_headers, json={**plain, "id": pin}
    )
    assert restored.status_code == 201, restored.text
    db.expire_all()
    assert db.get(FlowTombstone, pin) is None
    client.delete(f"{FLOWS}/{pin}", headers=superuser_token_headers)