"""Planning a rename and its document part (sections 1.5 and 5.1 of
docs/private/id_refac_rename.md, "The key rename operation" in id_refac.md)."""

import ast
import inspect
import json
import time
from collections.abc import Iterator
from datetime import UTC, datetime
from pathlib import Path
from types import SimpleNamespace

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import JSON, String
from sqlmodel import Session, SQLModel, delete
from sqlmodel.sql.sqltypes import AutoString

from fluksio.core.config import settings
from fluksio.core.db import engine
from fluksio.flow.alerts import AlertsConfig, Channel
from fluksio.flow.controller import FlowController
from fluksio.flow.dashboards import (
    DASHBOARD_BINDINGS,
    WIDGET_BINDINGS,
    DashboardDef,
    DashboardStore,
    SettingDef,
    WidgetDef,
)
from fluksio.flow.executor import ExecutionService
from fluksio.flow.messages import MessageSpec
from fluksio.flow.namespace import FLOW_FIELDS, Rename, names
from fluksio.flow.queue import MemoryWorkQueue, WorkItem
from fluksio.flow.rename import SQL_NAMES, RenamePlan, plan_rename, rewrite_documents
from fluksio.flow.schemas import FlowDef, FlowInput, FlowOrigin, NodeDef
from fluksio.flow.state import MemoryState
from fluksio.flow.store import FlowStore
from fluksio.models import AlertSent, Run, RunArtifact, RunMetric, RunNode
from tests.utils.stores import created

NOW = datetime.now(UTC)


def _flow(name: str, *nodes: NodeDef, **fields) -> FlowDef:
    return FlowDef(name=name, nodes=list(nodes), **fields)


def _reads(node_id: str, *messages: str) -> NodeDef:
    return NodeDef(id=node_id, requires=[MessageSpec(name=m) for m in messages])


def _writes(node_id: str, *messages: str) -> NodeDef:
    return NodeDef(id=node_id, provides=[MessageSpec(name=m) for m in messages])


@pytest.fixture
def world(tmp_path: Path) -> Iterator[SimpleNamespace]:
    """`heating`, published with a draft over it; `lights` reading it, published;
    `notes` reading it, only a draft; `heating_extra`, which only looks alike;
    and `home`, a dashboard bound to it, published with a draft over it."""
    store = FlowStore(tmp_path / "flows")
    dashboards = DashboardStore(store)
    queue = MemoryWorkQueue()
    controller = FlowController(
        store, state=MemoryState(), execution=ExecutionService(queue)
    )

    heating = _flow(
        "heating",
        _writes("sensor", "temp", "heating.setpoint"),
        inputs=[FlowInput(spec=MessageSpec(name="heating.target"))],
        outputs=["temp", "heating.setpoint"],
    )
    store.write_flow(heating)
    draft = store.read_flow("heating", draft=True)
    draft.nodes.append(_reads("valve", "heating.temp"))
    store.write_draft(draft, draft.version)
    store.write_flow(_flow("lights", _reads("lamp", "heating.temp", "lights.level")))
    created(store, _flow("notes", _reads("note", "heating.mode")))
    store.write_flow(_flow("heating_extra", _writes("x", "heating_extra.x")))

    home = DashboardDef(
        name="home",
        settings={"theme": SettingDef(message="heating.mode", dtype="str")},
        widgets=[
            WidgetDef(
                id="t",
                type="stat",
                title="heating.temp",
                config={"message": "heating.temp"},
            ),
            WidgetDef(
                id="s",
                type="slider",
                config={
                    "target": "heating.setpoint",
                    "reading": {"message": "heating.temp", "label": "heating.temp"},
                },
            ),
            WidgetDef(id="n", type="markdown", config={"text": "heating.temp"}),
        ],
    )
    dashboards.write(home)
    current = dashboards.read("home", draft=True)
    dashboards.write_draft(
        current.model_copy(update={"title": "Home"}), current.version
    )
    dashboards.write(DashboardDef(name="away"))

    yield SimpleNamespace(
        store=store, dashboards=dashboards, controller=controller, queue=queue
    )


def _plan(
    world, new: str = "warmth", kind: str = "flow", obj: str = "heating", alerts=None
) -> RenamePlan:
    return plan_rename(
        world.controller, world.dashboards, kind, obj, new, alerts or AlertsConfig()
    )


def _codes(found) -> set[str]:
    return {one.code for one in found}


