path: root/packages/meshbay-node/src/meshbay_node/ops/settings.py

"""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"}