"""hive — the swarm address book service (see hive/app.py).

The app module reads env config at import time, so each test reimports it with
fresh env into its own key file / database.
"""

import importlib
import sqlite3

import pytest
from fastapi.testclient import TestClient


@pytest.fixture()
def hive(tmp_path, monkeypatch):
    monkeypatch.setenv("HIVE_KEY_FILE", str(tmp_path / ".swarm-key"))
    monkeypatch.setenv("HIVE_DB", str(tmp_path / "hive.db"))
    monkeypatch.setenv("HIVE_TTL_SEC", "300")
    # Deterministic admin key: no ambient .hive-admin-key from the checkout.
    monkeypatch.setenv("HIVE_ADMIN_KEY", "admin-secret")
    monkeypatch.setenv("HIVE_ADMIN_KEY_FILE", str(tmp_path / ".hive-admin-key"))
    (tmp_path / ".swarm-key").write_text("secret-key-123")
    import hive.app as hive_app

    module = importlib.reload(hive_app)
    yield module
    module._db.close()


def _headers(key: str = "secret-key-123") -> dict:
    return {"X-Swarm-Key": key}


def _announce(client, name="quiet-otter", instance_id="11111111-2222-3333-4444-555555555555", port=8099):
    return client.post(
        "/announce",
        json={"name": name, "instance_id": instance_id, "version": "0.1.0", "port": port,
              "meta": {"hostname": "gpu-box", "os": "Linux 6.1", "cpu_cores": 16, "ram_mb": 65536}},
        headers=_headers(),
    )


def test_health_open_without_key(hive):
    with TestClient(hive.app) as client:
        resp = client.get("/health")
        assert resp.status_code == 200
        assert resp.json()["hive"] is True


def test_announce_requires_key(hive):
    with TestClient(hive.app) as client:
        assert client.post("/announce", json={"name": "x", "instance_id": "12345678", "port": 8099}).status_code == 401
        assert client.post(
            "/announce",
            json={"name": "x", "instance_id": "12345678", "port": 8099},
            headers={"X-Swarm-Key": "wrong"},
        ).status_code == 403


def test_peers_requires_key(hive):
    with TestClient(hive.app) as client:
        assert client.get("/peers").status_code == 401
        assert client.get("/peers", headers={"X-Swarm-Key": "wrong"}).status_code == 403


def test_announce_upsert_and_peers_listing(hive):
    with TestClient(hive.app) as client:
        assert _announce(client).status_code == 200
        # same instance_id again = upsert (renamed, different port), not a second row
        assert _announce(client, name="renamed-otter", port=8123).status_code == 200
        # a second machine
        assert _announce(client, name="amber-falcon", instance_id="99999999-2222-3333-4444-555555555555").status_code == 200

        resp = client.get("/peers", headers=_headers())
        assert resp.status_code == 200
        body = resp.json()
        peers = body["peers"]
        assert len(peers) == 2
        assert [p["name"] for p in peers] == ["amber-falcon", "renamed-otter"]  # sorted by name
        otter = peers[1]
        assert otter["port"] == 8123
        assert otter["address"] == f"{otter['host']}:8123"
        assert otter["online"] is True
        assert otter["meta"]["hostname"] == "gpu-box"
        assert otter["meta"]["ram_mb"] == 65536
        assert otter["version"] == "0.1.0"
        assert body["ttl_sec"] == 300


def test_offline_after_ttl(hive, tmp_path):
    with TestClient(hive.app) as client:
        _announce(client)
        # backdate the last announcement beyond the TTL — machine went quiet
        conn = sqlite3.connect(tmp_path / "hive.db")
        conn.execute("UPDATE machines SET last_seen = last_seen - 400")
        conn.commit()
        conn.close()
        peers = client.get("/peers", headers=_headers()).json()["peers"]
        assert peers[0]["online"] is False


def test_previous_key_still_accepted(hive, tmp_path):
    # rotation window: .swarm-key.previous keeps working
    (tmp_path / ".swarm-key.previous").write_text("old-key-42")
    import hive.app as module

    importlib.reload(module)
    with TestClient(module.app) as client:
        resp = client.post(
            "/announce",
            json={"name": "x", "instance_id": "12345678", "port": 8099},
            headers={"X-Swarm-Key": "old-key-42"},
        )
        assert resp.status_code == 200
    module._db.close()


def test_hive_without_key_file_refuses(hive, tmp_path, monkeypatch):
    monkeypatch.setenv("HIVE_KEY_FILE", str(tmp_path / ".no-key"))
    monkeypatch.setenv("HIVE_DB", str(tmp_path / "hive2.db"))
    import hive.app as module

    importlib.reload(module)
    with TestClient(module.app) as client:
        resp = client.post(
            "/announce",
            json={"name": "x", "instance_id": "12345678", "port": 8099},
            headers={"X-Swarm-Key": "whatever"},
        )
        assert resp.status_code == 503
    module._db.close()

# ── config documents ────────────────────────────────────────────────────────


def _admin(key: str = "admin-secret") -> dict:
    return {"X-Admin-Key": key}