def _files(root: Path) -> dict[str, bytes]:
    return {
        str(path.relative_to(root)): path.read_bytes()
        for path in sorted(root.rglob("*"))
        if path.is_file() and ".git" not in path.parts
    }


def _apply(world, rewrites) -> None:
    for rewrite in rewrites:
        for path, _, after in rewrite.files:
            (world.store.root / path).write_text(after)


def test_a_plan_lists_the_flow_what_names_it_and_what_moves(world):
    world.controller.state.set("heating.temp", 20.5)
    world.controller.state.set("__version__:heating.temp", 3)
    world.queue.add(
        WorkItem(
            kind="cascade",
            node="lights.lamp",
            flow="lights",
            outputs={"heating.temp": 1},
        )
    )
    alerts = AlertsConfig(
        channels=[
            Channel(
                name="board", kind="dashboard", config={"message": "heating.alerts"}
            ),
            Channel(name="mail", kind="smtp", config={"to": "heating.x@example.com"}),
        ]
    )
    with Session(engine) as session:
        session.add(Run(id="r-plan", created_at=NOW, flow_id="heating", status="ok"))
        session.add(RunMetric(run_id="r-plan", name="heating.loss", step=0))
        session.add(
            RunArtifact(
                run_id="r-plan",
                name="heating.plot",
                digest="d",
                size=1,
                media_type="image/png",
            )
        )
        session.add(
            RunNode(
                run_id="r-plan", node="heating.sensor", outputs='{"heating.temp": 1}'
            )
        )
        session.add(AlertSent(key="node_failed:heating.sensor", ts=time.time()))
        session.add(AlertSent(key="node_failed:heating_extra.x", ts=time.time()))
        session.commit()
    try:
        plan = _plan(world, alerts=alerts)
    finally:
        with Session(engine) as session:
            for model in (RunMetric, RunArtifact, RunNode, Run, AlertSent):
                session.execute(delete(model))
            session.commit()

    assert plan.allowed, plan.refusals
    assert (plan.name, plan.new_name, plan.version) == ("heating", "warmth", 2)
    listed = {
        (d.kind, d.id): (d.published_version, d.draft_version, d.fields)
        for d in plan.documents
    }
    assert listed == {
        # Its own qualified names: three in the published variant, four in
        # the draft.
        ("flow", "heating"): (1, 2, 4),
        ("flow", "lights"): (1, None, 1),
        ("flow", "notes"): (None, 1, 1),
        ("dashboard", "home"): (1, 2, 4),
    }
    assert plan.fingerprint.startswith("sha256:")
    assert plan.history is not None and plan.history.model_dump() == {
        "runs": 1,
        "metric_rows": 1,
        "artifact_rows": 1,
        "output_rows": 1,
    }
    assert plan.alerts is not None and plan.alerts.model_dump() == {
        "channels": 1,
        "ledger_rows": 1,
    }
    assert plan.runtime is not None
    assert (
        plan.runtime.durable,
        plan.runtime.state_keys,
        plan.runtime.queue_items,
    ) == (False, 2, 1)
    assert _codes(plan.warnings) == {
        "external_references",
        "drafts_rewritten",
        "not_durable",
        "engine_pause",
    }


def test_a_dry_run_changes_no_bytes(world):
    before = _files(world.store.root)
    head = world.store._git("rev-parse", "HEAD").stdout

    _plan(world)
    _plan(world, "house", "dashboard", "home")

    assert _files(world.store.root) == before
    assert world.store._git("rev-parse", "HEAD").stdout == head


def test_the_fingerprint_follows_every_listed_documents_version(world):
    first = _plan(world)
    assert _plan(world).fingerprint == first.fingerprint

    lights = world.store.read_flow("lights")
    world.store.write_draft(
        lights.model_copy(update={"title": "Lights"}), lights.version
    )

    assert _plan(world).fingerprint != first.fingerprint


