"""
MeshBay Node daemon — main process.

Startup sequence:
  1. Load config (~/.config/meshbay/node.toml)
  2. Load or create keystore (Argon2id unlock)
  3. Connect to hub: register → login → announce node
  4. Fetch GEK bundle from hub (if group configured)
  5. Start directory indexer (watchdog)
  6. Create chat stores (one SQLite DB per group)
  7. Create WebRTC transport (browser + native clients via DataChannel)
  8. Start QUIC chunk server (LAN / port-forwarded / hub-less direct access)
  9. (Phase 11.5: the unauthenticated HTTP file API and the TCP+TLS server were removed)
  10. Start hub WebSocket (signaling, revocations, WebRTC offers)
  11. Start local web UI on node.ui_port (localhost only)
  12. Run until SIGINT/SIGTERM

Usage:
  meshbay-node                    # interactive password prompt
  meshbay-node --config /path     # custom config
  meshbay-node status             # node state + public key (works while stopped)
  meshbay-node ui                 # print the local admin UI URL
  meshbay-node gek-init           # initialise the group key (no browser needed)
  meshbay-node init               # write example config + create keystore
  meshbay-node --calibrate-argon2 # benchmark Argon2id, suggest parameters
"""

import asyncio
from dataclasses import asdict
import base64
import json
import logging
import os
import signal
import sys
from pathlib import Path

import uvicorn

from meshbay_common import MNP_VERSION
from meshbay_common.protocol import MNP
from meshbay_node.audit import AuditStore
from meshbay_node.bundle_store import BundleStore
from meshbay_node.chat.store import ChatStore
from meshbay_node.config import Config, DEFAULT_CONFIG_PATH, load_config, write_example_config
from meshbay_node.roots import RootSet, RootError
from meshbay_node.hub_client import HubClient, HubConfig
from meshbay_node.indexer import DirectoryIndexer
from meshbay_node.keystore import load_or_create_keystore
from meshbay_node.roster import Roster
from meshbay_node.transport import (
    Denylist,
    QUIC_AVAILABLE,
    WEBRTC_AVAILABLE,
)

if QUIC_AVAILABLE:
    from meshbay_node.transport import QuicChunkServer
if WEBRTC_AVAILABLE:
    from meshbay_node.transport import WebRTCTransport

log = logging.getLogger(__name__)


# ── Argon2id calibration ──────────────────────────────────────────────────────

def calibrate_argon2(target_ms: int = 500) -> None:
    """Benchmark Argon2id and suggest parameters targeting ~target_ms."""
    import time
    import os

    print(f"Calibrating Argon2id (target: {target_ms}ms) ...")
    salt = os.urandom(16)
    for mem in [65536, 131072, 262144, 524288]:
        from cryptography.hazmat.primitives.kdf.argon2 import Argon2id
        t0 = time.perf_counter()
        Argon2id(salt=salt, length=32, iterations=3,
                 lanes=4, memory_cost=mem).derive(b"benchmark")
        elapsed_ms = (time.perf_counter() - t0) * 1000
        print(f"  memory_cost={mem:>7} ({mem//1024:>4}MB): {elapsed_ms:.0f}ms", end="")
        if abs(elapsed_ms - target_ms) < target_ms * 0.3:
            print("  ← recommended")
        else:
            print()
    print("Set memory_cost in meshbay_common/crypto.py: ARGON2_MEMORY_COST")


# ── Hub WS sender bridge ─────────────────────────────────────────────────────

class _WsSender:
    """Thin bridge so WebRTC context can call hub_ws.send() for chat_notify."""

    def __init__(self, hub_client: HubClient):
        self._hub = hub_client

    async def send(self, data: str) -> None:
        await self._hub.send_ws(data)


# ── Daemon ────────────────────────────────────────────────────────────────────

