"""The overview (X6, M6.6c): what each agent is doing, what waits and why,
from items the agents' own code wrote."""

import hashlib
import sqlite3
from datetime import UTC, datetime, timedelta
from pathlib import Path

import pytest
from ui_fixtures import HOSTILE, recorded, store_at

from nisse_core.models import calls
from nisse_dev.loop import items as dev_items
from nisse_dev.loop.items import Loop
from nisse_dev.state import StateStore as DevStore
from nisse_review import state as review
from nisse_review.state import StateStore as ReviewStore
from nisse_ui import links, overview
from nisse_ui import main as ui_main
from nisse_ui.app import App
from nisse_ui.links import Links
from nisse_ui.sources import Source

LINKS = Links(
    gitea="https://gitea.test",
    room="!room:matrix.test",
    repos={"web": "dev/web"},
    hosts={"web": "root@build-web.test"},
)


def page(app: App, path: str = "/?refresh=0") -> str:
    response = app.handle(path)
    assert response.status == 200, response.body.decode()[:500]
    return response.body.decode()


def new_store(root: Path, store_class: type[DevStore] | type[ReviewStore]):
    root.mkdir(parents=True, exist_ok=True)
    return store_at(root, store_class)


def dev_item(
    store: DevStore, item_id: str, position: str, loop: Loop, scope: str = "web"
) -> None:
    store.open_work_item(
        item_id, title=f"Item {item_id}", scope=scope, position="ready", cause="t"
    )
    dev_items.save(store, item_id, position, loop, cause="t")


@pytest.fixture
def dev(tmp_path: Path) -> Source:
    """A nisse-dev store with an item in each place, written by its own
    code: the queue, a VM block, waits, a dependency, and a closed item."""
    store = new_store(tmp_path / "dev", DevStore)
    dev_item(store, "web-1-aaaaaa", "ready", Loop(issue=1, thread="$t1"))
    dev_item(store, "web-2-bbbbbb", "ready", Loop(issue=2, first=1))
    dev_item(store, "web-3-cccccc", "ready", Loop(issue=3), scope="other")
    dev_item(
        store,
        "web-4-dddddd",
        "waiting",
        Loop(
            issue=4,
            wait=dev_items.MERGE,
            note="PR #9 is up for review",
            pr=9,
            pr_url="https://gitea.test/dev/web/pulls/9",
        ),
    )
    dev_item(
        store,
        "web-5-eeeeee",
        "waiting",
        Loop(wait=dev_items.CHAT, note="asked which branch"),
    )
    dev_item(
        store,
        "web-6-ffffff",
        "blocked",
        Loop(issue=6, blocked_by=("dev/lib#12", "dev/web#2")),
    )
    dev_item(store, "web-7-000000", "running", Loop(issue=7))
    store.set_active("web-7-000000", cause="t")
    dev_item(store, "web-8-111111", "closed", Loop(issue=8))
    store.close_work_item("web-8-111111", outcome="merged", cause="t")
    store.close()
    return Source("dev", tmp_path / "dev" / "state.db", "agent")


def view(source: Source, links: Links = LINKS) -> overview.AgentView:
    app = App([source], {source.name: links})
    return app._agent(source, datetime.now(UTC))


def group(found: overview.AgentView, key: str) -> overview.Group:
    return next(g for g in found.groups if g.key == key)


# --- nisse-dev --------------------------------------------------------------------


def test_dev_items_fall_into_their_places_with_their_reasons(dev: Source) -> None:
    found = view(dev)
    assert found.kind == overview.DEV
    assert found.active is not None and found.active.item.id == "web-7-000000"
    assert [r.item.id for r in group(found, "running").rows] == ["web-7-000000"]
    parked = {r.item.id: r for r in group(found, "waiting").rows}
    assert parked["web-4-dddddd"].what == "waiting for merge"
    assert "PR #9 is up for review" in parked["web-4-dddddd"].details
    assert parked["web-5-eeeeee"].what == "talking; your turn"
    [blocked] = group(found, "blocked").rows
    assert [(l.text, l.href) for l in blocked.links][1:] == [
        ("dev/lib#12", "https://gitea.test/dev/lib/issues/12"),
        ("dev/web#2", "https://gitea.test/dev/web/issues/2"),
    ]
    assert [r.item.id for r in found.closed] == ["web-8-111111"]
    assert found.closed[0].what == "merged"


