"""A key rename's journal, its steps and what the next start does with one a
crash interrupted (section 4 of docs/private/id_refac_rename.md). Memory
always, Redis with FLUKSIO_TEST_REDIS_HOST (and optional PORT)."""

import os
import shutil
import subprocess
import uuid
from collections import Counter
from collections.abc import Iterator
from datetime import UTC, datetime
from pathlib import Path
from types import SimpleNamespace

import pytest
from sqlmodel import Session, col, delete, select

from fluksio.core.config import settings
from fluksio.core.db import engine
from fluksio.flow import identity
from fluksio.flow import rename as renaming
from fluksio.flow import state as state_module
from fluksio.flow.alerts import AlertsConfig, Channel
from fluksio.flow.controller import FlowController
from fluksio.flow.dashboards import DashboardDef, DashboardStore, WidgetDef
from fluksio.flow.executor import ExecutionService
from fluksio.flow.identity import plan_ids
from fluksio.flow.messages import MessageSpec
from fluksio.flow.namespace import Rename
from fluksio.flow.queue import MemoryWorkQueue, RedisWorkQueue, WorkItem
from fluksio.flow.rename import (
    RenameConflict,
    RenameResult,
    apply_documents,
    begin,
    finish,
    plan_rename,
    read_journal,
    recover_documents,
    recover_runtime,
    repeated,
    rewrite_documents,
    roll_back,
    run_steps,
)
from fluksio.flow.schemas import FlowDef, FlowInput, NodeDef
from fluksio.flow.state import MemoryState, RedisState
from fluksio.flow.store import FlowStore
from fluksio.models import AlertSent, EngineEvent, Run, RunArtifact, RunMetric, RunNode
from tests.utils.rename import leftovers, stability
from tests.utils.stores import created

HEAT = Rename("heating", "warmth")
BACK = Rename("warmth", "heating")
#: The flows whose documents name `heating`: itself, `lights` and `notes`.
AFFECTED = ["heating", "lights", "notes"]
#: Document variants the rename rewrites: two of `heating`, one each of
#: `lights` and `notes`, two of the dashboard `home`.
DOCUMENTS = 6
#: Outputs a node returned, as the stage cache keeps them.
CACHED = '{"heating.temp":20.5,"lights.level":2}'
NOW = datetime.now(UTC)


class Crash(Exception):
    """The process stopping where it stands."""


def git(root: Path, *args: str) -> str:
    return subprocess.run(
        ["git", "-C", str(root), *args], capture_output=True, text=True, check=True
    ).stdout


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 _reads(node_id: str, *messages: str) -> NodeDef:
    return NodeDef(id=node_id, requires=[MessageSpec(name=m) for m in messages])


@pytest.fixture(scope="module")
def template(tmp_path_factory: pytest.TempPathFactory) -> Path:
    """The store each test copies: `heating`, published with a draft over it;
    `lights` reading it; `notes`, a draft reading it; `heating_extra`, which
    only looks alike; and `home`, a dashboard bound to it with a draft over it.
    Every flow has its key as its id, as one from before ids does."""
    root = tmp_path_factory.mktemp("template") / "flows"
    store = FlowStore(root)
    dashboards = DashboardStore(store)
    store.write_flow(
        FlowDef(
            name="heating",
            nodes=[
                NodeDef(
                    id="sensor",
                    provides=[
                        MessageSpec(name="temp"),
                        MessageSpec(name="heating.set"),
                    ],
                )
            ],
            inputs=[FlowInput(spec=MessageSpec(name="heating.target"))],
        )
    )
    draft = store.read_flow("heating", draft=True)
    draft.nodes.append(_reads("valve", "heating.temp"))
    store.write_draft(draft, draft.version)
    store.write_flow(FlowDef(name="lights", nodes=[_reads("lamp", "heating.temp")]))
    created(store, FlowDef(name="notes", nodes=[_reads("note", "heating.mode")]))
    store.write_flow(FlowDef(name="heating_extra", nodes=[_reads("x", "lights.x")]))
    dashboards.write(
        DashboardDef(
            name="home",
            widgets=[
                WidgetDef(id="t", type="stat", config={"message": "heating.temp"})
            ],
        )
    )
    home = dashboards.read("home", draft=True)
    dashboards.write_draft(home.model_copy(update={"title": "Home"}), home.version)
    return root