class NodeDaemon:
    def __init__(self, config: Config, config_path: Path = DEFAULT_CONFIG_PATH):
        self._config = config
        self._config_path = config_path
        self._state: dict = {
            "status":        "starting",
            "hub_url":       config.hub.url,
            "username":      config.hub.username,
            "groups":        [g.name for g in config.groups],
            "quic_port":     config.node.quic_port,
            "endpoint_hint": None,
            "indexes":       {},
        }
        self._quic_server = None
        self._webrtc = None
        # Persisted so a restart does not silently un-revoke everyone (H4)
        self._denylist = (
            Denylist(path=config.data_dir / "denylist.json") if Denylist else None)
        self._chat_stores: dict[str, ChatStore] = {}
        self._audit_store: AuditStore | None = None
        self._bundle_store: BundleStore | None = None
        self._roster: Roster | None = None
        self._indexers: list[DirectoryIndexer] = []
        self._tasks: list[asyncio.Task] = []
        self._hub: HubClient | None = None

    async def run(self) -> None:
        log.info("MeshBay Node starting up")

        # 1. Keystore
        keys = load_or_create_keystore(
            path=self._config.keystore.path,
            unlock_file=self._config.keystore.unlock_file,
        )
        log.info("Keys loaded: %s", keys.pk_ed25519_b64[:16])

        # 2. Start admin UI early (so operator can copy node key before hub login)
        self._state["pk_node_ed25519"] = keys.pk_ed25519_b64
        self._state["config"] = self._config
        # Where it came from, so `group add` appends to the file this process
        # actually read rather than guessing at the default.
        self._state["config_path"] = str(self._config_path)
        # Per-run token for the local admin UI (11.5.3). Not a password: it keeps
        # other local processes and rebound browser pages out of an API that can
        # re-initialise group keys.
        ui_token = base64.urlsafe_b64encode(os.urandom(18)).decode().rstrip("=")
        self._state["ui_token"] = ui_token
        # Persisted so `meshbay-node ui` can open the browser. Nobody should ever
        # have to copy a token out of a log or a terminal — that is not a workflow.
        self._config.data_dir.mkdir(parents=True, exist_ok=True)
        self._ui_token_file = self._config.data_dir / "ui-token"
        self._ui_token_file.write_text(ui_token)
        self._ui_token_file.chmod(0o600)
        from meshbay_node.ui import create_ui_app
        ui_app = create_ui_app(self._state)
        ui_cfg = uvicorn.Config(
            ui_app,
            host="127.0.0.1",
            port=self._config.node.ui_port,
            log_level="warning",
        )
        ui_server = uvicorn.Server(ui_cfg)
        self._tasks.append(asyncio.create_task(ui_server.serve()))
        log.info("Admin UI ready — open it with:  meshbay-node ui")

        # 3. Hub connection (Ed25519 auth — retries until node key is linked)
        hub_cfg = HubConfig(
            hub_url=self._config.hub.url,
            username=self._config.hub.username,
        )
        async with HubClient(hub_cfg, keys) as hub:
            self._hub = hub
            session = await self._login_with_retry(hub)
            self._state["endpoint_hint"] = session.node_id

            # 4. Bundle store (P2P GEK bundles)
            data_dir = self._config.data_dir
            data_dir.mkdir(parents=True, exist_ok=True)
            self._bundle_store = BundleStore(db_path=data_dir / "bundles.db")
            await self._bundle_store.open()
            log.info("Bundle store opened: %s", data_dir / "bundles.db")

            # 4b. Roster — who this node recognises and which keys are theirs.
            # Node authority is established here, locally, and never learned from
            # the hub: a hub that could name the operator's key could install
            # itself as node administrator.
            self._roster = Roster(db_path=data_dir / "roster.db")
            await self._roster.open()
            await self._roster.purge_expired()

            # X25519 key material for GEK unwrapping
            from cryptography.hazmat.primitives import serialization
            sk_x_raw = keys.sk_x25519.private_bytes(
                serialization.Encoding.Raw, serialization.PrivateFormat.Raw,
                serialization.NoEncryption())
            pk_x_raw = base64.b64decode(keys.pk_x25519_b64)

            # 4. Build per-group contexts
            groups_ctx: dict[str, dict] = {}
            for group_cfg in self._config.groups:
                if not group_cfg.id or not group_cfg.roots:
                    log.warning("Group %r has no id or no shared directory — "
                                "skipping", group_cfg.name)
                    continue

                try:
                    roots = RootSet.build([asdict(r) for r in group_cfg.roots])
                except RootError as e:
                    # Configuration the operator has to fix; guessing would put
                    # a member's file on the wrong disk or index one twice.
                    log.error("Group %r: %s — skipping", group_cfg.name, e)
                    continue

                roots.refresh_availability()
                if not any(r.available for r in roots):
                    # Not skipped for being empty: a group whose only drive is
                    # unplugged still exists, and its index is frozen rather
                    # than lost. But there is nothing to serve until it returns.
                    log.warning(
                        "Group %r: none of its %d root(s) are readable right now "
                        "(%s) — serving nothing until one returns",
                        group_cfg.name, len(roots),
                        ", ".join(str(r.path) for r in roots))

                gek = None
                if group_cfg.visibility == "private":
                    gek = await self._load_gek(
                        group_cfg.id, session.user_id, sk_x_raw, pk_x_raw)
                    if gek:
                        log.info("GEK loaded for group %s", group_cfg.id[:8])
                    else:
                        log.info("No GEK yet for group %s — will accept first setup",
                                 group_cfg.name)

                indexer = DirectoryIndexer(
                    roots=roots,
                    group_id=group_cfg.id,
                    sk_node=keys.sk_ed25519,
                    gek=gek,
                    on_change=self._on_index_change,
                )
                await indexer.start()
                self._indexers.append(indexer)
                self._state["indexes"][group_cfg.id] = indexer.index
                log.info("Indexing group %s: %s (%d files)",
                         group_cfg.name,
                         ", ".join(f"{r.name}={r.path}" for r in roots),
                         indexer.index.count)

                groups_ctx[group_cfg.id] = {
                    "gek": gek,
                    "roots": roots,
                    "index": indexer.index,
                    "visibility": group_cfg.visibility,
                    # Admission policy comes from node.toml, never from the hub:
                    # a hub that could declare a group open would be handed its key.
                    "join_policy": group_cfg.join_policy,
                    # Whether ordinary members may upload. Read once here, into
                    # the context, because the upload handler is synchronous and
                    # a database round trip per chunk would be absurd. The
                    # signed operation that changes it updates this dict in
                    # place, so the two never drift within a run.
                    "member_upload": await self._roster.member_upload_allowed(
                        group_cfg.id) if self._roster else True,
                }

            if not groups_ctx:
                log.error("No valid groups configured — exiting")
                return

            # 5. Chat stores (one SQLite DB per group)
            for gid in groups_ctx:
                chat_db = data_dir / gid[:16] / "chat.db"
                store = ChatStore(db_path=chat_db)
                await store.open()
                self._chat_stores[gid] = store
                groups_ctx[gid]["chat_store"] = store
            log.info("Chat stores opened: %d groups", len(self._chat_stores))

            # 6. Audit store (legal compliance — IP + action logging)
            audit_db = data_dir / "audit.db"
            self._audit_store = AuditStore(db_path=audit_db)
            await self._audit_store.open()
            log.info("Audit store opened: %s", audit_db)

            # 5. Denylist
            denylist = self._denylist

            # 6. WebRTC transport (browser clients)
            first = next(iter(groups_ctx.values()))
            if WEBRTC_AVAILABLE:
                self._webrtc = WebRTCTransport(
                    sk_node=keys.sk_ed25519,
                    hub_pk_pem=session.hub_pk_pem,
                    gek=first["gek"],
                    roots=first["roots"],
                    index=first["index"],
                    groups=groups_ctx,
                    denylist=denylist,
                    max_concurrent_streams=self._config.node.max_concurrent_streams,
                )
                # No global chat_store here: each group's store lives in
                # groups_ctx[gid]["chat_store"] and is resolved per session via
                # _group_ctx(). Assigning the first group's store transport-wide
                # sent every group's chat to one database and served it back to
                # members of every other group (finding H1).
                self._webrtc._ctx["hub_ws"] = _WsSender(hub)
                self._webrtc._ctx["node_user_id"] = session.user_id
                self._webrtc._ctx["audit_store"] = self._audit_store
                self._webrtc._ctx["bundle_store"] = self._bundle_store
                self._webrtc._ctx["sk_x25519_raw"] = sk_x_raw
                self._webrtc._ctx["pk_x25519_raw"] = pk_x_raw
                self._webrtc._ctx["pk_x25519_b64"] = keys.pk_x25519_b64

                self._webrtc._ctx["roster"] = self._roster
                # The MNP adapter calls the same operations as the loopback API
                # (meshbay_node.ops), and those take the daemon's state. Handing
                # the transport a second set of lookups is how two paths to one
                # operation start disagreeing — the shape of C1 and C6.
                self._webrtc._ctx["daemon_state"] = self._state
                self._webrtc._ctx["invite_ttl"] = (
                    self._config.node.invite_ttl_hours * 3600)
                self._webrtc._ctx["device_request_ttl"] = (
                    self._config.node.device_request_ttl_minutes * 60)
                paired = await self._roster.has_operator() if self._roster else False
                self._webrtc._ctx["has_admin_authority"] = paired
                if paired:
                    log.info("Node authority: paired operator")
                else:
                    log.warning(
                        "No operator paired — invites and file deletion are "
                        "refused. Run: meshbay-node operator pair")
                log.info("WebRTC transport ready")
            else:
                log.warning("WebRTC not available (aiortc not installed)")

            # 7. QUIC chunk server (LAN / port-forwarded / hub-less direct access)
            if QUIC_AVAILABLE:
                self._quic_server = QuicChunkServer(
                    sk_node=keys.sk_ed25519,
                    hub_pk_pem=session.hub_pk_pem,
                    gek=first["gek"],
                    roots=first["roots"],
                    index=first["index"],
                    host="::",
                    port=self._config.node.quic_port,
                    groups=groups_ctx,
                    denylist=denylist,
                )
                await self._quic_server.start()
                log.info("QUIC server on port %d (%d groups)",
                         self._config.node.quic_port, len(groups_ctx))

            # 8. Hub WebSocket (signaling + revocations + WebRTC offers)
            async def on_webrtc_offer(sdp, peer_id, ice_candidates):
                if not self._webrtc:
                    return None
                try:
                    answer_sdp, answer_ice = await self._webrtc.handle_offer(
                        sdp, peer_id)
                    log.info("WebRTC answer for peer=%s (%d peers)",
                             peer_id, self._webrtc.active_peers)
                    return (answer_sdp, answer_ice)
                except Exception as e:
                    log.error("WebRTC offer failed: %s", e)
                    return None

            async def on_incoming(peer_ip, peer_port):
                if self._quic_server:
                    self._quic_server.punch_nat(peer_ip, peer_port)

            def on_revocation(token):
                if denylist and token:
                    import jwt as _jwt
                    try:
                        payload = _jwt.decode(
                            token, session.hub_pk_pem, algorithms=["EdDSA"],
                            options={"verify_exp": False})
                        target = payload.get("target")
                        tid = payload.get("target_id", "")
                        if target == "user":
                            denylist.deny_user(tid)
                        elif target == "group":
                            # H4: previously dropped on the floor, so "suspend a
                            # group" was a hub-only gesture that no node enforced.
                            denylist.deny_group(tid)
                            self._drop_group_sessions(tid)
                        elif target == "jti":
                            denylist.deny_jti(tid)
                        else:
                            log.warning("Unknown revocation target: %r", target)
                    except Exception as e:
                        log.warning("Invalid revocation token: %s", e)

            ws_task = asyncio.create_task(hub.maintain_ws(
                on_incoming=on_incoming,
                on_revocation=on_revocation,
                on_webrtc_offer=on_webrtc_offer,
                group_ids=list(groups_ctx.keys()),
            ))
            self._tasks.append(ws_task)
            log.info("Hub WS task started")

            # 9. (removed in Phase 11.5) The per-group HTTP file API used to start here.
            # It served the Mesh Group Index and raw plaintext files on 0.0.0.0 with no
            # authentication, for private groups too — finding C1. Every client path now
            # goes through the MNP handshake (JWT + group claim + GEK proof).

            # 10. Update admin UI state (UI already running from step 2)
            self._state["groups_ctx"] = groups_ctx
            self._state["audit_store"] = self._audit_store
            self._state["bundle_store"] = self._bundle_store
            self._state["roster"] = self._roster
            self._state["node_user_id"] = session.user_id
            self._state["webrtc"] = self._webrtc
            self._state["quic_server"] = self._quic_server
            self._state["hub"] = hub
            # Rotating a key has to reach every transport holding a copy of it,
            # and clearing the denylist has to reach the one the handshake
            # consults — so both are published rather than reachable only
            # through the object that happens to own them.
            self._state["denylist"] = self._denylist
            self._state["pk_x25519_raw"] = pk_x_raw

            self._state["status"] = "running"
            log.info("Node ready — %d groups, WebRTC=%s, QUIC=%s",
                     len(groups_ctx),
                     "yes" if self._webrtc else "no",
                     "yes" if self._quic_server else "no")

            # 11. Initial swarm registration — PUBLIC groups only.
            # Finding H7: registering every group's hashes hands the hub a content
            # fingerprint of every private file on the node, which is exactly the
            # metadata the "hub stores no content metadata" claim rules out. It also
            # lets anyone confirm whether a known file exists in the network.
            endpoint = f"webrtc:{self._config.node.quic_port}"
            for gctx in groups_ctx.values():
                if gctx.get("visibility") != "public":
                    continue
                hashes = [e.id for e in gctx["index"].entries]
                if hashes:
                    asyncio.ensure_future(self._register_swarm(hashes, endpoint))

            # 12. Wait for shutdown
            stop_event = asyncio.Event()
            loop = asyncio.get_event_loop()
            for sig in (signal.SIGINT, signal.SIGTERM):
                loop.add_signal_handler(sig, stop_event.set)
            # Milestone 14.8: re-read node.toml without dropping connections.
            try:
                loop.add_signal_handler(
                    signal.SIGHUP,
                    lambda: asyncio.ensure_future(self._reload_config()))
            except (NotImplementedError, AttributeError):
                pass          # no SIGHUP on Windows; `reload` says so there
            await stop_event.wait()

        await self._shutdown()

    async def _reload_config(self) -> None:
        """
        Re-read node.toml on SIGHUP.

        Deliberately narrow: it picks up **root changes on groups already
        hosted**, which is what an operator adjusts day to day, and reports
        anything else as needing a restart. Adding or removing a whole group
        means new indexers, chat stores, GEK loads and transport contexts, and
        doing that under a live daemon is how a half-built group ends up serving
        content. Saying "restart for that" is honest and costs one restart.

        Nothing here touches connections: a member watching a film keeps
        watching it.
        """
        log.info("SIGHUP — re-reading %s", self._config_path)
        try:
            fresh = load_config(self._config_path)
        except Exception as e:
            log.error("Reload failed, keeping the running config: %s", e)
            return

        groups_ctx = self._state.get("groups_ctx") or {}
        hosted = set(groups_ctx)
        incoming = {g.id for g in fresh.groups if g.id}
        if incoming != hosted:
            added = ", ".join(sorted(incoming - hosted)) or "none"
            removed = ", ".join(sorted(hosted - incoming)) or "none"
            log.warning("Group set changed (added: %s, removed: %s) — restart the "
                        "daemon for that; roots of existing groups reloaded anyway",
                        added, removed)

        changed = 0
        for group_cfg in fresh.groups:
            ctx = groups_ctx.get(group_cfg.id)
            if not ctx:
                continue
            try:
                roots = RootSet.build([asdict(r) for r in group_cfg.roots])
            except RootError as e:
                log.error("Group %r: %s — keeping the roots already loaded",
                          group_cfg.name, e)
                continue
            before = {(r.name, str(r.path)) for r in ctx["roots"]}
            after = {(r.name, str(r.path)) for r in roots}
            if before == after:
                continue
            roots.refresh_availability()
            indexer = next((i for i in self._indexers
                            if i.group_id == group_cfg.id), None)
            if indexer is None:
                continue
            log.info("Group %r roots changed: %s", group_cfg.name,
                     ", ".join(f"{r.name}={r.path}" for r in roots))
            await indexer.retarget(roots)
            ctx["roots"] = roots
            changed += 1

        self._config = fresh
        self._state["config"] = fresh
        log.info("Reload complete — %d group(s) re-rooted", changed)

    async def _login_with_retry(self, hub: HubClient):
        """Login to hub, retrying if the node key hasn't been linked yet."""
        import httpx as _httpx
        while True:
            try:
                return await hub.startup(endpoint_hint=None)
            except _httpx.HTTPStatusError as e:
                body = e.response.text if hasattr(e.response, 'text') else ''
                # Any 401 here needs a human at a browser, and the operator needs
                # this daemon alive to read its public key out of the local admin
                # UI. Exiting would take that UI down and strand them — which is
                # exactly what happened when a node was started before its owner
                # had registered.
                if e.response.status_code == 401:
                    if "No node key" in body:
                        self._state["status"] = "waiting_for_node_key"
                        log.warning(
                            "Node key not linked. Open the admin UI, copy this "
                            "node's key, and paste it in Settings > Link Node on "
                            "%s. Retrying in 30s...",
                            self._config.hub.url,
                        )
                    else:
                        self._state["status"] = "waiting_for_account"
                        log.warning(
                            "Hub rejected the node credentials for user %r. "
                            "Register that account on %s first, then link this "
                            "node's key. Retrying in 30s...",
                            self._config.hub.username, self._config.hub.url,
                        )
                    await asyncio.sleep(30)
                else:
                    raise
            except Exception as e:
                log.warning("Hub login failed: %s — retrying in 10s", e)
                await asyncio.sleep(10)

    async def _load_gek(
        self,
        group_id: str,
        node_user_id: str,
        sk_x_raw: bytes,
        pk_x_raw: bytes,
    ) -> bytes | None:
        """Load GEK from local bundle store (node-only, hub never touches crypto)."""
        from meshbay_common.crypto import unwrap_gek_aes

        if not self._bundle_store:
            return None

        # Try node-specific bundle first (stored by init_gek for daemon reload),
        # then fall back to operator's user bundle (legacy / pre-dual-key)
        for user_key in [f"_node_{node_user_id}", node_user_id]:
            bundle = await self._bundle_store.fetch(group_id, user_key)
            if not bundle:
                continue
            try:
                gek = unwrap_gek_aes(bundle, sk_x_raw, pk_x_raw)
                log.info("GEK loaded from local bundle store for group %s (key=%s)",
                         group_id[:8], user_key[:16])
                return gek
            except Exception as e:
                log.debug("Failed to unwrap GEK bundle (key=%s): %s", user_key[:16], e)

        log.warning("No unwrappable GEK bundle found for group %s", group_id[:8])
        return None

    async def _on_index_change(self, indexer: DirectoryIndexer) -> None:
        """Called when a DirectoryIndexer detects file changes."""
        group_id = indexer.group_id
        idx = indexer.index
        log.info("Index changed for group %s: %d files (v%d)",
                 group_id[:8], idx.count, idx.version)

        # 11.5 — Push updated index to connected WebRTC peers in this group
        if self._webrtc:
            entries = [
                {
                    "id": e.id, "name": e.name, "path": e.path,
                    "size": e.size, "type": e.type, "added_at": e.added_at,
                }
                for e in idx.entries
            ]
            sync_msg = {
                "type": MNP.INDEX_SYNC,
                "v": MNP_VERSION,
                "group_id": idx.group_id,
                "version": idx.version,
                "entries": entries,
            }
            pushed = 0
            for session in list(self._webrtc._sessions.values()):
                if session._group_id == group_id:
                    try:
                        session._send(sync_msg)
                        pushed += 1
                    except Exception:
                        pass
            if pushed:
                log.info("Index pushed to %d WebRTC peers", pushed)

        # 11.9 — Register file hashes with hub swarm table (public groups only, H7)
        group_cfg = next(
            (g for g in self._config.groups if g.id == group_id), None)
        if (self._hub and self._state.get("endpoint_hint")
                and group_cfg and group_cfg.visibility == "public"):
            hashes = [e.id for e in idx.entries]
            if hashes:
                endpoint = f"webrtc:{self._config.node.quic_port}"
                asyncio.ensure_future(self._register_swarm(hashes, endpoint))

    def _drop_group_sessions(self, group_id: str) -> None:
        """Close live sessions for a revoked group (H4)."""
        if not self._webrtc or not group_id:
            return
        for session in list(self._webrtc._sessions.values()):
            if session._group_id == group_id:
                asyncio.ensure_future(session.close())
                log.info("Dropped session for revoked group %s", group_id[:8])

    async def _register_swarm(self, hashes: list[str], endpoint: str) -> None:
        try:
            n = await self._hub.register_swarm(hashes, endpoint)
            log.info("Swarm: registered %d/%d hashes", n, len(hashes))
        except Exception as e:
            log.warning("Swarm registration failed: %s", e)

    async def _shutdown(self) -> None:
        log.info("Shutting down...")
        self._state["status"] = "stopping"

        for task in self._tasks:
            task.cancel()
        for task in self._tasks:
            try:
                await task
            except (asyncio.CancelledError, Exception):
                pass

        if self._webrtc:
            await self._webrtc.close_all()

        if self._audit_store:
            await self._audit_store.close()

        if self._bundle_store:
            await self._bundle_store.close()

        if self._roster:
            await self._roster.close()

        for store in self._chat_stores.values():
            await store.close()

        for indexer in self._indexers:
            await indexer.stop()

        if self._quic_server:
            await self._quic_server.stop()

        token_file = getattr(self, "_ui_token_file", None)
        if token_file is not None:
            token_file.unlink(missing_ok=True)

        log.info("Node stopped")