def test_the_queue_is_in_the_order_the_agent_runs_it(dev: Source) -> None:
    # The operator's "do #N next" first, then the oldest.
    queue = group(view(dev), "ready").rows
    assert [(r.item.id, r.mark) for r in queue] == [
        ("web-2-bbbbbb", "1 of 3"),
        ("web-1-aaaaaa", "2 of 3"),
        ("web-3-cccccc", "3 of 3"),
    ]
    assert "you asked for it first" in queue[0].details


def test_an_item_whose_build_vm_is_blocked_is_held_out_of_line(
    tmp_path: Path, dev: Source
) -> None:
    store = store_at(dev.state_db.parent, DevStore)
    store.block_vm(
        "root@build-web.test",
        cause="unreachable",
        reason="no answer on port 22",
        work_item="web-1-aaaaaa",
    )
    dev_item(store, "web-9-222222", "ready", Loop(question=True))
    store.close()
    found = view(dev)
    queue = group(found, "ready").rows
    assert [(r.item.id, r.held) for r in queue] == [
        ("web-3-cccccc", False),  # another project's VM
        ("web-9-222222", False),  # a question is answered anyway (L5)
        ("web-1-aaaaaa", True),  # held, oldest first: not in line
        ("web-2-bbbbbb", True),
    ]
    assert queue[0].mark == "1 of 2"
    assert "no answer on port 22" in queue[2].mark
    [block] = found.blocks
    assert block["host"] == "root@build-web.test" and block["cause"] == "unreachable"
    html = page(App([dev], {"dev": LINKS}))
    assert "Blocked build VMs" in html and "held: its build VM is blocked" in html
    # Without its config the page can't tell which VM a project builds on:
    # nothing is held, and the block is still listed.
    bare = group(view(dev, Links()), "ready").rows
    assert not any(r.held for r in bare)
    assert view(dev, Links()).blocks


def test_dev_links_go_to_the_issue_the_pr_and_the_thread(dev: Source) -> None:
    rows = {r.item.id: r for g in view(dev).groups for r in g.rows}
    first = {l.text: l.href for l in rows["web-1-aaaaaa"].links}
    assert first == {
        "#1": "https://gitea.test/dev/web/issues/1",
        "thread": "https://matrix.to/#/!room:matrix.test/$t1",
    }
    merge = {l.text: l.href for l in rows["web-4-dddddd"].links}
    assert merge["PR #9"] == "https://gitea.test/dev/web/pulls/9"
    # A project the config doesn't name: the issue shows, unlinked.
    assert [(l.text, l.href) for l in rows["web-3-cccccc"].links] == [("#3", "")]


def test_with_no_config_there_are_no_links_and_no_errors(dev: Source) -> None:
    rows = [r for g in view(dev, Links()).groups for r in g.rows]
    hrefs = [l.href for r in rows for l in r.links]
    # Only the PR's own address, as Gitea gave it, links.
    assert set(hrefs) == {"", "https://gitea.test/dev/web/pulls/9"}
    html = page(App([dev]))
    assert "Item web-1-aaaaaa" in html and 'href="https://matrix.to' not in html


# --- nisse-review -----------------------------------------------------------------


@pytest.fixture
def reviewer(tmp_path: Path) -> Source:
    store = new_store(tmp_path / "rv", ReviewStore)
    for number, position, pr in (
        (1, review.READY, review.PR("dev/web", 1)),
        (2, review.REVIEWED, review.PR("dev/web", 2, rounds=2, happy=True)),
        (3, review.FAILED, review.PR("dev/web", 3, attempts=3, head="abc123def456")),
        (4, review.CAPPED, review.PR("dev/web", 4, rounds=6)),
        (5, review.WAITING, review.PR("dev/web", 5)),
        (6, review.STOPPED, review.PR("dev/web", 6)),
    ):
        store.open_work_item(
            review.item_id("dev/web", number),
            title=f"PR {number}",
            position=position,
            data=review.data(pr),
            cause="t",
        )
    store.close()
    return Source("review", tmp_path / "rv" / "state.db", "agent")