def test_a_rename_rewrites_bindings_and_nothing_else(world):
    legacy = world.dashboards.root / "legacy" / "dashboard.json"
    legacy.parent.mkdir()
    legacy.write_text(
        json.dumps(
            {
                "name": "legacy",
                "version": 4,
                "pages": [
                    {
                        "sections": [
                            {
                                "widgets": [
                                    {
                                        "id": "w",
                                        "type": "stat",
                                        "config": {"message": "heating.temp"},
                                    }
                                ]
                            }
                        ]
                    }
                ],
            },
            indent=2,
        )
    )
    device = world.store.root / "_devices" / "panel" / "device.json"
    device.parent.mkdir(parents=True)
    device.write_text('{"name": "panel", "dashboards": ["home", "legacy"]}\n')
    before = _files(world.store.root)

    rewrites = rewrite_documents(
        world.store, world.dashboards, "flow", "heating", Rename("heating", "warmth")
    )
    _apply(world, rewrites)

    changed = {path for r in rewrites for path, _, _ in r.files}
    after = _files(world.store.root)
    # The stability set: no file appears or goes, and nothing else changes.
    assert after.keys() == before.keys()
    assert {path for path in after if after[path] != before[path]} == changed
    assert "_devices/panel/device.json" not in changed
    for kind, stem, paths in (
        ("flow", "*/flow*.json", FLOW_FIELDS),
        ("dashboard", "_dashboards/*/dashboard*.json", DASHBOARD_BINDINGS),
    ):
        for path in world.store.root.glob(stem):
            data = json.loads(path.read_text())
            assert data.get("id", path.parent.name) == path.parent.name
            bound = names(data, paths)
            assert not [n for n in bound if n.startswith("heating.")], (kind, path)
    assert (
        world.store.read_flow("heating_extra").nodes[0].provides[0].name
        == "heating_extra.x"
    )

    renamed = world.store.read_flow("heating", draft=True)
    assert renamed.name == "warmth" and renamed.version == 3
    assert [s.name for s in renamed.nodes[0].provides] == ["temp", "warmth.setpoint"]
    assert renamed.inputs[0].spec.name == "warmth.target"
    assert renamed.outputs == ["temp", "warmth.setpoint"]
    assert world.store.read_flow("heating").version == 3
    assert (
        world.store.read_flow("notes", draft=True).nodes[0].requires[0].name
        == "warmth.mode"
    )

    home = world.dashboards.read("home", draft=True)
    assert (home.version, home.title) == (3, "Home")
    assert home.settings["theme"].message == "warmth.mode"
    assert home.widgets[0].messages == ["warmth.temp"]
    # Labels and free text are the person's.
    assert home.widgets[0].title == "heating.temp"
    assert home.widgets[1].target == "warmth.setpoint"
    assert home.widgets[1].reading == {
        "message": "warmth.temp",
        "label": "heating.temp",
    }
    assert home.widgets[2].config == {"text": "heating.temp"}
    assert world.dashboards.read("home").version == 3
    assert json.loads(legacy.read_text())["pages"][0]["sections"][0]["widgets"][0][
        "config"
    ] == {"message": "warmth.temp"}


def test_renaming_back_restores_the_documents_but_for_versions(world):
    def unversioned() -> dict:
        return {
            path: {k: v for k, v in json.loads(text).items() if k != "version"}
            for path, text in _files(world.store.root).items()
            if path.endswith(".json") and not path.endswith("runtime.json")
        }

    before = unversioned()
    _apply(
        world,
        rewrite_documents(
            world.store,
            world.dashboards,
            "flow",
            "heating",
            Rename("heating", "warmth"),
        ),
    )
    assert unversioned() != before
    _apply(
        world,
        rewrite_documents(
            world.store,
            world.dashboards,
            "flow",
            "heating",
            Rename("warmth", "heating"),
        ),
    )
    assert unversioned() == before


def test_a_dashboard_rename_changes_its_key_and_versions_only(world):
    plan = _plan(world, "house", "dashboard", "home")
    assert plan.allowed and plan.version == 2
    assert [
        (d.kind, d.id, d.published_version, d.draft_version, d.fields)
        for d in plan.documents
    ] == [("dashboard", "home", 1, 2, 0)]
    assert (plan.alerts, plan.history, plan.runtime, plan.warnings) == (
        None,
        None,
        None,
        [],
    )
    assert _codes(_plan(world, "away", "dashboard", "home").refusals) == {"name_taken"}

    (rewrite,) = rewrite_documents(
        world.store, world.dashboards, "dashboard", "home", Rename("home", "house")
    )
    _apply(world, [rewrite])
    published, draft = (
        world.dashboards.read("home"),
        world.dashboards.read("home", draft=True),
    )
    assert (published.name, published.version, draft.name, draft.version) == (
        "house",
        3,
        "house",
        3,
    )
    assert draft.widgets[0].messages == ["heating.temp"]


@pytest.mark.parametrize(
    ("new", "code"),
    [
        ("Warmth", "invalid_name"),
        ("f_abcdefgh", "reserved_shape"),
        ("heating", "same_name"),
        ("lights", "name_taken"),
        ("notes", "name_taken"),
    ],
)
def test_a_key_that_cannot_be_taken_is_refused(world, new, code):
    plan = _plan(world, new)
    assert not plan.allowed
    assert code in _codes(plan.refusals)