# ── CLI helpers ───────────────────────────────────────────────────────────────

def _daemon_api(cfg: Config, path: str, method: str = "GET",
                timeout: int = 30, body: dict | None = None) -> dict:
    """
    Call the daemon's loopback API.

    The daemon owns the roster, the hub session and the live group contexts, so
    the CLI asks it to act rather than opening its databases behind its back. It
    also means every operator action goes through the same authorization as the
    admin UI (the per-run session token, 11.5.3).
    """
    import json as _json
    import urllib.error
    import urllib.parse
    import urllib.request

    token_file = cfg.data_dir / "ui-token"
    if not token_file.exists():
        print("Node is not running — start it with: meshbay-node")
        sys.exit(1)

    sep = "&" if "?" in path else "?"
    url = (f"http://127.0.0.1:{cfg.node.ui_port}{path}"
           f"{sep}t={token_file.read_text().strip()}")
    try:
        data = _json.dumps(body).encode() if body is not None else None
        req = urllib.request.Request(
            url, method=method, data=data,
            headers={"Content-Type": "application/json"} if data else {})
        with urllib.request.urlopen(req, timeout=timeout) as r:
            return _json.loads(r.read())
    except urllib.error.HTTPError as e:
        raw = e.read().decode()[:600]
        try:
            parsed = _json.loads(raw)
            detail = parsed.get("error", raw)
            # Endpoints that refuse a name offer the ones that would work; a bare
            # "no such group" leaves the operator guessing at a UUID.
            for row in parsed.get("available", []):
                detail += f"\n  {row.get('name', ''):<24} {row.get('id', '')}"
        except Exception:
            detail = raw
        print(f"failed: {detail}")
        sys.exit(1)
    except Exception as e:
        print(f"failed: {e}")
        sys.exit(1)