def _publish(client, layers, note="", force=False, key="admin-secret"):
    return client.put(
        "/config",
        json={"layers": layers, "note": note, "force": force},
        headers=_admin(key),
    )


def _global(**settings):
    return {"name": "global", "config": {"settings": settings}}


def test_publish_requires_admin_key(hive):
    with TestClient(hive.app) as client:
        resp = client.put("/config", json={"layers": [_global(llm_complete_timeout=30)]})
        assert resp.status_code == 403
        bad = client.put(
            "/config",
            json={"layers": [_global(llm_complete_timeout=30)]},
            headers=_admin("wrong"),
        )
        assert bad.status_code == 403


def test_fetch_returns_404_before_anything_is_published(hive):
    with TestClient(hive.app) as client:
        assert client.get("/config", headers=_headers()).status_code == 404


def test_publish_then_fetch_global_layer(hive):
    with TestClient(hive.app) as client:
        published = _publish(client, [_global(llm_complete_timeout=30)]).json()
        assert published["status"] == "published"
        assert published["version"] == 1

        body = client.get("/config", headers=_headers()).json()
        assert body["version"] == 1
        assert body["layers"] == [
            {"name": "global", "config": {"settings": {"llm_complete_timeout": 30}}}
        ]


def test_layers_target_by_tag_and_name(hive):
    layers = [
        {"name": "global", "config": {"settings": {"llm_complete_timeout": 30}}},
        {"name": "lab", "match": {"tags": ["lab"]}, "config": {"settings": {"llm_complete_retries": 1}}},
        {"name": "yuki", "match": {"names": ["yuki"]}, "config": {"settings": {"navi_push_cooldown_sec": 5}}},
    ]
    with TestClient(hive.app) as client:
        _publish(client, layers)

        plain = client.get("/config", params={"name": "other"}, headers=_headers()).json()
        assert [layer["name"] for layer in plain["layers"]] == ["global"]

        lab = client.get(
            "/config", params={"name": "other", "tags": "beta,lab"}, headers=_headers()
        ).json()
        assert [layer["name"] for layer in lab["layers"]] == ["global", "lab"]

        named = client.get("/config", params={"name": "yuki"}, headers=_headers()).json()
        assert [layer["name"] for layer in named["layers"]] == ["global", "yuki"]

        # The hash is over the layers that instance receives, not the document.
        assert plain["hash"] != named["hash"]


def test_identical_publish_is_a_no_op(hive):
    with TestClient(hive.app) as client:
        _publish(client, [_global(llm_complete_timeout=30)])
        again = _publish(client, [_global(llm_complete_timeout=30)]).json()
        assert again["status"] == "unchanged"
        assert again["version"] == 1

        forced = _publish(client, [_global(llm_complete_timeout=30)], force=True).json()
        assert forced["status"] == "renotified"
        assert forced["version"] == 1
        assert len(client.get("/config/history", headers=_admin()).json()["versions"]) == 1


def test_empty_document_clears_the_overlay(hive):
    with TestClient(hive.app) as client:
        _publish(client, [_global(llm_complete_timeout=30)])
        cleared = _publish(client, [], note="back to local").json()
        assert cleared["status"] == "published"
        assert cleared["version"] == 2

        body = client.get("/config", headers=_headers()).json()
        assert body["layers"] == []


def test_invalid_document_is_rejected_with_a_reason(hive):
    with TestClient(hive.app) as client:
        resp = _publish(client, [{"name": "x"}])
        assert resp.status_code == 400
        assert "config" in resp.json()["detail"]

        resp = _publish(client, [{"name": "x", "config": {}, "match": {"nope": ["a"]}}])
        assert resp.status_code == 400


def test_announce_hint_tells_a_behind_instance_to_fetch(hive):
    with TestClient(hive.app) as client:
        _publish(client, [_global(llm_complete_timeout=30)])

        # An older build that reports no config: unknown, not behind.
        old = _announce(client).json()
        assert old["config"] is None

        # A current build with an empty hash is behind.
        behind = client.post(
            "/announce",
            json={
                "name": "quiet-otter",
                "instance_id": "11111111-2222-3333-4444-555555555555",
                "port": 8099,
                "meta": {"config": {"hash": "", "status": "applied"}},
            },
            headers=_headers(),
        ).json()
        assert behind["config"]["fetch"] is True
        assert behind["config"]["version"] == 1

        # One that reports the exact effective hash is up to date.
        effective = client.get("/config", headers=_headers()).json()["hash"]
        current = client.post(
            "/announce",
            json={
                "name": "quiet-otter",
                "instance_id": "11111111-2222-3333-4444-555555555555",
                "port": 8099,
                "meta": {"config": {"hash": effective, "status": "applied"}},
            },
            headers=_headers(),
        ).json()
        assert current["config"] is None


def test_ack_is_recorded_and_shown_in_status(hive):
    with TestClient(hive.app) as client:
        _publish(client, [_global(llm_complete_timeout=30)])
        _announce(client)
        resp = client.post(
            "/config/ack",
            json={
                "instance_id": "11111111-2222-3333-4444-555555555555",
                "version": 1,
                "hash": "sha256:x",
                "status": "staged",
                "restart_required": ["openai_api_key"],
            },
            headers=_headers(),
        )
        assert resp.status_code == 200

        status = client.get("/config/status", headers=_admin()).json()
        row = status["instances"][0]
        assert row["ack"]["status"] == "staged"
        assert row["ack"]["restart_required"] == ["openai_api_key"]


