"""Renaming a flow's or a dashboard's key over HTTP (sections 5.2 and 5.3 of
docs/private/id_refac_rename.md): who may, what a request is answered with,
what waits for it, and what keeps working by id afterwards."""

import threading
import time
import uuid
from pathlib import Path
from typing import Any

import httpx
import pytest
from fastapi.testclient import TestClient

from fluksio.api.routes import oauth
from fluksio.core.config import settings
from fluksio.flow import rename as renaming
from fluksio.main import app

API = settings.API_V1_STR
FLOWS = f"{API}/flows"
DASHBOARDS = f"{API}/dashboards"
DEVICES = f"{API}/devices"
MESSAGES = f"{API}/messages"

TWICE = "def process(tick):\n    return {'twice': tick * 2}\n"


@pytest.fixture(autouse=True)
def _journal(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    """A journal of the test's own, and pairing unlimited, as in the device
    tests."""
    monkeypatch.setattr(settings, "RENAME_JOURNAL_FILE", tmp_path / "rename.json")
    for bucket in [b for b in oauth._hits if b.startswith("pair:")]:
        del oauth._hits[bucket]


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


def _flow(
    client: TestClient,
    headers: dict[str, str],
    key: str,
    codes: dict[str, str] | None = None,
    **document: Any,
) -> str:
    """A flow under an issued id, its python nodes' *codes* written, and
    published; its id."""
    made = client.post(f"{FLOWS}/", headers=headers, json={"name": key, **document})
    assert made.status_code == 201, made.text
    definition = made.json()["definition"]
    for node_id, code in (codes or {}).items():
        saved = client.put(
            f"{FLOWS}/{definition['id']}/nodes/{node_id}/source",
            headers=headers,
            json={"code": code},
        )
        assert saved.status_code == 200, saved.text
    version = client.get(f"{FLOWS}/{definition['id']}", headers=headers).json()[
        "definition"
    ]["version"]
    published = client.post(
        f"{FLOWS}/{definition['id']}/publish",
        headers=headers,
        json={"version": version},
    )
    assert published.status_code == 200, published.text
    return str(definition["id"])


def _hooked(code_node: dict | None = None) -> dict[str, Any]:
    """A webhook putting ``tick`` in, and what reads it."""
    nodes: list[dict[str, Any]] = [
        {
            "id": "hook",
            "type": "http",
            "params": {"url": "/tick", "method": "POST"},
            "provides": [{"name": "tick", "dtype": "float"}],
        }
    ]
    if code_node is not None:
        nodes.append(code_node)
    return {"nodes": nodes}


def _request(
    client: TestClient,
    headers: dict[str, str],
    obj_id: str,
    new: str,
    kind: str = FLOWS,
    **override: Any,
) -> dict[str, Any]:
    plan = client.post(
        f"{kind}/{obj_id}/rename/plan", headers=headers, json={"new_name": new}
    )
    assert plan.status_code == 200, plan.text
    body = plan.json()
    return {
        "request_id": uuid.uuid4().hex,
        "expected_name": body["name"],
        "expected_version": body["version"],
        "new_name": new,
        "fingerprint": body["fingerprint"],
        **override,
    }


def _in_thread(call) -> tuple[threading.Thread, list[httpx.Response]]:
    answered: list[httpx.Response] = []
    thread = threading.Thread(target=lambda: answered.append(call()), daemon=True)
    thread.start()
    return thread, answered


def _until(check, timeout: float = 20.0) -> None:
    deadline = time.monotonic() + timeout
    while not check():
        assert time.monotonic() < deadline, "timed out"
        time.sleep(0.05)


def test_a_rename_is_a_superusers_and_answered_once(
    client: TestClient,
    superuser_token_headers: dict[str, str],
    normal_user_token_headers: dict[str, str],
) -> None:
    """Every refusal by its code, then the rename, the same answer for the
    same request, and its webhook where it was, by id."""
    su = superuser_token_headers
    key, new = _key(), _key()
    flow_id = _flow(client, su, key, **_hooked())
    try:
        body = _request(client, su, flow_id, new)
        url = f"{FLOWS}/{flow_id}/rename"
        assert (
            client.post(url, headers=normal_user_token_headers, json=body).status_code
            == 403
        )
        assert (
            client.post(f"{FLOWS}/f_nothere0/rename", headers=su, json=body).status_code
            == 404
        )
        for override, status, code in (
            ({"new_name": "Not A Key"}, 422, "invalid_name"),
            ({"expected_version": body["expected_version"] - 1}, 409, "stale"),
            ({"fingerprint": "sha256:moved"}, 409, "plan_changed"),
        ):
            refused = client.post(url, headers=su, json={**body, **override})
            assert (refused.status_code, refused.json()["code"]) == (status, code)
        assert "refusals" in refused.json()["plan"]

        renamed = client.post(url, headers=su, json=body)
        assert renamed.status_code == 200, renamed.text
        result = renamed.json()
        assert (
            result["kind"],
            result["id"],
            result["old_name"],
            result["new_name"],
        ) == (
            "flow",
            flow_id,
            key,
            new,
        )
        assert result["op"] == body["request_id"]
        assert client.post(url, headers=su, json=body).json() == result
        again = client.post(url, headers=su, json={**body, "request_id": "other"})
        assert (again.status_code, again.json()["code"]) == (409, "stale")

        detail = client.get(f"{FLOWS}/{flow_id}", headers=su).json()
        assert detail["definition"]["name"] == new
        # The webhook is under the id, which the rename left alone.
        hooked = client.post(f"/hooks/{flow_id}/tick", json={"tick": 7.0})
        assert hooked.status_code == 200, hooked.text

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

        # Once the cascade it queued has run.
        _until(lambda: f"{new}.tick" in values())
        assert values()[f"{new}.tick"]["value"] == 7.0
        assert f"{key}.tick" not in values()
        summary = client.get(f"{API}/observability/summary", headers=su).json()
        assert summary["rename"] is None
    finally:
        client.delete(f"{FLOWS}/{flow_id}", headers=su)


def test_a_devices_credential_follows_the_renamed_flow(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """A message it may send moves with the flow, and renaming its dashboard
    leaves the device's list as it was."""
    su = superuser_token_headers
    key, new, board, device = _key(), _key(), _key(), _key()
    flow_id = _flow(
        client, su, key, inputs=[{"spec": {"name": "button", "dtype": "float"}}]
    )
    made = client.post(f"{DASHBOARDS}/", headers=su, json={"name": board}).json()
    dash_id = made["id"]
    made["widgets"] = [
        {"id": "s", "type": "slider", "config": {"target": f"{key}.button"}}
    ]
    saved = client.put(f"{DASHBOARDS}/{dash_id}", headers=su, json=made)
    assert saved.status_code == 200, saved.text
    client.post(
        f"{DASHBOARDS}/{dash_id}/publish",
        headers=su,
        json={"version": saved.json()["version"]},
    )
    created = client.post(f"{DEVICES}/{device}", headers=su).json()
    client.put(
        f"{DEVICES}/{device}",
        headers=su,
        json={**created, "dashboards": [dash_id]},
    )
    started = client.post(f"{DEVICES}/pair").json()
    client.post(f"{DEVICES}/{device}/pair", headers=su, json={"code": started["code"]})
    token = client.get(
        f"{DEVICES}/pair/{started['code']}", params={"secret": started["secret"]}
    ).json()["access_token"]
    panel = {"Authorization": f"Bearer {token}"}

    def send(name: str) -> int:
        return client.post(
            f"{MESSAGES}/{name}", headers=panel, json={"value": 1.0}
        ).status_code

    try:
        assert send(f"{key}.button") == 200
        renamed = client.post(
            f"{FLOWS}/{flow_id}/rename",
            headers=su,
            json=_request(client, su, flow_id, new),
        )
        assert renamed.status_code == 200, renamed.text
        assert renamed.json()["documents"][-1]["id"] == dash_id
        assert send(f"{key}.button") == 403
        assert send(f"{new}.button") == 200

        renamed = client.post(
            f"{DASHBOARDS}/{dash_id}/rename",
            headers=su,
            json=_request(client, su, dash_id, f"{board}_b", kind=DASHBOARDS),
        )
        assert renamed.status_code == 200, renamed.text
        assert client.get(f"{DASHBOARDS}/{dash_id}", headers=su).json()["name"] == (
            f"{board}_b"
        )
        listed = client.get(f"{DEVICES}/{device}", headers=su).json()
        assert listed["dashboards"] == [dash_id]
        assert send(f"{new}.button") == 200
    finally:
        client.delete(f"{DEVICES}/{device}", headers=su)
        client.delete(f"{DASHBOARDS}/{dash_id}", headers=su)
        client.delete(f"{FLOWS}/{flow_id}", headers=su)


def test_what_arrives_during_a_rename_waits_or_is_turned_away(
    client: TestClient,
    superuser_token_headers: dict[str, str],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Saves, publishes, creates, deletes, runs, imports, sync and messages
    get 503 while the rename holds the engine, its own flow's webhook too.
    Another flow's webhook is taken, and what it sets off runs once the
    rename lets go. Reads carry on."""
    su = superuser_token_headers
    key, new, other_key = _key(), _key(), _key()
    flow_id = _flow(client, su, key, **_hooked())
    other = _flow(
        client,
        su,
        other_key,
        {"double": TWICE},
        **_hooked(
            {
                "id": "double",
                "type": "python",
                "requires": [{"name": "tick", "dtype": "float"}],
                "provides": [{"name": "twice", "dtype": "float"}],
            }
        ),
    )
    stale = client.get(f"{FLOWS}/{flow_id}", headers=su).json()["definition"]
    holding, go = threading.Event(), threading.Event()
    begin = renaming.begin

    def held_begin(*args: Any, **kwargs: Any) -> Any:
        holding.set()
        assert go.wait(20)
        return begin(*args, **kwargs)

    monkeypatch.setattr(renaming, "begin", held_begin)
    body = _request(client, su, flow_id, new)
    thread, answered = _in_thread(
        lambda: client.post(f"{FLOWS}/{flow_id}/rename", headers=su, json=body)
    )
    try:
        assert holding.wait(20)
        for method, path, payload in (
            ("put", f"{FLOWS}/{flow_id}", stale),
            ("post", f"{FLOWS}/{flow_id}/publish", {"version": stale["version"]}),
            ("post", f"{FLOWS}/", {"name": _key()}),
            ("delete", f"{FLOWS}/{other}", None),
            ("post", f"{API}/runs/flows/{other}", {"inputs": {}}),
            ("post", f"{API}/runs/import", []),
            ("post", f"{MESSAGES}/{other_key}.tick", {"value": 1.0}),
        ):
            refused = client.request(method, path, headers=su, json=payload)
            assert refused.status_code == 503, (path, refused.text)
            assert refused.json()["code"] == "renaming"
            assert refused.headers["retry-after"] == "5"
        assert (
            client.post(f"/hooks/{flow_id}/tick", json={"tick": 1.0}).status_code == 503
        )
        assert (
            client.post(f"/hooks/{other}/tick", json={"tick": 3.0}).status_code == 200
        )
        assert client.get(f"{FLOWS}/{flow_id}", headers=su).status_code == 200
        summary = client.get(f"{API}/observability/summary", headers=su).json()
        assert summary["rename"]["phase"] == "prepared"

        def doubled() -> Any:
            values = client.get(f"{FLOWS}/{other}/state", headers=su).json()["values"]
            return values.get(f"{other_key}.twice", {}).get("value")

        time.sleep(0.3)
        assert doubled() is None
    finally:
        go.set()
        thread.join(30)
    assert answered[0].status_code == 200, answered[0].text
    try:
        _until(lambda: doubled() == 6.0)
        # An editor still holding the old document is refused, not saved back.
        assert (
            client.put(f"{FLOWS}/{flow_id}", headers=su, json=stale).status_code == 409
        )
    finally:
        client.delete(f"{FLOWS}/{other}", headers=su)
        client.delete(f"{FLOWS}/{flow_id}", headers=su)


def test_a_rename_waits_for_running_work_and_can_be_called_off(
    client: TestClient, superuser_token_headers: dict[str, str], tmp_path: Path
) -> None:
    """It says what it waits for, and called off it changes nothing."""
    su = superuser_token_headers
    key, new = _key(), _key()
    go = tmp_path / "go"
    slow = (
        "import os, time\n"
        "def process():\n"
        f"    while not os.path.exists({str(go)!r}):\n"
        "        time.sleep(0.05)\n"
        "    return {'reading': 1.0}\n"
    )
    flow_id = _flow(
        client,
        su,
        key,
        {"sensor": slow},
        nodes=[
            {
                "id": "sensor",
                "type": "python",
                "provides": [{"name": "reading", "dtype": "float"}],
            }
        ],
    )
    trigger = f"{FLOWS}/{flow_id}/nodes/sensor/trigger"
    triggering, triggered = _in_thread(
        lambda: client.post(trigger, headers=su, json={"values": {}})
    )
    body = _request(client, su, flow_id, new)
    try:
        # In, and counted, before the gate closes.
        _until(lambda: f"POST {trigger}" in app.state.rename_gate._admitted)
        thread, answered = _in_thread(
            lambda: client.post(f"{FLOWS}/{flow_id}/rename", headers=su, json=body)
        )

        def waiting() -> Any:
            summary = client.get(f"{API}/observability/summary", headers=su).json()
            return summary["rename"]

        _until(
            lambda: waiting() is not None and waiting()["busy"] == [f"POST {trigger}"]
        )
        assert waiting()["phase"] == "draining"
        assert client.put(f"{FLOWS}/{flow_id}", headers=su, json={}).status_code == 503

        cancel = f"{FLOWS}/{flow_id}/rename/{body['request_id']}"
        assert (
            client.delete(f"{FLOWS}/{flow_id}/rename/other", headers=su).status_code
            == 404
        )
        assert client.delete(cancel, headers=su).status_code == 200
        thread.join(30)
        assert (answered[0].status_code, answered[0].json()["code"]) == (409, "aborted")
        assert waiting() is None
    finally:
        go.touch()
        triggering.join(60)
    assert triggered[0].status_code == 200, triggered[0].text
    try:
        definition = client.get(f"{FLOWS}/{flow_id}", headers=su).json()["definition"]
        assert (definition["name"], definition["version"]) == (
            key,
            body["expected_version"],
        )
        # The gate is open again, and the same rename now goes through.
        renamed = client.post(f"{FLOWS}/{flow_id}/rename", headers=su, json=body)
        assert renamed.status_code == 200, renamed.text
    finally:
        client.delete(f"{FLOWS}/{flow_id}", headers=su)