1 files changed, 340 insertions, 0 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/ops/settings.py b/packages/meshbay-node/src/meshbay_node/ops/settings.py
new file mode 100644
index 0000000..eade6ca
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/ops/settings.py
@@ -0,0 +1,340 @@
+"""Node-wide settings: denylist, node.toml values, transfer caps, scan timing, reload."""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+from meshbay_common.background import spawn
+
+from meshbay_node.config import DEFAULT_CONFIG_PATH
+from meshbay_node.ops.core import OpError, _config, _group_ctx, _roster
+from meshbay_node.ops.node_toml import _update_node_toml
+from meshbay_node.roster import Roster
+
+log = logging.getLogger("meshbay_node.ops")
+
+
+# ── Revocation denylist ──────────────────────────────────────────────────────
+
+async def read_denylist(state: dict) -> dict:
+    """Milestone 14.10 — what the node is currently refusing."""
+    denylist = state.get("denylist")
+    if not denylist:
+        return {"users": [], "groups": [], "jtis": [], "count": 0}
+    entries = denylist.entries()
+    return {**entries, "count": sum(len(v) for v in entries.values())}
+
+
+async def clear_denylist(state: dict, *, subject: str = "") -> dict:
+    """
+    Drop denylist entries — all of them, or one identifier.
+
+    Deliberately not silent: a cleared denylist re-admits whoever it was keeping
+    out, and the count is what tells the operator whether they undid one
+    revocation or all of them.
+    """
+    denylist = state.get("denylist")
+    if not denylist:
+        raise OpError("No denylist in this process", status=503)
+    removed = denylist.clear(subject)
+    log.warning("Denylist cleared (%s): %d entr(y/ies) removed",
+                subject or "all", removed)
+    return {"status": "cleared", "removed": removed, "subject": subject or "all"}
+
+
+# ── Node settings ────────────────────────────────────────────────────────────
+
+# What `set_node_settings` accepts, and how each value is validated. A module
+# constant so a test can hold its key set against `Roster.node_setting_keys()`:
+# this is the third list of the same settings, and the first two had already
+# drifted apart once — the reader's defaults covered fewer settings than the
+# resolver answered for, which is how node.toml's transfer pools came to be
+# parsed and then ignored. The kinds here are the *writer's* validation and
+# deliberately not the resolver's coercions.
+NODE_SETTING_WRITERS: dict[str, tuple[str, str]] = {
+    "invite_ttl_hours": ("int", Roster.SETTING_INVITE_TTL),
+    "pair_ttl_hours": ("int", Roster.SETTING_PAIR_TTL),
+    "device_request_ttl_minutes": ("int", Roster.SETTING_DEVICE_TTL),
+    "max_concurrent_streams": ("int", Roster.SETTING_MAX_STREAMS),
+    "max_concurrent_downloads": ("int", Roster.SETTING_MAX_DOWNLOADS),
+    "max_concurrent_uploads": ("int", Roster.SETTING_MAX_UPLOADS),
+    "max_upload_gb": ("size", Roster.SETTING_MAX_UPLOAD_GB),
+    "transcode_incompatible_video": ("bool", Roster.SETTING_TRANSCODE),
+    "stun_servers": ("stun_list", Roster.SETTING_STUN_SERVERS),
+    "ice_interfaces": ("list", Roster.SETTING_ICE_INTERFACES),
+}
+
+
+async def get_node_settings(state: dict) -> dict:
+    """Return current effective node settings."""
+    from meshbay_node.config import node_settings_defaults
+    roster = _roster(state)
+    config = _config(state)
+    defaults = node_settings_defaults(config.node)
+    if roster:
+        return await roster.node_settings(defaults)
+    return defaults
+
+async def set_node_settings(state: dict, settings: dict) -> dict:
+    """Update node-level daemon settings. Writes to both roster.db and node.toml."""
+    roster = _roster(state)
+    config = _config(state)
+    nd = config.node
+    conf_path = Path(state.get("config_path") or DEFAULT_CONFIG_PATH)
+
+    allowed_keys = NODE_SETTING_WRITERS
+
+    set_by = state.get("node_user_id", "")
+    updated = {}
+    for key, value in settings.items():
+        if key not in allowed_keys:
+            continue
+        kind, setting_key = allowed_keys[key]
+        if kind == "int":
+            try:
+                v = int(value)
+            except (TypeError, ValueError):
+                raise OpError(f"{key} must be an integer")
+            if v < 1:
+                raise OpError(f"{key} must be positive")
+            setattr(nd, key, v)
+            await roster.set_node_setting(setting_key, str(v), set_by)
+            updated[key] = v
+        elif kind == "size":
+            # A quantity, not a count: half a gigabyte is a legitimate ceiling
+            # on a small disk, so this one is not run through the `int` branch
+            # above, whose floor of 1 would round it to "refuse everything".
+            # bool before float, as config.py does it: `true` is not 1 GB.
+            if isinstance(value, bool):
+                raise OpError(f"{key} must be a number")
+            try:
+                fv = float(value)
+            except (TypeError, ValueError):
+                raise OpError(f"{key} must be a number")
+            if fv <= 0:
+                raise OpError(f"{key} must be greater than zero")
+            setattr(nd, key, fv)
+            await roster.set_node_setting(setting_key, repr(fv), set_by)
+            updated[key] = fv
+        elif kind == "bool":
+            v = bool(value)
+            setattr(nd, key, v)
+            await roster.set_node_setting(setting_key, "1" if v else "0", set_by)
+            updated[key] = v
+        elif kind in ("list", "stun_list"):
+            import json as _json
+            if not isinstance(value, list):
+                raise OpError(f"{key} must be a list")
+            v = [str(s) for s in value]
+            if kind == "stun_list":
+                for s in v:
+                    if not s.startswith("stun:"):
+                        raise OpError(f"Invalid STUN server: {s} (must start with stun:)")
+            setattr(nd, key, v)
+            await roster.set_node_setting(setting_key, _json.dumps(v), set_by)
+            updated[key] = v
+
+    if updated:
+        _update_node_toml(conf_path, updated)
+        if "max_concurrent_streams" in updated:
+            webrtc = state.get("webrtc")
+            # `webrtc._stream_sem` was assigned here for months. That attribute
+            # has never existed -- the pool is `ctx["_transcode_sem"]` -- so the
+            # `hasattr` guard was always False and the setting only ever took
+            # effect on a restart, which docs/MESHBAY_DESIGN.md §6.8 says it
+            # does not need.
+            if webrtc is not None:
+                webrtc.set_capacity(
+                    max_concurrent_streams=updated["max_concurrent_streams"])
+        if ("max_concurrent_downloads" in updated
+                or "max_concurrent_uploads" in updated):
+            webrtc = state.get("webrtc")
+            if webrtc is not None:
+                webrtc.set_capacity(
+                    max_concurrent_downloads=updated.get(
+                        "max_concurrent_downloads"),
+                    max_concurrent_uploads=updated.get(
+                        "max_concurrent_uploads"))
+        if "max_upload_gb" in updated:
+            webrtc = state.get("webrtc")
+            if webrtc is not None:
+                webrtc.set_capacity(max_upload_gb=updated["max_upload_gb"])
+        if "stun_servers" in updated:
+            webrtc = state.get("webrtc")
+            if webrtc and hasattr(webrtc, '_stun'):
+                webrtc._stun = updated["stun_servers"]
+            from meshbay_node.transport.stun_multi import set_servers as _set_stun
+            _set_stun(updated["stun_servers"])
+        if "ice_interfaces" in updated:
+            from meshbay_node.transport.ice_filter import install as install_ice_filter
+            install_ice_filter(updated["ice_interfaces"] or None)
+
+    log.info("Node settings updated: %s", updated)
+    return {"updated": updated}
+
+
+# ── Transfers ────────────────────────────────────────────────────────────────
+
+async def set_transfer_limits(state: dict, group_id: str,
+                              downloads: int, uploads: int) -> dict:
+    """How many transfers one member may run at once in this group.
+
+    Same shape as every other operator setting: lives on the node (roster.db,
+    not the hub and not node.toml, for the reason change 5 gives — a hub that
+    decided this would have authority over someone else's machine), signed
+    (webrtc_server checks the caller's admin authority before this runs), and
+    live, so the pools are updated in place rather than at the next restart.
+    """
+    roster = _roster(state)
+    ctx = _group_ctx(state, group_id)
+    limits = await roster.set_transfer_limits(
+        group_id, {"download": downloads, "upload": uploads},
+        set_by=state.get("node_user_id", ""))
+    ctx["transfer_limits"] = limits
+    webrtc = state.get("webrtc")
+    slots = getattr(webrtc, "_ctx", {}).get("_transfer_slots") if webrtc else None
+    granted = slots.set_group_limits(group_id, limits) if slots else []
+    # And **tell them**. The node-wide path (`WebRTCTransport.set_capacity`)
+    # does this and this one did not: the leases were granted in the pool and
+    # the peers waiting on them were never told, so a cap raised from 2 to 4
+    # left both transfers sitting at "waiting" until the client's own watchdog
+    # re-asked a minute later. That is §5.2's first row — "node granted a slot,
+    # the push was lost" — reached by writing the grant and forgetting the send,
+    # which is the same omission as the missing `touch()` one layer up.
+    for lease in granted:
+        webrtc._notify_granted(lease)
+    log.info("Transfer limits for group %s: %s (%d started at once)",
+             group_id[:8], limits, len(granted))
+    return {"group_id": group_id, "limits": limits,
+            "started": [x.tr for x in granted]}
+
+async def list_transfers(state: dict) -> dict:
+    """Live transfer leases and queue depth.
+
+    The operator's window into "is anything actually holding a slot". When
+    somebody reports a transfer stuck at waiting, this is the only thing that
+    says whether the node ever had them in a queue — the alternative is reading
+    a log for a line that, by definition, is not being printed.
+
+    Carries no filename and no path: a lease holds neither, and this is exactly
+    where it would be tempting to add one.
+    """
+    webrtc = state.get("webrtc")
+    ctx = getattr(webrtc, "_ctx", {}) if webrtc else {}
+    slots = ctx.get("_transfer_slots")
+    if slots is None:
+        from meshbay_node.transfers import DEFAULT_MAX_CONCURRENT, DEFAULT_MAX_PER_MEMBER, KINDS
+        # No pool built means nothing has transferred since the daemon started,
+        # which is a real answer and not an error.
+        #
+        # The caps still have to be the operator's own. Reporting the module
+        # defaults here was worse than reporting nothing: `transfers set 2 2`
+        # answered "applied now", and `transfers show` immediately said 0/8 —
+        # a setting written, acknowledged and displayed wrong, which reads
+        # exactly like the hot-swap that did nothing for months. Found by
+        # running it, not by a test: the test asserted the defaults and so
+        # agreed with the bug.
+        return {"pools": {
+            k: {"in_use": 0,
+                "cap": int(ctx.get(f"max_concurrent_{k}s")
+                           or DEFAULT_MAX_CONCURRENT),
+                "per_member": DEFAULT_MAX_PER_MEMBER,
+                "queued": 0}
+            for k in KINDS}, "leases": [], "groups": _group_limits(state)}
+    out = slots.snapshot()
+    out["groups"] = _group_limits(state)
+    return out
+
+
+def _group_limits(state: dict) -> list[dict]:
+    """Each group's per-member caps, as the operator set them.
+
+    Reported because `transfers show` used to print only the node's default and
+    an operator reading "2 per member" had no way to tell whether that was this
+    group's setting or the fallback — and no way to change it either, since the
+    signed op had no door but MNP. Both were the same bug wearing two faces.
+    """
+    from meshbay_node.transfers import DEFAULT_MAX_PER_MEMBER
+
+    config = state.get("config")
+    groups_ctx = state.get("groups_ctx") or {}
+    out = []
+    for group in (getattr(config, "groups", None) or []):
+        limits = (groups_ctx.get(group.id) or {}).get("transfer_limits") or {}
+        out.append({
+            "group_id": group.id,
+            "name": group.name,
+            "download": int(limits.get("download") or DEFAULT_MAX_PER_MEMBER),
+            "upload": int(limits.get("upload") or DEFAULT_MAX_PER_MEMBER),
+            "set": bool(limits),
+        })
+    return out
+
+
+# ── Scan settings ────────────────────────────────────────────────────────────
+
+async def set_scan_settings(state: dict, group_id: str, reconcile_interval_secs: float,
+                            debounce_secs: float) -> dict:
+    """
+    How often the indexer's reconciliation backstop runs, and how long a
+    changed file is left alone before being hashed (indexer.py
+    DirectoryIndexer). Persisted like set_enabled_apps —
+    but there is also a *live* DirectoryIndexer object to update, since it
+    reads these once at construction and runs its own background loop with
+    them rather than consulting groups_ctx on every use.
+    """
+    roster = _roster(state)
+    await roster.set_scan_settings(group_id, reconcile_interval_secs, debounce_secs,
+                                   set_by=state.get("node_user_id", ""))
+    indexer = state.get("indexers", {}).get(group_id)
+    if indexer:
+        indexer.reconcile_secs = reconcile_interval_secs
+        indexer.debounce_secs = debounce_secs
+        # Apply the new interval now rather than after whatever backoff had
+        # already stretched the wait to.
+        indexer.note_activity()
+    # Optional, unlike _group_ctx(): a group can be persisted here before it
+    # is hot-loaded (or in a test that only cares about the roster/indexer
+    # side), and that must not turn a successful write into a 404.
+    ctx = state.get("groups_ctx", {}).get(group_id)
+    if ctx is not None:
+        ctx["reconcile_interval_secs"] = reconcile_interval_secs
+        ctx["debounce_secs"] = debounce_secs
+    log.info("Scan settings for group %s: reconcile=%.0fs debounce=%.0fs",
+             group_id[:8], reconcile_interval_secs, debounce_secs)
+    return {"reconcile_interval_secs": reconcile_interval_secs,
+            "debounce_secs": debounce_secs, "group_id": group_id}
+
+
+# ── Reload ──────────────────────────────────────────────────────────────────
+
+async def reload_config(state: dict) -> dict:
+    """Hot-reload node.toml without dropping connections. Blocks until the
+    reload actually finishes — see start_reload for why the loopback route
+    uses that instead."""
+    reload_fn = state.get("reload_fn")
+    if not reload_fn:
+        raise OpError("Reload not available", status=503)
+    await reload_fn()
+    return {"status": "reloaded"}
+
+
+async def start_reload(state: dict) -> dict:
+    """
+    Same as reload_config, but does not wait for the reload to finish.
+
+    The loopback route uses this one: the Electron bridge caps every call at
+    a fixed 30s (main.js node:call), and hot-loading a brand-new group runs
+    its full initial scan synchronously inside _reload_config_inner()
+    (daemon.py) before that coroutine returns — minutes, not seconds, on a
+    real library (found against a 45 GB group on the same slow disk the
+    StarWars benchmark used). The reload keeps running on the daemon's own
+    event loop either way; add_root/remove_root below already fire it the
+    same way for exactly this reason.
+    """
+    reload_fn = state.get("reload_fn")
+    if not reload_fn:
+        raise OpError("Reload not available", status=503)
+    spawn(reload_fn())
+    return {"status": "reloading"}