"""
Pointing an application at folders, through the real signed-op path.

`app_directories` is one operator instruction for every application, keyed by
the app's own name. What only this file can check is the path from the message
to the database: everything else either calls `ops.set_app_directories`
directly or mocks out `_issue_admin_challenge`, and neither one exercises real
signature verification (`_verify_admin_sig`, `_do_admin_response`) or the shared
groups_ctx/roster wiring `_run_op` depends on.

Found live, on the per-app op this replaced: a save that looked like it worked —
the Music tab showed content right afterwards — did not survive a reload. Worth
ruling out a break in that real path specifically, and not just in the setter.

Note what is deliberately *not* checked before the challenge: whether the path
exists. `_do_app_directories` validates the app name and the shape of
`directories`, then asks for a signature; `ops._validate_app_dirs` refuses a
path outside the group's roots afterwards. A settings change is not a
capability, so refusing after the signature costs a round trip and nothing else.
"""

import base64
from pathlib import Path

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.adminop import OP_APP_DIRECTORIES, 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 open_roster
from meshbay_node.transport.webrtc_server import WebRTCPeerSession

from conftest import one_root

pytestmark = pytest.mark.asyncio

# 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", "music_directories": []}
    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_app_directories(
            {"app": "music", "directories": ["shared/Music"]})
        challenge = session.sent[-1]
        assert challenge["type"] == "admin_challenge", challenge

        transcript = admin_transcript(
            op=OP_APP_DIRECTORIES, node_pk_b64=session._node_pk_b64(),
            group_id=GROUP, subject="music: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"] == "app_directories_ack", ack
        assert ack["app"] == "music"
        assert ack["directories"] == ["shared/Music"]

        assert (session._ctx["groups"][GROUP]["music_directories"]
                == ["shared/Music"]), (
            "the live in-memory context must reflect the new folder at once")
        assert await roster.app_directories(GROUP, "music") == ["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.app_directories(GROUP, "music") == ["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()