def _resolve_group(cfg: Config, group: str | None) -> str:
    """
    The group argument as an id, or the only configured one.

    Accepts a name as well, because node.toml already gives every group one and
    nobody remembers a UUID. A name that matches nothing configured says so, and
    lists what is — silently passing it through produced a 404 from the daemon
    that read like the group did not exist on the hub.
    """
    if group:
        by_id = [g for g in cfg.groups if g.id == group]
        if by_id:
            return by_id[0].id
        by_name = [g for g in cfg.groups if g.name == group and g.id]
        if len(by_name) == 1:
            return by_name[0].id
        if len(by_name) > 1:
            print(f"several groups in node.toml are named {group!r} — use the id")
            sys.exit(1)
        # An id this node does not host is still worth passing on: the daemon
        # gives the better error, naming the group it does host.
        if "-" in group and len(group) == 36:
            return group
        print(f"no group named {group!r} in {DEFAULT_CONFIG_PATH}")
        if cfg.groups:
            print("configured groups:")
            for g in cfg.groups:
                print(f"  {g.name or '(unnamed)':<24} {g.id or '(no id yet)'}")
        sys.exit(1)
    configured = [g.id for g in cfg.groups if g.id]
    if len(configured) == 1:
        return configured[0]
    print("--group is required (several groups configured)"
          if configured else "no group configured in node.toml")
    sys.exit(1)


