"""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), set on the
    node's own machine (loopback API, CLI), 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"}