def test_review_prs_are_grouped_with_rounds_and_attempts(reviewer: Source) -> None:
    found = view(reviewer, Links(gitea="https://gitea.test"))
    assert found.kind == overview.REVIEW
    by_key = {g.key: [r.item.id for r in g.rows] for g in found.groups}
    assert by_key == {
        "running": [],
        "ready": ["dev/web#1"],
        "waiting": ["dev/web#5"],
        "reviewed": ["dev/web#2"],
        "capped": ["dev/web#4"],
        "failed": ["dev/web#3"],
        "stopped": ["dev/web#6"],
    }
    rows = {r.item.id: r for g in found.groups for r in g.rows}
    assert rows["dev/web#2"].details == ("2 rounds", "signed off")
    assert "3 failed runs at this head" in rows["dev/web#3"].details
    assert [(l.text, l.href) for l in rows["dev/web#1"].links] == [
        ("dev/web#1", "https://gitea.test/dev/web/pulls/1")
    ]


# --- what's running now -----------------------------------------------------------


def test_an_open_run_shows_as_running_and_a_stale_one_as_interrupted(
    tmp_path: Path,
) -> None:
    source, store = recorded(tmp_path / "a")  # its runs all ended
    recorder = calls.CallRecorder(source.calls_db)
    # A run under way: begun, with no end recorded.
    recorder.run_started("open-run", None, "w1", "driver", "work")
    recorder.close()
    store.close()
    now = datetime.now(UTC)
    found = App([source])._agent(source, now)
    [running] = found.running
    assert running.run["work_item"] == "w1" and running.turns == 0
    assert not found.interrupted
    html = page(App([source], clock=lambda: now))
    assert "running" in html and "Idle" not in html
    later = App([source])._agent(source, now + timedelta(minutes=16))
    assert not later.running and later.interrupted[0].run["work_item"] == "w1"
    assert "maybe interrupted" in page(
        App([source], clock=lambda: now + timedelta(minutes=16))
    )


def test_with_no_run_open_the_agent_is_idle_since_its_last_activity(
    tmp_path: Path,
) -> None:
    source, store = recorded(tmp_path / "a")
    store.close()
    found = App([source])._agent(source, datetime.now(UTC))
    assert not found.running and not found.interrupted
    assert found.last_activity is not None
    assert "Idle since" in page(App([source]))


def test_an_agent_without_a_call_record_still_shows_its_items(dev: Source) -> None:
    found = view(dev)
    assert not found.running and "no call record" in found.record_note
    assert found.last_activity is not None  # from its items' changes


# --- the page ---------------------------------------------------------------------


def test_the_overview_reloads_itself_unless_asked_not_to(dev: Source) -> None:
    app = App([dev])
    assert '<meta http-equiv="refresh" content="60">' in page(app, "/")
    assert 'http-equiv="refresh"' not in page(app, "/?refresh=0")


def test_kept_runs_are_listed_below_the_agents(tmp_path: Path, dev: Source) -> None:
    _, store = recorded(tmp_path / "kept" / "fix-bug-1a2b")
    store.close()
    app = App(
        [
            dev,
            Source(
                "kept.fix-bug-1a2b",
                tmp_path / "kept" / "fix-bug-1a2b" / "state.db",
                "run",
            ),
        ]
    )
    html = page(app)
    assert html.index("Item web-1-aaaaaa") < html.index("Recorded runs")
    assert "/s/kept.fix-bug-1a2b/" in html


def test_unknown_positions_and_fields_still_show(tmp_path: Path) -> None:
    store = new_store(tmp_path / "x", DevStore)
    store.open_work_item(
        "odd",
        title="Odd item",
        position="pondering",
        data={
            "loop": {
                "issue": "not a number",
                "wait": 3,
                "blocked_by": "dev/x#1",
                "new_field": [1],
            }
        },
        cause="t",
    )
    store.open_work_item("bare", title="No data", position="ready", cause="t")
    store.close()
    source = Source("x", tmp_path / "x" / "state.db", "agent")
    found = view(source)
    [other] = [g for g in found.groups if g.key == "other"]
    assert [(r.item.id, r.what) for r in other.rows] == [("odd", "pondering")]
    assert "No data" in page(App([source]))


def test_an_unreadable_store_shows_as_a_problem_not_a_broken_page(
    tmp_path: Path, dev: Source
) -> None:
    bad = tmp_path / "bad.db"
    with sqlite3.connect(bad) as conn:
        conn.execute("CREATE TABLE meta (key TEXT, value TEXT)")
        conn.execute("INSERT INTO meta VALUES ('schema_version', '99')")
    html = page(App([Source("bad", bad, "agent"), dev]))
    assert (
        "Couldn't read its store: state store schema 99" in html
        and "Item web-1-aaaaaa" in html
    )