# ── Entry point ───────────────────────────────────────────────────────────────

def main() -> None:
    import argparse

    parser = argparse.ArgumentParser(description="MeshBay Node daemon")
    parser.add_argument("command", nargs="?",
                        choices=["init", "status", "ui", "gek-init", "gek",
                                 "operator", "member", "group", "file",
                                 "denylist", "reload", "calibrate-argon2"],
                        help="init: write example config | status: node state and keys "
                             "| ui: print the admin UI URL | operator pair: pair a "
                             "browser with this node | member list|invite|revoke|unpin "
                             "| group list|add | gek init|rotate | file list|rm "
                             "| denylist show|clear | reload: re-read node.toml "
                             "| calibrate-argon2: benchmark")
    parser.add_argument("subcommand", nargs="?",
                        help="'pair' for operator; list|invite|revoke|unpin for "
                             "member; list|add for group; init|rotate for gek; "
                             "list|rm for file; show|clear for denylist")
    parser.add_argument("target", nargs="?",
                        help="username for member invite|revoke|unpin; group name "
                             "for group add; file id for file rm; identifier for "
                             "denylist clear")
    parser.add_argument("--dir", default=None,
                        help="shared directory, for group add")
    parser.add_argument("--upload-dir", default=None,
                        help="separate upload directory, for group add")
    parser.add_argument("--yes", action="store_true",
                        help="skip the confirmation for destructive commands")
    parser.add_argument("--config", type=Path, default=None,
                        help="Config file path")
    parser.add_argument("--group", default=None,
                        help="group id (optional if only one is configured)")
    parser.add_argument("--log-level", default="INFO",
                        choices=["DEBUG", "INFO", "WARNING", "ERROR"])
    args = parser.parse_args()

    # Query commands print a report; library logging would interleave with it.
    quiet = args.command in ("status", "ui", "gek-init", "gek", "operator",
                             "member", "group", "file", "denylist", "reload")
    logging.basicConfig(
        level=logging.ERROR if quiet else getattr(logging, args.log_level),
        format="%(asctime)s %(levelname)-8s %(name)s: %(message)s",
    )

    if args.command == "init":
        write_example_config()
        print("Example config written. Edit it and run: meshbay-node")
        return

    if args.command == "calibrate-argon2":
        calibrate_argon2()
        return

    if args.command == "status":
        import json as _json
        import urllib.request

        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
        print(f"hub          {cfg.hub.url}  (user {cfg.hub.username or '—'})")

        # Read straight from the keystore: the operator needs this key to link the
        # node, and that happens before the daemon can ever stay running.
        try:
            keys = load_or_create_keystore(
                path=cfg.keystore.path, unlock_file=cfg.keystore.unlock_file)
            print(f"node key     {keys.pk_ed25519_b64}")
        except Exception as e:
            print(f"node key     <keystore locked: {e}>")

        token_file = cfg.data_dir / "ui-token"
        live = None
        if token_file.exists():
            try:
                url = (f"http://127.0.0.1:{cfg.node.ui_port}"
                       f"/api/status?t={token_file.read_text().strip()}")
                with urllib.request.urlopen(url, timeout=3) as r:
                    live = _json.loads(r.read())
            except Exception:
                live = None

        if live:
            print(f"daemon       running — {live.get('status')}")
            print(f"node_id      {live.get('endpoint_hint') or '—'}")
            print(f"groups       {live.get('group_count', 0)}"
                  f"  files {live.get('total_files', 0)}"
                  f"  peers {live.get('webrtc_peers', 0)}")
            print(f"admin UI     meshbay-node ui")
        else:
            print("daemon       not running")

        print(f"config       {DEFAULT_CONFIG_PATH}")
        if not cfg.groups:
            print("groups       none configured — create a group on the hub, then add")
            print("             a [[groups]] entry with its id and a directory")
        else:
            for g in cfg.groups:
                print(f"  group      {g.name} [{g.visibility}] {g.id or '<no id>'}")
                if not g.roots:
                    print("             <no directory configured>")
                for r in g.roots:
                    label = r.name or Path(r.path).name
                    flag = "  (uploads)" if r.upload else ""
                    live = "" if Path(r.path).expanduser().is_dir() else "  [UNAVAILABLE]"
                    print(f"             {label} → {r.path}{flag}{live}")
        # Node authority: the roster is the source of truth, node.toml the legacy
        # form. Read the DB directly so this reports correctly while the daemon is
        # stopped — the state an operator is most often in when checking.
        import asyncio as _asyncio

        from meshbay_node.roster import Roster as _Roster

        async def _read_roster() -> tuple[list, int]:
            r = _Roster(db_path=cfg.data_dir / "roster.db")
            await r.open()
            try:
                return (await r.list_members()), len(await r.list_invites())
            finally:
                await r.close()

        try:
            members, pending = _asyncio.run(_read_roster())
        except Exception as e:
            members, pending = [], 0
            print(f"roster       <unreadable: {e}>")

        operators = [m for m in members if m["role"] == "operator"
                     and m["status"] == "active"]
        if operators:
            for op in operators:
                print(f"operator     {op.get('username') or op['user_id'][:8]}"
                      f"  key {(op.get('pk_ed25519') or '')[:16]}…"
                      f"  paired {op.get('pinned_at', '?')}")
        else:
            print("operator     NONE PAIRED — file deletion and member invites are")
            print("             refused. Run: meshbay-node operator pair")
        if pending:
            print(f"invites      {pending} pending code(s)")
        return

    if args.command == "member":
        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
        sub = args.subcommand or "list"

        if sub == "list":
            group = args.group or ""
            out = _daemon_api(
                cfg, f"/api/roster?group_id={group}" if group else "/api/roster")
            identities = {i["user_id"]: i for i in out.get("identities", [])}

            members = out.get("members", [])
            if not members:
                print("no members admitted yet")
                print("invite someone:  meshbay-node member invite <username>")
            for m in members:
                ident = identities.get(m["user_id"], {})
                scope = m["group_id"][:8] if m["group_id"] else "node-wide"
                print(f"{(ident.get('username') or m['user_id'])[:20]:20} "
                      f"{m['role']:9} {m['status']:8} {scope:10} "
                      f"pinned {ident.get('pinned_at', '?')} "
                      f"({ident.get('pinned_via', '?')})")

            invites = out.get("invites", [])
            if invites:
                print()
                for i in invites:
                    print(f"pending invite  user {i['user_id'][:12]}  "
                          f"group {(i['group_id'] or 'node-wide')[:8]}  "
                          f"expires {i['expires_at']}")
            return

        if not args.target:
            print(f"usage: meshbay-node member {sub} <username>")
            sys.exit(1)

        if sub == "invite":
            group_id = _resolve_group(cfg, args.group)
            out = _daemon_api(
                cfg, f"/api/groups/{group_id}/invites?username={args.target}",
                method="POST")
            from meshbay_node.roster import write_code_file
            path = write_code_file(cfg.data_dir, out["code"],
                                   out.get("expires_at", ""), name="invite-code")
            print(f"INVITATION CODE  {out['code']}")
            print(f"valid until      {out.get('expires_at', '?')}")
            print()
            print(f"Send it to {args.target} however you normally talk. It works")
            print("once, for that account only, and never passes through the hub.")
            print("They enter it the first time they open the group — you do not")
            print("need to be online then.")
            print()
            print(f"also written to {path}")
            return

        # revoke and unpin both name a person; the daemon resolves the account.
        # It tries its own roster first and falls back to the hub, so a node that
        # pinned someone before invitations carried a name is still manageable.
        match = _daemon_api(cfg, f"/api/resolve?username={args.target}")

        if sub == "revoke":
            group_id = _resolve_group(cfg, args.group)
            out = _daemon_api(
                cfg, f"/api/members/{match['user_id']}/revoke?group_id={group_id}",
                method="POST")
            print(f"{args.target} revoked from {group_id[:8]}")
            print("They stop receiving the group key on their next connection.")
            print("They still hold the current one — rotate it:")
            print(f"  meshbay-node gek-init --group {group_id}")
            return

        if sub == "unpin":
            _daemon_api(cfg, f"/api/members/{match['user_id']}/unpin", method="POST")
            print(f"{args.target} unpinned — they can pair again with a new key")
            print(f"issue a code:  meshbay-node member invite {args.target}")
            return

        print("usage: meshbay-node member list|invite|revoke|unpin")
        sys.exit(1)

    if args.command in ("gek-init", "gek"):
        # `gek-init` is the original spelling and still works. `gek rotate` is
        # the one that matters after a revocation: the ex-member holds the
        # current key and nothing else takes it from them.
        sub = "init" if args.command == "gek-init" else (args.subcommand or "init")
        if sub not in ("init", "rotate"):
            print("usage: meshbay-node gek init|rotate [--group NAME]")
            sys.exit(1)

        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
        group_id = _resolve_group(cfg, args.group)

        if sub == "rotate" and not args.yes:
            print("Rotating replaces this group's key.")
            print("  · every member re-receives it automatically on their next connect")
            print("  · anyone revoked keeps the OLD key and loses access to new content")
            print("  · content already downloaded stays readable to whoever has it")
            if input("rotate now? [y/N] ").strip().lower() not in ("y", "yes"):
                print("cancelled")
                return

        out = _daemon_api(cfg, f"/api/groups/{group_id}/gek"
                               f"{'?rotate=true' if sub == 'rotate' else ''}",
                          method="POST", timeout=60)

        verb = "rotated" if out.get("rotated") else "ready"
        print(f"GEK {verb} for {group_id}")
        print(f"  {out.get('authorized_members', 0)} authorized member(s) — each "
              f"receives the key on connect")
        for err in out.get("errors") or []:
            print(f"  ! {err}")
        return

    if args.command == "reload":
        # Milestone 14.8. The daemon re-reads node.toml; groups that appeared or
        # whose roots changed are picked up without dropping live connections.
        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
        import os as _os
        import signal as _signal
        import subprocess as _subprocess
        # `--` is pgrep's own end-of-options marker and must be its own argument;
        # folded into the pattern it searches for a process literally called
        # "-- -m …". Anchored to the end of the command line so it matches the
        # daemon and never a shell that merely mentions it — the same trap
        # deploy-node.sh documents, where an unanchored pattern kills the script.
        pid_out = _subprocess.run(
            ["pgrep", "-f", "--", r"-m meshbay_node\.daemon$"],
            capture_output=True, text=True)
        pids = [int(x) for x in pid_out.stdout.split()]
        if not pids:
            print("Node is not running — start it with: meshbay-node")
            sys.exit(1)
        for pid in pids:
            _os.kill(pid, _signal.SIGHUP)
        print(f"sent SIGHUP to {len(pids)} daemon process(es)")
        print("watch the result:  tail -f /tmp/meshbay-node.log")
        return

    if args.command == "denylist":
        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
        sub = args.subcommand or "show"

        if sub == "show":
            out = _daemon_api(cfg, "/api/denylist")
            total = out.get("count", 0)
            if not total:
                print("denylist     empty — nothing is being refused")
                return
            for kind in ("users", "groups", "jtis"):
                for entry in out.get(kind, []):
                    print(f"  {kind[:-1]:<6} {entry}")
            print(f"\n{total} entr(y/ies). These survive a restart (finding H4).")
            return

        if sub == "clear":
            if not args.yes:
                what = args.target or "EVERY entry"
                print(f"Clearing the denylist re-admits {what}.")
                print("A revocation the hub sent will not come back on its own.")
                if input("clear now? [y/N] ").strip().lower() not in ("y", "yes"):
                    print("cancelled")
                    return
            out = _daemon_api(cfg, f"/api/denylist/clear?subject={args.target or ''}",
                              method="POST")
            print(f"removed {out['removed']} entr(y/ies) ({out['subject']})")
            return

        print("usage: meshbay-node denylist show|clear [identifier] [--yes]")
        sys.exit(1)

    if args.command == "file":
        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
        sub = args.subcommand or "list"
        group_id = _resolve_group(cfg, args.group)

        if sub == "list":
            out = _daemon_api(cfg, f"/api/groups/{group_id}/files")
            files = sorted(out.get("files", []), key=lambda f: (f["path"], f["name"]))
            if not files:
                print("no files indexed")
                return
            for f in files:
                print(f"  {f['id'][:12]}  {f['size']:>12}  {f['path']}/{f['name']}")
            print(f"\n{len(files)} file(s). Remove one with: "
                  f"meshbay-node file rm <id>")
            return

        if sub == "rm":
            # Milestone 14.11 — the last operator action that needed a browser.
            if not args.target:
                print("usage: meshbay-node file rm <file-id> [--group NAME]")
                sys.exit(1)
            out = _daemon_api(cfg, f"/api/groups/{group_id}/files",)
            matches = [f for f in out.get("files", [])
                       if f["id"].startswith(args.target)]
            if not matches:
                print(f"no file whose id starts with {args.target!r}")
                sys.exit(1)
            if len(matches) > 1:
                print(f"{args.target!r} matches {len(matches)} files — be more specific:")
                for f in matches[:10]:
                    print(f"  {f['id'][:16]}  {f['path']}/{f['name']}")
                sys.exit(1)
            target = matches[0]
            if not args.yes:
                print(f"Delete {target['path']}/{target['name']} "
                      f"({target['size']} bytes) from disk?")
                print("This removes the file itself, not just the listing.")
                if input("delete? [y/N] ").strip().lower() not in ("y", "yes"):
                    print("cancelled")
                    return
            _daemon_api(cfg, f"/api/groups/{group_id}/files/{target['id']}",
                        method="DELETE")
            print(f"deleted {target['path']}/{target['name']}")
            return

        print("usage: meshbay-node file list|rm <id> [--group NAME] [--yes]")
        sys.exit(1)

    if args.command == "group":
        if args.subcommand in (None, "list"):
            # Milestone 14.2.
            cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
            out = _daemon_api(cfg, "/api/groups")
            groups = out.get("groups", [])
            if not groups:
                print("no groups hosted — add one with: "
                      "meshbay-node group add <name> --dir <path>")
                return
            for g in groups:
                key = "GEK" if g.get("has_gek") else "NO KEY"
                print(f"  {g['name']}  [{g['visibility']}/{g.get('join_policy')}]  "
                      f"{key}  {g['file_count']} file(s)  "
                      f"{g.get('peers', 0)} peer(s)")
                print(f"      {g['id']}")
                for r in g.get("roots", []):
                    flags = ""
                    if r.get("upload"):
                        flags = " (uploads, direct)" if r.get("direct") else " (uploads)"
                    live = "" if r.get("available", True) else "  [UNAVAILABLE]"
                    print(f"      root {r['name']}{flags}{live}")
                if not g.get("has_gek"):
                    print(f"      give it a key:  meshbay-node gek init "
                          f"--group {g['name']}")
            return

        if args.subcommand != "add":
            print("usage: meshbay-node group list|add <name> --dir <path> [--upload-dir <path>]")
            sys.exit(1)
        if not args.target or not args.dir:
            print("usage: meshbay-node group add <name> --dir <path> [--upload-dir <path>]")
            print()
            print("The group must already exist on the hub and be yours. This")
            print("only tells the node to host it, and picks the directory.")
            print("--upload-dir sets a separate directory for uploaded files.")
            sys.exit(1)

        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
        body = {"name": args.target, "shared_dir": args.dir}
        if args.upload_dir:
            body["upload_dir"] = args.upload_dir
        out = _daemon_api(cfg, "/api/groups/attach", method="POST", body=body)
        print(f"{out['name']} ({out['group_id'][:8]}) added to {out['config']}")
        print(f"  shared_dir   {out['shared_dir']}")
        if out.get("upload_dir"):
            print(f"  upload_dir   {out['upload_dir']}")
        print()
        print("Tell the daemon to re-read its config, then give the group a key:")
        print("  meshbay-node reload")
        print(f"  meshbay-node gek init --group {out['name']}")
        print()
        print("The key is this group's own — members of your other groups cannot")
        print("read it, and joining one says nothing about the other.")
        return

    if args.command == "operator":
        if args.subcommand != "pair":
            print("usage: meshbay-node operator pair")
            sys.exit(1)
        if args.group:
            # Silently ignoring it invited the reading that a code belongs to a
            # group, and then that pairing had not worked because the group did
            # not change.
            print("operator pair takes no --group: pairing is node-wide.")
            print("One paired browser can invite to, and delete files in, every")
            print("group this node hosts.")
            sys.exit(1)

        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
        out = _daemon_api(cfg, "/api/operator/pair", method="POST")

        from meshbay_node.roster import write_code_file
        path = write_code_file(cfg.data_dir, out["code"], out.get("expires_at", ""))

        print(f"PAIRING CODE   {out['code']}")
        print(f"valid until    {out.get('expires_at', '?')}")
        print()
        print("Sign in to the web app as this node's operator, open one of your")
        print("groups, go to the Members tab and enter the code there.")
        print("It works once, for that account only, and authorizes invites and")
        print("file deletion from that browser.")
        print()
        print(f"also written to {path}")
        return

    if args.command == "ui":
        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
        token_file = cfg.data_dir / "ui-token"
        if not token_file.exists():
            print("Node does not appear to be running — start it with: meshbay-node")
            sys.exit(1)
        print(f"http://127.0.0.1:{cfg.node.ui_port}"
              f"/?t={token_file.read_text().strip()}")
        print()
        print("The UI listens on loopback only. From another machine:")
        print(f"  ssh -L {cfg.node.ui_port}:127.0.0.1:{cfg.node.ui_port} <this-host>")
        return

    cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
    if not cfg.hub.username:
        print("Error: hub.username not set in config. Run: meshbay-node init")
        sys.exit(1)

    daemon = NodeDaemon(cfg, Path(args.config or DEFAULT_CONFIG_PATH))
    asyncio.run(daemon.run())


if __name__ == "__main__":
    main()