def test_code_defined_flows_are_refused(world):
    lights = world.store.read_flow("lights")
    world.store.write_flow(lights.model_copy(update={"origin": FlowOrigin()}))
    plan = _plan(world)
    assert [(r.code, r.ids) for r in plan.refusals] == [
        ("code_defined_dependent", ["lights"])
    ]

    heating = world.store.read_flow("heating")
    world.store.write_flow(heating.model_copy(update={"origin": FlowOrigin()}))
    assert _codes(_plan(world).refusals) == {"code_defined", "code_defined_dependent"}


def test_an_active_run_of_an_affected_flow_is_refused(world):
    with Session(engine) as session:
        session.add(
            Run(id="r-busy", created_at=NOW, flow_id="lights", status="running")
        )
        session.add(
            Run(
                id="r-train",
                created_at=NOW,
                flow_id="trainer",
                for_node="heating.model",
                status="queued",
            )
        )
        session.add(
            Run(id="r-other", created_at=NOW, flow_id="heating_extra", status="running")
        )
        session.commit()
    try:
        plan = _plan(world)
    finally:
        with Session(engine) as session:
            session.execute(delete(Run))
            session.commit()
    assert [(r.code, sorted(r.ids)) for r in plan.refusals] == [
        ("run_active", ["r-busy", "r-train"])
    ]


def test_what_still_lives_under_the_new_key_is_refused(world):
    state = world.controller.state
    # A node's own key is its data, even when it reads like a name.
    state.set("__node__:lights.lamp:last:warmth.temp", 1)
    assert _plan(world).allowed

    state.set("__last_seen__:lights.lamp:warmth.temp", 1)
    world.queue.add(WorkItem(kind="cascade", node="warmth.sensor", flow="warmth"))
    plan = _plan(world)
    assert [r.code for r in plan.refusals] == ["target_in_use"]
    assert "__last_seen__:lights.lamp:warmth.temp" in plan.refusals[0].detail
    assert world.controller.key_residue("warmth") == [
        "__last_seen__:lights.lamp:warmth.temp",
        "queued work of 'warmth.sensor'",
    ]

    state.delete("__last_seen__:lights.lamp:warmth.temp")
    world.queue.clear_flow("warmth")
    away = world.dashboards.read("away")
    world.dashboards.write(
        away.model_copy(
            update={
                "widgets": [
                    WidgetDef(id="w", type="stat", config={"message": "warmth.temp"})
                ]
            }
        )
    )
    plan = _plan(world)
    assert [(r.code, r.ids) for r in plan.refusals] == [("target_in_use", ["away"])]


def test_history_that_would_collide_or_overflow_is_refused(world):
    long = "heating." + "x" * (128 - len("heating."))
    with Session(engine) as session:
        session.add(Run(id="r-hist", created_at=NOW, flow_id="heating", status="ok"))
        session.add(RunMetric(run_id="r-hist", name="heating_v2.loss", step=0))
        session.add(RunMetric(run_id="r-hist", name=long, step=0))
        session.commit()
    try:
        assert _codes(_plan(world, "heating_v2").refusals) == {
            "history_collision",
            "name_too_long",
        }
        # A key no longer than the old one fits.
        assert _codes(_plan(world, "warmth").refusals) == set()
    finally:
        with Session(engine) as session:
            for model in (RunMetric, Run):
                session.execute(delete(model))
            session.commit()


def test_another_engine_or_an_old_redis_is_refused(world, monkeypatch):
    monkeypatch.setattr(
        MemoryWorkQueue, "other_consumers", lambda self, ms: ["engine-b"]
    )
    monkeypatch.setattr(MemoryWorkQueue, "server_version", lambda self: (6, 2))
    assert _codes(_plan(world).refusals) == {"other_engine", "redis_too_old"}


def test_what_a_rename_leaves_behind_is_said(world):
    # A flow whose id is the new key, under a key of its own.
    world.store.create(_flow("elsewhere"), "warmth")
    world.dashboards.write(DashboardDef(name="heating_results"))
    world.queue.park(
        "heating", WorkItem(kind="cascade", node="heating.sensor", flow="heating")
    )

    plan = _plan(world)
    assert plan.allowed
    assert {"new_name_is_id", "results_dashboard", "paused"} <= _codes(plan.warnings)