def test_stored_text_is_escaped_and_never_becomes_a_link(tmp_path: Path) -> None:
    store = new_store(tmp_path / "h", DevStore)
    dev_item(
        store,
        "h1",
        "waiting",
        Loop(
            wait=HOSTILE,
            note=HOSTILE,
            pr=3,
            pr_url="javascript:alert(1)",
            thread='"><script>x</script>',
        ),
    )
    store.close()
    source = Source("h", tmp_path / "h" / "state.db", "agent")
    html = page(App([source], {"h": LINKS}))
    assert "<script>" not in html and "javascript:" not in html
    assert "&lt;script&gt;" in html


def test_browsing_the_overview_changes_no_file(tmp_path: Path, dev: Source) -> None:
    source, store = recorded(tmp_path / "a")
    store.close()

    def files() -> dict[str, str]:
        return {
            str(p): hashlib.sha256(p.read_bytes()).hexdigest()
            for p in sorted(tmp_path.rglob("*"))
            if p.is_file()
        }

    before = files()
    app = App([source, dev], {"dev": LINKS})
    page(app, "/")
    page(app, "/?refresh=0")
    assert files() == before


# --- the config -------------------------------------------------------------------

CONFIG = """
state_db: /var/lib/nisse-dev/state.db
llm:
  base_url: http://litellm.test:4000
  token: sk-planted-secret-value
matrix:
  homeserver: https://matrix.test
  token: ${NISSE_MATRIX_TOKEN}
  room_id: "!room:matrix.test"
gitea:
  url: https://gitea.test/
  token: gitea-planted-secret
projects:
  web:
    repo: dev/web
    workspace: /root/work/web
    build_host:
      address: build-web.test
      key: ~/.ssh/nisse-dev
    push_token: push-planted-secret
  lib:
    repo: ${LIB_REPO}
    build_host: {address: build-lib.test, user: builder, key: k}
"""


def test_links_come_from_four_settings_and_nothing_else(tmp_path: Path) -> None:
    path = tmp_path / "nisse-dev.yaml"
    path.write_text(CONFIG)
    found = links.load(path)
    assert found == Links(
        gitea="https://gitea.test",
        room="!room:matrix.test",
        repos={"web": "dev/web"},  # a ${VAR} isn't resolved: unset
        hosts={"web": "root@build-web.test", "lib": "builder@build-lib.test"},
    )
    held = repr(found)
    for secret in ("planted", "sk-", "NISSE_MATRIX_TOKEN", "litellm"):
        assert secret not in held


def test_a_gitea_address_that_isnt_http_gives_no_links(tmp_path: Path) -> None:
    path = tmp_path / "c.yaml"
    path.write_text("gitea: {url: 'javascript:alert(1)'}\n")
    found = links.load(path)
    assert found.gitea == "" and found.pull("dev/web", 1) == ""
    assert links.web_address("HTTPS://x.test") == "HTTPS://x.test"
    assert links.web_address(" data:text/html,x") == ""


def test_the_command_line_takes_a_config_for_a_named_source(
    tmp_path: Path, dev: Source, monkeypatch: pytest.MonkeyPatch
) -> None:
    path = tmp_path / "nisse-dev.yaml"
    path.write_text(CONFIG)
    built: dict[str, App] = {}

    class Bound(Exception):
        pass

    def server(address: tuple[str, int], handler: object) -> None:
        raise Bound

    real_app = ui_main.App

    def app(*args: object, **kwargs: object) -> App:
        built["app"] = real_app(*args, **kwargs)  # type: ignore[arg-type]
        return built["app"]

    monkeypatch.setattr(ui_main, "ThreadingHTTPServer", server)
    monkeypatch.setattr(ui_main, "App", app)
    with pytest.raises(Bound):
        ui_main.main(["--source", f"dev={dev.state_db}", "--config", f"dev={path}"])
    assert built["app"].links["dev"].repos == {"web": "dev/web"}
    for bad in (["--config", f"nope={path}"], ["--config", f"dev={tmp_path}/none"]):
        with pytest.raises(SystemExit):
            ui_main.main(["--source", f"dev={dev.state_db}", *bad])