def test_status_marks_behind_instances(hive):
    with TestClient(hive.app) as client:
        _publish(client, [_global(llm_complete_timeout=30)])
        effective = client.get("/config", headers=_headers()).json()["hash"]
        client.post(
            "/announce",
            json={
                "name": "quiet-otter",
                "instance_id": "11111111-2222-3333-4444-555555555555",
                "port": 8099,
                "meta": {"config": {"hash": "sha256:old", "status": "applied"}},
            },
            headers=_headers(),
        )
        status = client.get("/config/status", headers=_admin()).json()
        assert status["behind"] == ["quiet-otter"]
        assert status["instances"][0]["effective_hash"] == effective

        client.post(
            "/announce",
            json={
                "name": "quiet-otter",
                "instance_id": "11111111-2222-3333-4444-555555555555",
                "port": 8099,
                "meta": {"config": {"hash": effective, "status": "applied"}},
            },
            headers=_headers(),
        )
        assert client.get("/config/status", headers=_admin()).json()["behind"] == []


def test_history_version_and_rollback(hive):
    with TestClient(hive.app) as client:
        _publish(client, [_global(llm_complete_timeout=30)], note="first")
        _publish(client, [_global(llm_complete_timeout=45)], note="second")

        history = client.get("/config/history", headers=_admin()).json()["versions"]
        assert [v["id"] for v in history] == [2, 1]
        assert history[1]["note"] == "first"

        version = client.get("/config/version/1", headers=_admin()).json()
        assert version["layers"][0]["config"]["settings"]["llm_complete_timeout"] == 30

        rolled = client.post("/config/rollback/1", headers=_admin()).json()
        assert rolled["status"] == "rolled_back"
        assert rolled["version"] == 3
        assert client.get("/config", headers=_headers()).json()["layers"][0]["config"] == {
            "settings": {"llm_complete_timeout": 30}
        }
        # History keeps the bad version rather than rewriting it.
        assert len(client.get("/config/history", headers=_admin()).json()["versions"]) == 3


async def test_notify_all_pings_every_registered_instance(hive, monkeypatch):
    with TestClient(hive.app) as client:
        _announce(client, name="one", instance_id="aaaaaaaa-0000-0000-0000-000000000001")
        _announce(client, name="two", instance_id="aaaaaaaa-0000-0000-0000-000000000002")
        calls = []

        async def fake(peer, version, doc_hash):
            calls.append((peer["name"], version, doc_hash))
            return True

        monkeypatch.setattr(hive, "_notify_one", fake)
        await hive._notify_all(4, "sha256:x")
        assert sorted(name for name, _, _ in calls) == ["one", "two"]


async def test_retry_sweep_targets_only_the_instance_that_is_behind(hive, monkeypatch):
    with TestClient(hive.app) as client:
        _announce(client, name="one", instance_id="aaaaaaaa-0000-0000-0000-000000000001")
        _announce(client, name="two", instance_id="aaaaaaaa-0000-0000-0000-000000000002")
        _publish(client, [_global(llm_complete_timeout=30)])
        effective = client.get("/config", headers=_headers()).json()["hash"]

        client.post(
            "/announce",
            json={
                "name": "one",
                "instance_id": "aaaaaaaa-0000-0000-0000-000000000001",
                "port": 8099,
                "meta": {"config": {"hash": effective, "status": "applied"}},
            },
            headers=_headers(),
        )
        client.post(
            "/announce",
            json={
                "name": "two",
                "instance_id": "aaaaaaaa-0000-0000-0000-000000000002",
                "port": 8099,
                "meta": {"config": {"hash": "sha256:old", "status": "applied"}},
            },
            headers=_headers(),
        )

        calls = []

        async def fake(peer, version, doc_hash):
            calls.append(peer["name"])
            return True

        monkeypatch.setattr(hive, "_notify_one", fake)
        await hive._notify_behind()
        assert calls == ["two"]


def test_a_fresh_ack_beats_a_stale_announce(hive):
    """The announce may lag an interval behind; an ack for the current hash is proof."""
    iid = "11111111-2222-3333-4444-555555555555"
    with TestClient(hive.app) as client:
        _publish(client, [_global(llm_complete_timeout=30)])
        effective = client.get("/config", headers=_headers()).json()["hash"]

        client.post(
            "/announce",
            json={
                "name": "quiet-otter",
                "instance_id": iid,
                "port": 8099,
                "meta": {"config": {"hash": "sha256:old", "status": "applied"}},
            },
            headers=_headers(),
        )
        assert client.get("/config/status", headers=_admin()).json()["behind"] == [
            "quiet-otter"
        ]

        client.post(
            "/config/ack",
            json={"instance_id": iid, "version": 1, "hash": effective, "status": "applied"},
            headers=_headers(),
        )
        assert client.get("/config/status", headers=_admin()).json()["behind"] == []