def test_every_binding_a_widget_reads_is_in_the_key_table():
    """A new binding in an accessor fails here until the table knows it, so a
    rename cannot leave it behind. Keys that hold no message name are listed
    as such."""
    plain = {
        "source",
        "dtype",
        "inner_dtype",
        "request_dtype",
        "history",
        "points",
        "format",
        "counters",
        "ids",
        "group",
        "flow",
        "latest",
    }
    tree = ast.parse(inspect.getsource(WidgetDef))
    read = {
        call.args[0].value
        for call in ast.walk(tree)
        if isinstance(call, ast.Call)
        and isinstance(call.func, ast.Attribute)
        and call.func.attr == "get"
        and call.args
        and isinstance(call.args[0], ast.Constant)
    }
    bound = {part for path in WIDGET_BINDINGS for part in path if part != "*"}
    assert read - plain == bound
    assert set(SettingDef.model_fields) - {"value", "dtype"} == {"message"}
    assert ("settings", "*", "message") in DASHBOARD_BINDINGS


def test_every_history_column_is_classified():
    """A column added to the history fails here until it says whether it
    names a message, which a rename then rewrites (`SQL_NAMES`), or not."""
    unnamed = {
        # Ids, which never change: flows and nodes are filed by id.
        "run": {
            "id",
            "flow",
            "commit",
            "origin_commit",
            "code_digest",
            "params",
            "passed",
            "declared",
            "params_digest",
            "group_id",
            "parent_id",
            "idempotency_key",
            "external_id",
            "cause",
            "for_node",
            "status",
            "status_reason",
            "labels",
            "needs",
            "result",
            "engine",
            "actor",
        },
        "run_node": {
            "run_id",
            "node",
            "status",
            "worker",
            "error",
            "logs",
            "cache_key",
            "cached_from",
        },
        "run_metric": {"run_id", "node"},
        "run_artifact": {"run_id", "digest", "media_type", "filename", "node"},
        "flow_run": {"id", "flow", "source", "parent", "status"},
        "metric_minute": {"flow", "node"},
        "engine_event": {"type", "flow", "node", "detail", "actor", "run"},
        "flow_tombstone": {"id", "name"},
    }
    # Tables that hold no flow, node or message at all.
    unrelated = {
        "user",
        "oauth_client",
        "oauth_authorization_code",
        "oauth_refresh_token",
        "flavor",
        "worker_token",
        "connector_state",
        "alembic_version",
    }
    found: dict[str, set[str]] = {}
    for table in SQLModel.metadata.sorted_tables:
        if table.name in unrelated:
            continue
        found[table.name] = {
            column.name
            for column in table.columns
            if isinstance(column.type, String | AutoString | JSON)
        }
    classified = dict(unnamed)
    for table, column in SQL_NAMES:
        classified.setdefault(table, set()).add(column)
    assert found == classified


def test_planning_is_a_superusers(
    client: TestClient, superuser_token_headers, normal_user_token_headers
):
    flows = f"{settings.API_V1_STR}/flows"
    made = client.post(
        f"{flows}/", headers=superuser_token_headers, json={"name": "plan_me"}
    )
    assert made.status_code == 201, made.text
    flow_id = made.json()["definition"]["id"]
    try:
        body = {"new_name": "plan_me_too"}
        assert (
            client.post(
                f"{flows}/{flow_id}/rename/plan",
                headers=normal_user_token_headers,
                json=body,
            ).status_code
            == 403
        )
        assert (
            client.post(
                f"{flows}/f_nothere0/rename/plan",
                headers=superuser_token_headers,
                json=body,
            ).status_code
            == 404
        )
        plan = client.post(
            f"{flows}/{flow_id}/rename/plan", headers=superuser_token_headers, json=body
        )
        assert plan.status_code == 200, plan.text
        assert (plan.json()["name"], plan.json()["allowed"]) == ("plan_me", True)

        boards = f"{settings.API_V1_STR}/dashboards"
        board = client.post(
            f"{boards}/", headers=superuser_token_headers, json={"name": "plan_board"}
        ).json()
        try:
            plan = client.post(
                f"{boards}/{board['id']}/rename/plan",
                headers=superuser_token_headers,
                json=body,
            )
            assert plan.status_code == 200 and plan.json()["kind"] == "dashboard"
        finally:
            client.delete(f"{boards}/{board['id']}", headers=superuser_token_headers)
    finally:
        client.delete(f"{flows}/{flow_id}", headers=superuser_token_headers)