"""
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
import base64
import json
import logging
import os
import signal
import sys
from pathlib import Path

import uvicorn
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey

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.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):
        self._config = config
        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
        # 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.shared_dir:
                    log.warning("Group %r missing id or shared_dir — skipping",
                                group_cfg.name)
                    continue

                shared_root = Path(group_cfg.shared_dir).expanduser().resolve()
                if not shared_root.exists():
                    log.warning("Shared dir not found: %s — skipping group %s",
                                shared_root, group_cfg.name)
                    continue

                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(
                    root=shared_root,
                    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, shared_root, indexer.index.count)

                groups_ctx[group_cfg.id] = {
                    "gek": gek,
                    "shared_root": shared_root,
                    "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,
                }

            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"],
                    shared_root=first["shared_root"],
                    index=first["index"],
                    groups=groups_ctx,
                    denylist=denylist,
                )
                # 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
                self._webrtc._ctx["invite_ttl"] = (
                    self._config.node.invite_ttl_hours * 3600)
                admin_pk = self._legacy_admin_pk()
                paired = await self._roster.has_operator() if self._roster else False
                if admin_pk:
                    self._webrtc._ctx["admin_pk_ed25519"] = admin_pk
                self._webrtc._ctx["has_admin_authority"] = paired
                if paired or admin_pk:
                    sources = ([] if not paired else ["paired operator"]) + \
                              ([] if not admin_pk else ["node.toml admin_pk"])
                    log.info("Node authority: %s", " + ".join(sources))
                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"],
                    shared_root=first["shared_root"],
                    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["hub"] = hub
            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)
            await stop_event.wait()

        await self._shutdown()

    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

    def _legacy_admin_pk(self) -> Ed25519PublicKey | None:
        """
        The pre-roster way of naming the operator: `admin_pk_ed25519` in node.toml.

        Still honoured so a deployment configured that way keeps working, but no
        longer the only path — and the auto-pin that used to stand in for it is
        gone. It pinned the node's *keystore* key while the browser signed with the
        user's *identity* key, so admin operations failed closed with a signature
        error that looked like a bug elsewhere (finding M3). An operator now pairs
        a browser with `meshbay-node operator pair`.
        """
        if not self._config.admin_pk_ed25519:
            return None
        try:
            raw = base64.b64decode(self._config.admin_pk_ed25519)
            return Ed25519PublicKey.from_public_bytes(raw)
        except Exception as e:
            log.error("Invalid admin_pk_ed25519 in config: %s", e)
            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) -> 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:
        req = urllib.request.Request(url, method=method)
        with urllib.request.urlopen(req, timeout=timeout) as r:
            return _json.loads(r.read())
    except urllib.error.HTTPError as e:
        body = e.read().decode()[:300]
        try:
            detail = _json.loads(body).get("error", body)
        except Exception:
            detail = body
        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, or the only configured one."""
    if group:
        return group
    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", "operator",
                                 "member", "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 "
                             "| calibrate-argon2: benchmark")
    parser.add_argument("subcommand", nargs="?",
                        help="'pair' for operator; list|invite|revoke|unpin for member")
    parser.add_argument("target", nargs="?",
                        help="username, for member invite|revoke|unpin")
    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", "operator", "member")
    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 shared_dir")
        else:
            for g in cfg.groups:
                print(f"  group      {g.name} [{g.visibility}] {g.id or '<no id>'}")
                print(f"             {g.shared_dir or '<no shared_dir>'}")
        # 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', '?')}")
        elif cfg.admin_pk_ed25519:
            print("operator     node.toml admin_pk_ed25519 (legacy)")
            print("             run `meshbay-node operator pair` to replace it")
        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 == "gek-init":
        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
        group_id = _resolve_group(cfg, args.group)
        out = _daemon_api(cfg, f"/api/groups/{group_id}/gek",
                          method="POST", timeout=60)

        print(f"GEK ready 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 == "operator":
        if args.subcommand != "pair":
            print("usage: meshbay-node operator pair")
            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)
    asyncio.run(daemon.run())


if __name__ == "__main__":
    main()