"""
Which folder (possibly a subfolder of a shared root) is the Music app's
entry point for a group. Same shape as test_video_root_policy.py — a
signed operator instruction, per-group, stored via roster.py's
group_settings table, added later once a real messy library showed
musicbay.md's original "no root, whole shared tree" call was wrong.
"""

import base64
from pathlib import Path

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

from meshbay_common.adminop import OP_AUDIO_ROOT, admin_transcript
from meshbay_common.crypto import pk_to_b64
from meshbay_common.join import ROLE_OPERATOR
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.roster import Roster, open_roster
from meshbay_node.transport.webrtc_server import WebRTCPeerSession

from conftest import one_root

pytestmark = pytest.mark.asyncio


def _session(tmp_path: Path, user_id: str, *, operator: str | None = None) -> WebRTCPeerSession:
    shared_root = tmp_path / "shared"
    shared_root.mkdir(exist_ok=True)
    (shared_root / "Music").mkdir()
    (shared_root / "Podcasts").mkdir()
    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
    ctx = {
        "roots": one_root(shared_root),
        "index": index,
        "sk_node": index.sk_node,
        "node_user_id": operator,
    }
    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
    session._ctx = ctx
    session._group_id = None
    session._user_id = user_id
    session._pk_user = ""
    session.sent = []
    session._send = session.sent.append
    session._audit = lambda *a, **k: None
    return session


# ── Refused before a challenge is even issued ───────────────────────────────

async def test_missing_path_is_refused(tmp_path):
    session = _session(tmp_path, "op", operator="op")
    session._has_admin_authority = lambda: True
    issued = []
    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))

    session._do_audio_root({})

    assert not issued
    assert [m for m in session.sent if m.get("type") == "error"]


async def test_a_nonexistent_folder_is_refused(tmp_path):
    session = _session(tmp_path, "op", operator="op")
    session._has_admin_authority = lambda: True
    issued = []
    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))

    session._do_audio_root({"path": "shared/Nonexistent"})

    assert not issued, "a mistyped path must be refused before a signature round trip"
    assert [m for m in session.sent if m.get("type") == "error"]


async def test_path_traversal_is_refused(tmp_path):
    session = _session(tmp_path, "op", operator="op")
    session._has_admin_authority = lambda: True
    issued = []
    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))

    session._do_audio_root({"path": "../../etc"})

    assert not issued
    assert [m for m in session.sent if m.get("type") == "error"]


async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path):
    session = _session(tmp_path, "member-1", operator="the-operator")
    session._has_admin_authority = lambda: False

    session._do_audio_root({"path": "shared/Music"})

    assert [m for m in session.sent if m.get("type") == "error"]


# ── Accepted cases ───────────────────────────────────────────────────────────

async def test_an_empty_path_is_always_accepted(tmp_path):
    """Empty means 'unset' — Music shows nothing yet, always valid to clear."""
    session = _session(tmp_path, "op", operator="op")
    session._has_admin_authority = lambda: True
    issued = []
    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))

    session._do_audio_root({"path": ""})

    assert issued == [(OP_AUDIO_ROOT, "")]


async def test_a_real_subfolder_is_accepted_and_signed(tmp_path):
    session = _session(tmp_path, "op", operator="op")
    session._has_admin_authority = lambda: True
    issued = []
    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))

    session._do_audio_root({"path": "shared/Music"})

    assert issued == [(OP_AUDIO_ROOT, "shared/Music")]


# ── Where it is stored ──────────────────────────────────────────────────────

async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path):
    roster = Roster(db_path=tmp_path / "roster.db")
    await roster.open()
    try:
        assert await roster.audio_root("g1") == "", "absent must mean unset"
        await roster.set_audio_root("g1", "shared/Music", set_by="op")
        assert await roster.audio_root("g1") == "shared/Music"
    finally:
        await roster.close()

    reopened = Roster(db_path=tmp_path / "roster.db")
    await reopened.open()
    try:
        assert await reopened.audio_root("g1") == "shared/Music"
        assert await reopened.audio_root("g2") == "", "one group's setting must not answer for another"
    finally:
        await reopened.close()