@pytest.fixture
def world(
    template: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> Iterator[SimpleNamespace]:
    """The template's store, alert channels and history: runs of `heating`,
    of `lights` reading it, of `heating_extra` and of a deleted flow that
    once had the key `heating`."""
    root = tmp_path / "flows"
    shutil.copytree(template, root)
    monkeypatch.setattr(settings, "RENAME_JOURNAL_FILE", tmp_path / "rename.json")
    monkeypatch.setattr(settings, "ALERTS_FILE", tmp_path / "alerts.json")
    channels = [
        Channel(name="board", kind="dashboard", config={"message": "heating.alert"}),
        Channel(name="phone", kind="ntfy", config={"topic": "heating.alert"}),
    ]
    settings.ALERTS_FILE.write_text(
        AlertsConfig(channels=channels).model_dump_json(indent=2)
    )
    with Session(engine) as session:
        for run_id, flow_id in (
            ("r1", "heating"),
            ("r2", "heating"),
            ("r3", "lights"),
            ("r4", "heating_extra"),
            ("r9", "f_gone2xa"),
        ):
            session.add(
                Run(
                    id=run_id,
                    flow_id=flow_id,
                    created_at=NOW,
                    status="ok",
                    params={"note": "heating.temp"},
                )
            )
        session.commit()
        session.add_all(
            [
                RunMetric(run_id="r1", name="heating.loss", step=0, node="heating.s"),
                RunMetric(run_id="r1", name="heating.loss", step=1, node="heating.s"),
                RunMetric(run_id="r1", name="heating_extra.x", step=0),
                RunMetric(run_id="r3", name="heating.temp", step=0, node="lights.lamp"),
                RunMetric(run_id="r4", name="heating_extra.x", step=0),
                RunMetric(run_id="r9", name="heating.loss", step=0),
                RunArtifact(run_id="r2", name="heating.img", node="heating.s"),
                RunArtifact(run_id="r9", name="heating.img"),
                RunNode(run_id="r1", node="heating.s", cache_key="k1", outputs=CACHED),
                RunNode(
                    run_id="r2",
                    node="heating.s",
                    cache_key="k2",
                    cached_from="r1",
                    outputs=CACHED,
                ),
                RunNode(
                    run_id="r3",
                    node="lights.lamp",
                    cache_key="k3",
                    outputs='{"lights.level":2}',
                ),
                RunNode(
                    run_id="r9", node="f_gone2xa.s", cache_key="k9", outputs=CACHED
                ),
                AlertSent(key="node_error:heating.s", ts=1.0),
                AlertSent(key="flow_quarantined:heating", ts=1.0),
                AlertSent(key="node_error:heating_extra.x", ts=1.0),
                EngineEvent(
                    ts=NOW,
                    type="node_error",
                    flow_id="heating",
                    node="heating.s",
                    detail="heating.temp went missing",
                ),
            ]
        )
        session.commit()
    store = FlowStore(root)
    yield SimpleNamespace(root=root, store=store, dashboards=DashboardStore(store))
    with Session(engine) as session:
        for model in (RunMetric, RunArtifact, RunNode, Run, AlertSent, EngineEvent):
            session.execute(delete(model))
        session.commit()


def _redis_server() -> tuple[str, int]:
    host = os.environ.get("FLUKSIO_TEST_REDIS_HOST")
    if not host:
        pytest.skip("Set FLUKSIO_TEST_REDIS_HOST to run Redis integration checks")
    return host, int(os.environ.get("FLUKSIO_TEST_REDIS_PORT", "6379"))


@pytest.fixture
def backends(request: pytest.FixtureRequest) -> Iterator[SimpleNamespace]:
    """State and queue, and how the next start gets them. Memory ones stand
    in for durable backends here, so a restart hands back the same objects."""
    if request.param == "memory":
        state, queue = MemoryState(), MemoryWorkQueue()
        yield SimpleNamespace(open=lambda: (state, queue))
        return
    host, port = _redis_server()
    namespace = f"test-journal-{uuid.uuid4().hex}"
    opened: list = []

    def open_() -> tuple[RedisState, RedisWorkQueue]:
        state = RedisState(host, port, namespace=namespace)
        queue = RedisWorkQueue(
            host, port, namespace=f"{namespace}-queue", consumer="engine-a"
        )
        opened.extend((state, queue))
        return state, queue

    try:
        yield SimpleNamespace(open=open_)
    finally:
        client = opened[0]._client
        stale = list(client.scan_iter(match=f"{namespace}*"))
        if stale:
            client.delete(*stale)
        for one in opened:
            one.close()


def _item(node: str, kind: str = "cascade", **fields) -> WorkItem:
    return WorkItem(kind=kind, node=node, flow=node.split(".")[0], **fields)


def _fill(state, queue) -> None:
    """Live state and queued work, naming `heating` or not."""
    for key, value in (
        ("heating.temp", 20.5),
        ("__version__:heating.temp", 3),
        ("__ts__:heating.temp", 1.0),
        ("__last_seen__:lights.lamp:heating.temp", 3),
        ("__node__:heating.valve:last:x", 1),
        ("heating_extra.x", 1),
        ("lights.level", 2),
    ):
        state.set(key, value)
    queue.add(_item("heating.acked"))
    queue.ack(*queue.claim(1, 10))
    queue.add(_item("heating.valve", outputs={"heating.temp": 21}))
    queue.add(_item("lights.lamp", outputs={"heating.temp": 21}))
    queue.add(_item("heating_extra.x"))
    queue.add_delayed(_item("heating.pump", kind="flush"), 100.0)
    queue.move_due(150.0)
    (claimed,) = queue.claim(1, 10, due_only=True)
    queue.mark_done(claimed.entry_id, "heating.pump")
    queue.add_delayed(_item("heating.limit"), 9e9)
    queue.park("heating", _item("heating.p"))


def _rename(
    world, state, queue, op: str = "op-1", durable: bool = True
) -> RenameResult:
    """What the rename route does once the engine is held (S8): the journal,
    the documents, the steps and the end."""
    rewrites = rewrite_documents(world.store, world.dashboards, "flow", "heating", HEAT)
    journal = begin(
        op,
        "flow",
        "heating",
        HEAT,
        rewrites,
        actor="admin@example.com",
        durable=durable,
    )
    apply_documents(journal, world.store)
    run_steps(journal, state, queue)
    return finish(journal)


def _recover(world, state, queue) -> None:
    """What the next start does before it serves (`main.lifespan`)."""
    recover_documents(world.root)
    # What assigning ids checks first, and refuses to start on.
    assert plan_ids(world.root).problems == []
    recover_runtime(state, queue)
    queue.release_held()


def _crash_at(patch: pytest.MonkeyPatch, point: str, nth: int) -> None:
    seen: Counter[str] = Counter()

    def reached(name: str) -> None:
        seen[name] += 1
        if seen[name] == nth:
            raise Crash(f"{name} #{nth}")

    if point == "document":
        write = identity._replace

        def replace(path: Path, text: str) -> None:
            write(path, text)
            reached("document")

        patch.setattr(identity, "_replace", replace)
        return

    def fault(name: str) -> None:
        if name == point:
            reached(name)

    patch.setattr(renaming, "fault", fault)


def _audits() -> list[tuple[str, str, str]]:
    with Session(engine) as session:
        return [
            (row.flow_id, row.actor, row.detail)
            for row in session.exec(
                select(EngineEvent).where(col(EngineEvent.type) == "audit")
            )
        ]


#: Every point a crash can stop a rename at, by backend: after each document,
#: the commit, each run's history, each state batch and queue structure, each
#: step, and after the journal is done but before anyone was told.
_EVERYWHERE = [
    *(("document", n) for n in range(1, DOCUMENTS + 1)),
    ("commit", 1),
    *(("sql:run", n) for n in (1, 2, 3)),
    ("alert_ledger", 1),
    ("alerts_file", 1),
    ("state:keys", 1),
    ("state", 1),
    ("queue", 1),
    ("audit", 1),
    ("done", 1),
]
_QUEUE_STRUCTURES = ("due", "backlog", "delayed", "lists", "done markers")
CRASHES = [
    *(("memory", point, n) for point, n in [*_EVERYWHERE, ("queue:queue", 1)]),
    *(
        ("redis", point, n)
        for point, n in [
            *_EVERYWHERE,
            ("state:keys", 2),
            *((f"queue:{one}", 1) for one in _QUEUE_STRUCTURES),
        ]
    ),
]


@pytest.mark.parametrize(
    ("backends", "point", "nth"),
    CRASHES,
    indirect=["backends"],
    ids=[f"{kind}-{point}-{n}" for kind, point, n in CRASHES],
)
def test_a_crash_anywhere_is_undone_or_finished_by_the_next_start(
    world, backends, point: str, nth: int, monkeypatch: pytest.MonkeyPatch
):
    # Small batches, so a crash can fall between two of them.
    monkeypatch.setattr(state_module, "RENAME_BATCH", 2)
    state, queue = backends.open()
    _fill(state, queue)
    counts = state.count(HEAT), queue.count(HEAT)
    files = _files(world.root)
    stable = stability(world.root)
    found = leftovers("heating", world.root, AFFECTED, state, queue)
    assert found

    with monkeypatch.context() as patch:
        _crash_at(patch, point, nth)
        with pytest.raises(Crash):
            _rename(world, state, queue)
    if (point, nth) == ("document", 1):
        # A published document and its draft with different keys, which
        # assigning ids would refuse to start on.
        assert plan_ids(world.root).problems

    state, queue = backends.open()
    _recover(world, state, queue)

    journal = read_journal()
    assert journal is not None
    if point in ("document", "commit"):
        assert journal.phase == "rolled_back"
        assert _files(world.root) == files
        assert leftovers("heating", world.root, AFFECTED, state, queue) == found
        assert (state.count(HEAT), queue.count(HEAT)) == counts
        assert _audits() == []
    else:
        assert journal.phase == "done"
        assert journal.result is not None
        assert leftovers("heating", world.root, AFFECTED, state, queue) == []
        # Moved, none lost and none twice.
        assert (state.count(BACK), queue.count(BACK)) == counts
        assert stability(world.root) == stable
        assert len(_audits()) == 1
    assert git(world.root, "status", "--porcelain") == ""

    # Settled is settled: another start changes nothing.
    settled = _files(world.root), journal
    _recover(world, state, queue)
    assert (_files(world.root), read_journal()) == settled


def test_history_moves_for_the_affected_flows_only(world):
    """By flow id, so a deleted flow that had the key keeps its history, and
    payloads and log text keep whatever they say."""
    _rename(world, MemoryState(), MemoryWorkQueue())

    with Session(engine) as session:
        metrics = session.exec(select(RunMetric.run_id, RunMetric.name)).all()
        artifacts = session.exec(select(RunArtifact.run_id, RunArtifact.name)).all()
        nodes = {
            (row.run_id, row.node): (row.cache_key, row.cached_from, row.outputs)
            for row in session.exec(select(RunNode))
        }
        ledger = session.exec(select(AlertSent.key)).all()
        params = session.exec(select(Run.params)).all()
        details = session.exec(
            select(EngineEvent.detail).where(col(EngineEvent.type) != "audit")
        ).all()
    assert sorted(metrics) == [
        ("r1", "heating_extra.x"),
        ("r1", "warmth.loss"),
        ("r1", "warmth.loss"),
        ("r3", "warmth.temp"),
        ("r4", "heating_extra.x"),
        ("r9", "heating.loss"),
    ]
    assert sorted(artifacts) == [("r2", "warmth.img"), ("r9", "heating.img")]
    moved = '{"lights.level":2,"warmth.temp":20.5}'
    # A renamed output is no longer what its cache key promised, and where
    # it was restored from still holds.
    assert nodes == {
        ("r1", "heating.s"): ("", "", moved),
        ("r2", "heating.s"): ("", "r1", moved),
        ("r3", "lights.lamp"): ("k3", "", '{"lights.level":2}'),
        ("r9", "f_gone2xa.s"): ("k9", "", CACHED),
    }
    assert sorted(ledger) == [
        "flow_quarantined:warmth",
        "node_error:heating_extra.x",
        "node_error:warmth.s",
    ]
    assert params == [{"note": "heating.temp"}] * 5
    assert details == ["heating.temp went missing"]
    assert _audits() == [
        ("heating", "admin@example.com", "renamed 'heating' to 'warmth' [op op-1]")
    ]
    alerts = AlertsConfig.model_validate_json(settings.ALERTS_FILE.read_text())
    assert [c.config for c in alerts.channels] == [
        {"message": "warmth.alert"},
        {"topic": "heating.alert"},
    ]


def test_a_rollback_restores_its_own_files_and_leaves_other_changes(world):
    """Whatever else in the store is uncommitted stays so, and out of the
    revert."""
    extra = world.root / "heating_extra" / "flow.json"
    extra.write_text(extra.read_text().replace('"lights.x"', '"lights.y"'))
    (world.root / "scratch.txt").write_text("mine")
    status = git(world.root, "status", "--porcelain")
    files = _files(world.root)

    state, queue = MemoryState(), MemoryWorkQueue()
    with pytest.MonkeyPatch.context() as patch:
        _crash_at(patch, "commit", 1)
        with pytest.raises(Crash):
            _rename(world, state, queue)
    recover_documents(world.root)

    journal = read_journal()
    assert journal is not None and journal.phase == "rolled_back"
    assert _files(world.root) == files
    assert git(world.root, "status", "--porcelain") == status
    assert git(world.root, "log", "-2", "--format=%s").splitlines() == [
        "Revert rename of flow 'heating' to 'warmth'",
        "Rename flow 'heating' to 'warmth' (heating)",
    ]
    assert sorted(
        git(world.root, "show", "--name-only", "--format=", "HEAD").split()
    ) == (sorted(doc.path for doc in journal.files))


def test_a_commit_that_fails_is_undone_in_process(world, monkeypatch):
    files, head = _files(world.root), git(world.root, "rev-parse", "HEAD")
    rewrites = rewrite_documents(world.store, world.dashboards, "flow", "heating", HEAT)
    journal = begin("op-1", "flow", "heating", HEAT, rewrites)
    with monkeypatch.context() as patch:
        patch.setattr(identity, "_commit", lambda *_: False)
        with pytest.raises(RuntimeError):
            apply_documents(journal, world.store)
    assert world.store.read_flow("heating").name == "warmth"

    roll_back(journal, world.store)

    assert read_journal().phase == "rolled_back"
    assert _files(world.root) == files
    assert git(world.root, "rev-parse", "HEAD") == head
    # The caches read the files again.
    assert world.store.read_flow("heating").name == "heating"
    assert world.store.key_index()["heating"] == "heating"


def test_a_store_without_git_renames_and_recovers(world):
    shutil.rmtree(world.root / ".git")
    files = _files(world.root)
    state, queue = MemoryState(), MemoryWorkQueue()
    with pytest.MonkeyPatch.context() as patch:
        _crash_at(patch, "commit", 1)
        with pytest.raises(Crash):
            _rename(world, state, queue)
    _recover(world, state, queue)
    assert _files(world.root) == files

    _rename(world, state, queue, op="op-2")
    assert leftovers("heating", world.root, AFFECTED, state, queue) == []


def test_a_restart_on_memory_backends_finishes_the_rest(world):
    """What was in memory went with the process, so there is nothing of it
    left to move, and the rest is finished as on Redis."""
    state, queue = MemoryState(), MemoryWorkQueue()
    _fill(state, queue)
    with pytest.MonkeyPatch.context() as patch:
        _crash_at(patch, "sql:run", 2)
        with pytest.raises(Crash):
            _rename(world, state, queue, durable=False)

    fresh = MemoryState(), MemoryWorkQueue()
    _recover(world, *fresh)

    assert read_journal().phase == "done"
    assert leftovers("heating", world.root, AFFECTED, *fresh) == []
    assert fresh[0].keys() == []


def test_a_repeated_request_gets_the_stored_result_and_a_stale_one_409(world):
    version = world.store.read_flow("heating", draft=True).version
    assert repeated("op-1", "heating", version, "heating", version) is None

    result = _rename(world, MemoryState(), MemoryWorkQueue())

    assert [(d.kind, d.id, d.version, d.draft_version) for d in result.documents] == [
        ("flow", "heating", version + 1, version + 1),
        ("flow", "lights", 2, None),
        ("flow", "notes", 2, 2),
        ("dashboard", "home", 3, 3),
    ]
    now = world.store.read_flow("heating", draft=True)
    assert (now.name, now.version) == ("warmth", version + 1)
    assert repeated("op-1", now.name, now.version, "heating", version) == result
    # The same rename asked for again under another id: its key matches,
    # which proves nothing.
    with pytest.raises(RenameConflict) as stale:
        repeated("op-2", now.name, now.version, "heating", version)
    assert (stale.value.status, stale.value.code) == (409, "stale")

    # One still finishing turns every other away, and the planner says so.
    journal = read_journal()
    journal.phase = "forward"
    journal.save()
    for request, code in (("op-1", "completing"), ("op-3", "rename_in_progress")):
        with pytest.raises(RenameConflict) as busy:
            repeated(request, now.name, now.version, now.name, now.version)
        assert (busy.value.status, busy.value.code) == (503, code)
    controller = FlowController(
        world.store,
        state=MemoryState(),
        execution=ExecutionService(MemoryWorkQueue()),
    )
    plan = plan_rename(
        controller, world.dashboards, "flow", "lights", "lamps", AlertsConfig()
    )
    assert [r.code for r in plan.refusals] == ["rename_in_progress"]