# ── End to end through the real signed-op path ──────────────────────────────
#
# Everything above either calls ops.set_audio_root directly or mocks out
# _issue_admin_challenge — neither one ever exercises real signature
# verification (_verify_admin_sig, _do_admin_response) or the shared
# groups_ctx/roster wiring _run_op depends on. Found live: a save that
# looked like it worked (the Music tab showed content right afterward) did
# not survive a reload — worth ruling out a break somewhere in that real
# path specifically, not just in the pure-Python setter. Session shape
# mirrors test_admin_ops_mnp.py's _session helper.

GROUP = "g" * 32


def _keypair():
    sk = Ed25519PrivateKey.generate()
    return sk, pk_to_b64(sk.public_key())


async def _full_session(tmp_path: Path, roster) -> tuple[WebRTCPeerSession, Ed25519PrivateKey]:
    shared = tmp_path / "shared"
    (shared / "Music").mkdir(parents=True)
    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
    roots = one_root(shared)

    sk_op, pk_op = _keypair()
    await roster.pin_identity("grenet", "grenet", pk_op, pk_op, "code")
    await roster.set_member("", "grenet", ROLE_OPERATOR, "active", "local-cli")

    group_ctx = {"gek": b"\x01" * 32, "roots": roots, "index": index,
                 "join_policy": "invite", "audio_root": ""}
    state = {
        "groups_ctx": {GROUP: group_ctx},
        "roster": roster,
        "node_user_id": "node-user",
    }

    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
    session._ctx = {
        "roots": roots, "index": index, "sk_node": index.sk_node,
        "roster": roster, "groups": {GROUP: group_ctx},
        "has_admin_authority": True,
        "daemon_state": state,
    }
    session._group_id = GROUP
    session._user_id = "grenet"
    session._pk_user = ""
    session._admin_ops = {}
    session.sent = []
    session._send = session.sent.append
    session._audit = lambda *a, **k: None
    session.spawned = []
    session._spawn = session.spawned.append
    # A real session registers itself here on handshake completion
    # (`self._peer_registry()[self._user_id] = self`) — without it, the
    # broadcast loop in _admin_exec_audio_root (and every other admin op)
    # has nobody to send the final ack to, including the requester itself.
    group_ctx["_peers"] = {"grenet": session}
    session._peer_registry = lambda: group_ctx["_peers"]
    return session, sk_op


async def _drain(session):
    for coro in session.spawned:
        await coro
    session.spawned.clear()


async def test_a_real_signed_save_persists_and_survives_a_fresh_roster_read(tmp_path):
    """
    The exact question a "worked, then reverted after reload" report raises:
    does the value set through the real challenge/response path actually
    land in the database, in a form any later connection — this one, or a
    freshly-opened Roster after a restart — reads back correctly?
    """
    roster = await open_roster(tmp_path)
    try:
        session, sk_op = await _full_session(tmp_path, roster)

        session._do_audio_root({"path": "shared/Music"})
        challenge = session.sent[-1]
        assert challenge["type"] == "admin_challenge", challenge

        transcript = admin_transcript(
            op=OP_AUDIO_ROOT, node_pk_b64=session._node_pk_b64(), group_id=GROUP,
            subject="shared/Music", nonce=base64.b64decode(challenge["nonce"]),
            ts=challenge["ts"])
        session._do_admin_response({
            "op_id": challenge["op_id"],
            "signature": base64.b64encode(sk_op.sign(transcript)).decode(),
        })
        await _drain(session)

        ack = session.sent[-1]
        assert ack["type"] == "audio_root_ack", ack
        assert ack["path"] == "shared/Music"

        assert session._ctx["groups"][GROUP]["audio_root"] == "shared/Music", (
            "the live in-memory context must reflect the new root immediately")
        assert await roster.audio_root(GROUP) == "shared/Music", (
            "the same Roster instance must read back what it just wrote")
    finally:
        await roster.close()

    # A fresh connection (or a restarted daemon) never touches the Roster
    # instance above at all — it opens its own. This is the check that
    # actually answers "does it survive a reload".
    reopened = await open_roster(tmp_path)
    try:
        assert await reopened.audio_root(GROUP) == "shared/Music", (
            "a freshly-opened Roster against the same db file must see the "
            "committed value — anything else means the write was never "
            "durable in the first place")
    finally:
        await reopened.close()