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

"""
Node roster — who this node recognises, and which keys are theirs.

The node keeps its own answer to "may this person have the group key", derived from
what the operator authorized locally. It is deliberately NOT derived from the hub:
the hub decides group membership, and a hub that invents an account and mints a
token for it would otherwise collect the GEK on connect. Hub membership is an input
to the decision; it is not the decision.

Three tables:

  identities — one row per person, not per group. Someone paired for one group
               needs no code for the next one on the same node.
  members    — role and status per (group, user).
  invites    — one-time pairing codes, stored as a hash. The code itself exists
               only in the operator's hands and the invitee's.

The code is what binds a public key to an account without asking the hub
(finding H3). See `docs/invite-pairing-v1.md`.
"""

from __future__ import annotations

import hashlib
import json
import logging
import os
import secrets
from datetime import datetime, timedelta, timezone
from pathlib import Path

import aiosqlite

from meshbay_common.paths import fold

log = logging.getLogger(__name__)

# Crockford base32 without I, L, O and U: no character pair a human can confuse
# when reading a code aloud or typing it from a phone screen.
_ALPHABET = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
CODE_LEN = 8                      # 8 × 5 bits = 40 bits of entropy

# Two different rhythms, so two different lifetimes.
#
# An invitation crosses a human conversation: it is sent by mail or message and
# answered whenever the other person next looks. A day is not enough — the code
# dies over a weekend and someone has to be at a browser, with the node online, to
# issue another one.
#
# Operator pairing crosses an SSH session: the code is printed and typed minutes
# later. There is no reason for it to outlive the sitting.
#
# The longer window costs little: a code is single use, bound to one account,
# never seen by the hub, and 40 bits do not fall to guessing in a week against the
# node-wide lockout.
DEFAULT_INVITE_TTL = 7 * 24 * 3600   # seconds — member invitations
DEFAULT_PAIR_TTL = 24 * 3600         # seconds — operator pairing
# A device-add code is read off one screen and typed into another, in one
# sitting. An hour is comfort, not security: the code is bound to the requesting
# keys by its hash, so a longer window widens nothing an attacker can use.
DEFAULT_DEVICE_REQUEST_TTL = 3600

_SCHEMA = """\
-- One row per DEVICE, not per person. A browser and a desktop client are two
-- keys belonging to one account, and `user_id` alone as the key made the second
-- silently overwrite the first (INSERT OR REPLACE). See docs/desktop-client-v1.md §4.
CREATE TABLE IF NOT EXISTS identities (
    user_id     TEXT NOT NULL,
    username    TEXT NOT NULL,
    pk_ed25519  TEXT NOT NULL,
    pk_x25519   TEXT NOT NULL,
    pinned_at   TEXT NOT NULL,
    pinned_via  TEXT NOT NULL,
    label       TEXT NOT NULL DEFAULT '',
    -- Which already-pinned key countersigned this one into existence. Empty for
    -- the first device of an account, which an operator code admitted.
    added_by_pk TEXT NOT NULL DEFAULT '',
    -- The countersignature itself, and the two fields needed to rebuild what it
    -- signed. `added_by_pk` alone says *which* key approved and proves nothing:
    -- a third party cannot check a signature it does not have. And the
    -- transcript binds `nonce_node` — the approving connection's handshake
    -- nonce — so even a stored signature is unverifiable without it.
    --
    -- This is what Tier 2 needs (docs/desktop-client-v1.md §4.8): relayed with
    -- the roster, it lets a member verify for themselves that a second device
    -- belongs to an account whose first device they have already pinned,
    -- instead of taking the node's word. Verified and discarded until
    -- 2026-09-07; a device pinned before that has no evidence and is
    -- trust-on-first-use only, which the client is told rather than left to
    -- infer.
    add_sig     TEXT NOT NULL DEFAULT '',
    add_nonce   TEXT NOT NULL DEFAULT '',
    add_ts      INTEGER NOT NULL DEFAULT 0,
    revoked_at  TEXT,
    PRIMARY KEY (user_id, pk_ed25519)
);

-- A device asking to be added, waiting for an existing one to approve it.
-- `code_hash` binds the code to the keys: sha256(code ‖ pk_ed ‖ pk_x). The
-- approver looks the request up by recomputing that, so a node returning
-- different keys produces no match and the client refuses before signing.
CREATE TABLE IF NOT EXISTS device_requests (
    code_hash  TEXT PRIMARY KEY,
    user_id    TEXT NOT NULL,
    username   TEXT NOT NULL DEFAULT '',
    pk_ed25519 TEXT NOT NULL,
    pk_x25519  TEXT NOT NULL,
    created_at TEXT NOT NULL,
    expires_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS members (
    group_id    TEXT NOT NULL,
    user_id     TEXT NOT NULL,
    role        TEXT NOT NULL,
    status      TEXT NOT NULL,
    approved_by TEXT NOT NULL,
    approved_at TEXT NOT NULL,
    PRIMARY KEY (group_id, user_id)
);

-- Per-group settings the operator changes while the node runs.
--
-- Not node.toml: that file is hand-written, full of comments explaining
-- decisions, and `ops.py` deliberately appends to it rather than round-tripping
-- it through a TOML writer. A setting toggled from a panel has to take effect
-- without an edit to the operator's file and without a restart, so it lives
-- here, where the node already keeps what it decided rather than what it was
-- configured with.
--
-- Absent means default. Nothing writes a row until someone changes something,
-- so an existing node has the same behaviour it had before this table existed.
CREATE TABLE IF NOT EXISTS group_settings (
    group_id TEXT NOT NULL,
    key      TEXT NOT NULL,
    value    TEXT NOT NULL,
    set_by   TEXT NOT NULL DEFAULT '',
    set_at   TEXT NOT NULL DEFAULT '',
    PRIMARY KEY (group_id, key)
);

CREATE TABLE IF NOT EXISTS invites (
    code_hash  TEXT PRIMARY KEY,
    group_id   TEXT NOT NULL,
    user_id    TEXT NOT NULL,
    username   TEXT NOT NULL DEFAULT '',
    role       TEXT NOT NULL,
    created_by TEXT NOT NULL,
    created_at TEXT NOT NULL,
    expires_at TEXT NOT NULL,
    used_at    TEXT
);
"""


def generate_code() -> str:
    """A fresh pairing code, formatted for a human to read out: XXXX-XXXX."""
    raw = "".join(secrets.choice(_ALPHABET) for _ in range(CODE_LEN))
    return f"{raw[:4]}-{raw[4:]}"


def normalize_code(code: str) -> str:
    """
    Fold what a human typed onto what was generated.

    Crockford's rules: case-insensitive, dashes and spaces are decoration, and the
    excluded letters map onto the digits they resemble. Someone reading a code over
    the phone should not be able to get it wrong in a way we could have absorbed.
    """
    out = []
    for ch in code.upper():
        if ch in "- \t":
            continue
        if ch in "IL":
            out.append("1")
        elif ch == "O":
            out.append("0")
        elif ch == "U":
            out.append("V")
        else:
            out.append(ch)
    return "".join(out)


def hash_code(code: str) -> str:
    """
    Store codes hashed: a stolen roster DB must not yield usable invitations.

    SHA-256 rather than a password KDF on purpose — the input is 40 bits of
    uniformly random secret, not a human-chosen string, so there is nothing for a
    slow hash to defend.
    """
    return hashlib.sha256(normalize_code(code).encode()).hexdigest()


def _now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def _iso_in(seconds: int) -> str:
    return (datetime.now(timezone.utc)
            + timedelta(seconds=seconds)).isoformat(timespec="seconds")


class Roster:
    def __init__(self, db_path: Path):
        self._db_path = db_path
        self._db: aiosqlite.Connection | None = None

    async def open(self) -> None:
        self._db_path.parent.mkdir(parents=True, exist_ok=True)
        self._db = await aiosqlite.connect(str(self._db_path))
        self._db.row_factory = aiosqlite.Row
        # WAL: the CLI writes invites (`operator pair`) while the daemon reads them.
        await self._db.execute("PRAGMA journal_mode=WAL")
        await self._db.executescript(_SCHEMA)
        # invites.username was added after the first deployments: the name is what
        # the operator types, and it cannot be recovered from the JWT because the
        # hub does not put one there. CREATE TABLE IF NOT EXISTS will not add a
        # column to a table that already exists.
        async with self._db.execute("PRAGMA table_info(invites)") as cur:
            columns = {r[1] for r in await cur.fetchall()}
        if "username" not in columns:
            await self._db.execute(
                "ALTER TABLE invites ADD COLUMN username TEXT NOT NULL DEFAULT ''")

        await self._migrate_identities_to_devices()
        await self._db.commit()

    async def _migrate_identities_to_devices(self) -> None:
        """
        Widen `identities` from one key per person to one row per device.

        `CREATE TABLE IF NOT EXISTS` leaves an existing table alone, so a roster
        written before device linking still has `user_id` as its sole primary
        key — where a second device would overwrite the first rather than being
        refused. SQLite cannot change a primary key in place, so the table is
        rebuilt.

        Existing pins are carried over untouched and become each account's first
        device. Nobody has to re-pair.
        """
        assert self._db
        async with self._db.execute("PRAGMA table_info(identities)") as cur:
            info = list(await cur.fetchall())
        columns = {r[1] for r in info}
        # `pk` is the column's position in the primary key, 0 when not part of it.
        key_columns = {r[1] for r in info if r[5]}

        # The countersignature evidence (Tier 2), added 2026-09-07. Done
        # **before** the early return below, which fires on any roster already
        # widened to one row per device — i.e. on every node that has run since
        # 2026-08-18, which is all of them. Putting these inside that branch
        # would have meant they never arrived, and the symptom would have been a
        # roster response whose devices all read as unverifiable.
        for column in ("add_sig", "add_nonce"):
            if column not in columns:
                await self._db.execute(
                    f"ALTER TABLE identities ADD COLUMN {column} "
                    f"TEXT NOT NULL DEFAULT ''")
        if "add_ts" not in columns:
            await self._db.execute(
                "ALTER TABLE identities ADD COLUMN add_ts INTEGER NOT NULL "
                "DEFAULT 0")

        if key_columns == {"user_id", "pk_ed25519"} and "revoked_at" in columns:
            return

        log.info("Roster: widening identities to one row per device")
        for column, decl in (("label", "TEXT NOT NULL DEFAULT ''"),
                             ("added_by_pk", "TEXT NOT NULL DEFAULT ''"),
                             ("revoked_at", "TEXT")):
            if column not in columns:
                await self._db.execute(
                    f"ALTER TABLE identities ADD COLUMN {column} {decl}")

        if key_columns != {"user_id", "pk_ed25519"}:
            await self._db.execute("ALTER TABLE identities RENAME TO identities_old")
            await self._db.executescript(_SCHEMA)
            await self._db.execute(
                "INSERT OR IGNORE INTO identities "
                "(user_id, username, pk_ed25519, pk_x25519, pinned_at, "
                " pinned_via, label, added_by_pk, revoked_at) "
                "SELECT user_id, username, pk_ed25519, pk_x25519, pinned_at, "
                "       pinned_via, label, added_by_pk, revoked_at "
                "FROM identities_old")
            await self._db.execute("DROP TABLE identities_old")
            log.info("Roster: identities rebuilt, existing pins preserved")

    async def close(self) -> None:
        if self._db:
            await self._db.close()
            self._db = None

    # ── Identities ───────────────────────────────────────────────────────────

    # How many devices one person may hold on this node. A chain of devices
    # inherits the weakness of its weakest ancestor — whoever cracks a browser's
    # keypair bundle can add one — so the answer to "how many" is visibility and
    # a ceiling, not cryptography.
    MAX_DEVICES_PER_USER = 5

    async def pin_identity(
        self,
        user_id: str,
        username: str,
        pk_ed25519: str,
        pk_x25519: str,
        via: str,
        *,
        label: str = "",
        added_by_pk: str = "",
        add_sig: str = "",
        add_nonce: str = "",
        add_ts: int = 0,
    ) -> None:
        """
        Record a device for an account.

        `INSERT OR REPLACE` on (user_id, pk_ed25519) now updates *that device*
        rather than overwriting whatever key the person had before — which is
        what it did while `user_id` was the whole primary key, silently, and
        would have become a hole the moment a second device was legitimate.
        """
        assert self._db
        await self._db.execute(
            "INSERT OR REPLACE INTO identities "
            "(user_id, username, pk_ed25519, pk_x25519, pinned_at, pinned_via, "
            " label, added_by_pk, add_sig, add_nonce, add_ts, revoked_at) "
            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL)",
            (user_id, username, pk_ed25519, pk_x25519, _now(), via,
             label, added_by_pk, add_sig, add_nonce, add_ts),
        )
        await self._db.commit()

    async def get_identity(self, user_id: str) -> dict | None:
        """
        This account's oldest live device.

        Kept for callers that only need "is this person known here" — the
        operator pin, `status`, attribution. Anything deciding whether a *key*
        is admitted must use `find_device`, or a second device is refused where
        the first is not.
        """
        assert self._db
        async with self._db.execute(
            "SELECT * FROM identities WHERE user_id = ? AND revoked_at IS NULL "
            "ORDER BY pinned_at LIMIT 1", (user_id,)
        ) as cur:
            row = await cur.fetchone()
        return dict(row) if row else None

    async def find_device(self, user_id: str, pk_ed25519: str) -> dict | None:
        """The device with this exact key, if it is live. None if revoked."""
        assert self._db
        async with self._db.execute(
            "SELECT * FROM identities WHERE user_id = ? AND pk_ed25519 = ? "
            "AND revoked_at IS NULL", (user_id, pk_ed25519)
        ) as cur:
            row = await cur.fetchone()
        return dict(row) if row else None

    async def list_devices(self, user_id: str,
                           include_revoked: bool = False) -> list[dict]:
        assert self._db
        sql = "SELECT * FROM identities WHERE user_id = ?"
        if not include_revoked:
            sql += " AND revoked_at IS NULL"
        async with self._db.execute(sql + " ORDER BY pinned_at",
                                    (user_id,)) as cur:
            return [dict(r) for r in await cur.fetchall()]

    async def revoke_device(self, user_id: str, pk_ed25519: str) -> bool:
        """
        Retire one device, leaving the account's others alone.

        Marked rather than deleted: a revoked key must stay refused, and a row
        that is gone is a key the node would happily pin again on the next
        device-add — which is the laptop somebody just reported lost.
        """
        assert self._db
        cur = await self._db.execute(
            "UPDATE identities SET revoked_at = ? "
            "WHERE user_id = ? AND pk_ed25519 = ? AND revoked_at IS NULL",
            (_now(), user_id, pk_ed25519))
        await self._db.commit()
        return cur.rowcount > 0

    async def unpin(self, user_id: str) -> bool:
        """
        Forget an account entirely — every device it holds.

        Deliberately all of them: `member unpin` is what an operator runs when
        someone must start over, and leaving one device behind would let the
        person walk back in with a key the operator meant to forget.
        """
        assert self._db
        cur = await self._db.execute(
            "DELETE FROM identities WHERE user_id = ?", (user_id,))
        await self._db.execute(
            "DELETE FROM members WHERE user_id = ?", (user_id,))
        await self._db.commit()
        return cur.rowcount > 0

    async def group_devices(self, group_id: str) -> list[dict]:
        """
        Every live device of every active member of one group, with the evidence
        that admitted it.

        For Tier 2 (`docs/desktop-client-v1.md` §4.8), and therefore
        **member-visible** — unlike `list_identities`, which answers the
        operator. Two consequences of that, and both are the price of the
        feature rather than oversights:

        - it tells every member of a group how many devices each other member
          holds, and their public keys. It stays inside the group, and the hub
          is not involved;
        - it is scoped to *this* group. A person in two groups on one node is
          not disclosed to the second by being in the first.

        "Member of this group" is `_MEMBER_OF_GROUP`, shared with
        `is_authorized` — **the operator belongs to every group this node
        hosts**, with their authority recorded under an empty group_id. Spelling
        that out a second time here is exactly what went wrong first time:
        the operator was missing from the roster, so every one of their messages
        reached other members as a key nobody could vouch for.

        `DISTINCT` because an operator who is *also* an explicit member of the
        group matches both halves of that clause.

        `add_sig`/`add_nonce`/`add_ts` are empty for a device pinned before the
        evidence was kept, and for the first device of any account — which an
        operator code admitted, not a countersignature. Both read as
        "trust on first use" to a client, which is what they are; the client
        must not silently treat an absent signature as a valid one.
        """
        assert self._db
        async with self._db.execute(
            f"SELECT DISTINCT i.user_id, i.username, i.pk_ed25519, i.pk_x25519, "
            f"       i.added_by_pk, i.add_sig, i.add_nonce, i.add_ts, i.pinned_at "
            f"FROM identities i "
            f"JOIN members m ON m.user_id = i.user_id "
            f"WHERE i.revoked_at IS NULL AND m.{self._MEMBER_OF_GROUP} "
            f"ORDER BY i.user_id, i.pinned_at", (group_id,)
        ) as cur:
            rows = await cur.fetchall()
        return [
            {"user_id": r[0], "username": r[1], "pk_ed25519": r[2],
             "pk_x25519": r[3], "added_by_pk": r[4], "add_sig": r[5],
             "add_nonce": r[6], "add_ts": r[7], "pinned_at": r[8]}
            for r in rows
        ]

    async def list_identities(self) -> list[dict]:
        assert self._db
        async with self._db.execute(
            "SELECT * FROM identities WHERE revoked_at IS NULL "
            "ORDER BY pinned_at"
        ) as cur:
            return [dict(r) for r in await cur.fetchall()]

    # ── Device requests ──────────────────────────────────────────────────────

    async def file_device_request(
        self, user_id: str, username: str, pk_ed25519: str, pk_x25519: str,
        code_hash: str, ttl: int = DEFAULT_DEVICE_REQUEST_TTL,
    ) -> str:
        """
        Record a device waiting to be approved. Returns its expiry.

        The node stores only `code_hash`, which the new device computed over the
        code **and its own keys**. That binding is what stops the node itself
        from substituting a key: an approver recomputes the hash from the code
        they typed and the keys they were handed, and a mismatch means no
        request is found.
        """
        assert self._db
        expires = _iso_in(ttl)
        await self._db.execute(
            "INSERT OR REPLACE INTO device_requests "
            "(code_hash, user_id, username, pk_ed25519, pk_x25519, created_at, "
            " expires_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
            (code_hash, user_id, username, pk_ed25519, pk_x25519, _now(), expires))
        await self._db.commit()
        return expires

    async def take_device_request(self, code_hash: str,
                                  user_id: str) -> dict | None:
        """
        Claim a pending request by its hash, for this account only.

        Single use and scoped to the account: a request filed for one person
        cannot be redeemed by another even with the code, and a code that has
        been spent is gone.
        """
        assert self._db
        async with self._db.execute(
            "SELECT * FROM device_requests WHERE code_hash = ? AND user_id = ? "
            "AND expires_at > ?", (code_hash, user_id, _now())
        ) as cur:
            row = await cur.fetchone()
        if row is None:
            return None
        await self._db.execute(
            "DELETE FROM device_requests WHERE code_hash = ?", (code_hash,))
        await self._db.commit()
        return dict(row)

    async def list_device_requests(self, user_id: str) -> list[dict]:
        """
        This account's pending requests, hashes included.

        The hash is what the approver matches against, so it has to travel.
        Handing it out is safe: it is `sha256(code ‖ keys)` over 40 bits of
        secret the node does not hold, and knowing the code authorizes nothing
        on its own — only a countersignature by an already-pinned key does.
        """
        assert self._db
        async with self._db.execute(
            "SELECT * FROM device_requests WHERE user_id = ? AND expires_at > ? "
            "ORDER BY created_at", (user_id, _now())
        ) as cur:
            return [dict(r) for r in await cur.fetchall()]

    async def pending_device_requests(self, user_id: str) -> int:
        """How many this account has waiting. For display and for a ceiling."""
        assert self._db
        async with self._db.execute(
            "SELECT COUNT(*) AS n FROM device_requests WHERE user_id = ? "
            "AND expires_at > ?", (user_id, _now())
        ) as cur:
            row = await cur.fetchone()
        return int(row["n"]) if row else 0

    # ── Authority ────────────────────────────────────────────────────────────

    async def operator_pks(self) -> list[str]:
        """
        Base64 Ed25519 keys allowed to authorize admin operations on this node.

        Read fresh on every check rather than cached: an unpin must take effect at
        once, and this runs only on admin operations, which are rare.
        """
        assert self._db
        async with self._db.execute(
            "SELECT i.pk_ed25519 FROM identities i "
            "JOIN members m ON m.user_id = i.user_id "
            "WHERE m.role = 'operator' AND m.status = 'active' "
            # An operator with two browsers has two keys and both may sign; a
            # retired one must not.
            "AND i.revoked_at IS NULL"
        ) as cur:
            return [r["pk_ed25519"] for r in await cur.fetchall()]

    async def has_operator(self) -> bool:
        return bool(await self.operator_pks())

    # Who counts as a member of a group, in SQL, in **one** place.
    #
    # The operator's authority is node-wide and is stored with an empty
    # group_id, so "belongs to this group" is not `group_id = ?`. Writing that
    # clause a second time is how `group_devices` came to omit the operator
    # from the roster it relays — and the symptom was a member seeing "this
    # account is using a key you have not seen before" on every single message
    # from the person running the node. Found on a live pair of machines, not
    # by a test.
    _MEMBER_OF_GROUP = ("status = 'active' AND "
                        "(group_id = ? OR (group_id = '' AND role = 'operator'))")

    async def is_authorized(self, group_id: str, user_id: str) -> bool:
        """
        May this person be handed the group key?

        The node's own answer, not the hub's. Hub membership is what lets someone
        reach the node; this is what decides whether the key is wrapped for them —
        otherwise a hub that invents an account and mints a token for it would be
        served the GEK on connect.

        An operator is authorized for every group this node hosts: their authority
        is node-wide and is recorded with an empty group_id.
        """
        assert self._db
        async with self._db.execute(
            f"SELECT 1 FROM members WHERE user_id = ? AND {self._MEMBER_OF_GROUP} "
            f"LIMIT 1",
            (user_id, group_id),
        ) as cur:
            return await cur.fetchone() is not None

    # ── Members ──────────────────────────────────────────────────────────────

    async def set_member(
        self,
        group_id: str,
        user_id: str,
        role: str,
        status: str,
        approved_by: str,
    ) -> None:
        assert self._db
        await self._db.execute(
            "INSERT OR REPLACE INTO members "
            "(group_id, user_id, role, status, approved_by, approved_at) "
            "VALUES (?, ?, ?, ?, ?, ?)",
            (group_id, user_id, role, status, approved_by, _now()),
        )
        await self._db.commit()

    async def get_member(self, group_id: str, user_id: str) -> dict | None:
        assert self._db
        async with self._db.execute(
            "SELECT * FROM members WHERE group_id = ? AND user_id = ?",
            (group_id, user_id),
        ) as cur:
            row = await cur.fetchone()
        return dict(row) if row else None

    async def list_members(self, group_id: str | None = None) -> list[dict]:
        assert self._db
        sql = (
            "SELECT m.*, i.username, i.pk_ed25519, i.pinned_at, i.pinned_via "
            "FROM members m LEFT JOIN identities i ON i.user_id = m.user_id"
        )
        args: tuple = ()
        if group_id is not None:
            sql += " WHERE m.group_id = ?"
            args = (group_id,)
        async with self._db.execute(sql + " ORDER BY m.approved_at", args) as cur:
            return [dict(r) for r in await cur.fetchall()]

    async def set_status(self, group_id: str, user_id: str, status: str) -> bool:
        assert self._db
        cur = await self._db.execute(
            "UPDATE members SET status = ? WHERE group_id = ? AND user_id = ?",
            (status, group_id, user_id),
        )
        await self._db.commit()
        return cur.rowcount > 0

    # ── Invites ──────────────────────────────────────────────────────────────

    # ── Group settings ──────────────────────────────────────────────────────

    # Whether a root is ejected. Runtime state, one key per root, keyed by the
    # *folded* name so it agrees with the case-insensitive comparison the rest
    # of the root code makes. It lives here rather than in node.toml because it
    # is not configuration — an operator's hand-written config file should not
    # be rewritten because a USB drive was unplugged — and it has to survive a
    # restart, or the rescan that follows reads an empty mount point as an
    # erased library, which is the whole thing eject exists to prevent.
    SETTING_ROOT_EJECTED_PREFIX = "root_ejected:"

    @classmethod
    def root_ejected_key(cls, root_name: str) -> str:
        return cls.SETTING_ROOT_EJECTED_PREFIX + fold(root_name)

    async def set_root_ejected(self, group_id: str, root_name: str,
                               ejected: bool, set_by: str = "") -> None:
        await self.set_setting(group_id, self.root_ejected_key(root_name),
                               "1" if ejected else "0", set_by)

    async def ejected_roots(self, group_id: str) -> set[str]:
        """
        The folded names of this group's ejected roots.

        Matched in Python rather than with `LIKE 'root_ejected:%'`: `_` is a
        single-character wildcard there, so that pattern also matches keys this
        does not own. A group has a handful of settings rows, so reading them
        all costs nothing and the prefix test is then exact.
        """
        prefix = self.SETTING_ROOT_EJECTED_PREFIX
        async with self._db.execute(
                "SELECT key, value FROM group_settings WHERE group_id = ?",
                (group_id,)) as cur:
            rows = await cur.fetchall()
        return {r["key"][len(prefix):] for r in rows
                if r["key"].startswith(prefix) and r["value"] == "1"}

    async def get_setting(self, group_id: str, key: str,
                          default: str | None = None) -> str | None:
        async with self._db.execute(
                "SELECT value FROM group_settings WHERE group_id = ? AND key = ?",
                (group_id, key)) as cur:
            row = await cur.fetchone()
        return row["value"] if row else default

    async def set_setting(self, group_id: str, key: str, value: str,
                          set_by: str = "") -> None:
        await self._db.execute(
            "INSERT INTO group_settings (group_id, key, value, set_by, set_at) "
            "VALUES (?, ?, ?, ?, ?) "
            "ON CONFLICT(group_id, key) DO UPDATE SET "
            "value = excluded.value, set_by = excluded.set_by, "
            "set_at = excluded.set_at",
            (group_id, key, value, set_by, _now()))
        await self._db.commit()

    # Which group "applications" (Chat, Files, and whatever registers later in
    # apps.js) are shown to members. Unset means every app that exists — an
    # existing group's tabs must not disappear because a node was upgraded.
    SETTING_ENABLED_APPS = "enabled_apps"
    DEFAULT_APPS = ("chat", "files")

    # How many transfers one member may run at once in this group. Unset means
    # the node's default (transfers.DEFAULT_MAX_PER_MEMBER), never "unlimited":
    # a group that predates this coming back unlimited would leave the
    # node-wide pool as the only control.
    SETTING_TRANSFER_LIMITS = "transfer_limits"

    async def transfer_limits(self, group_id: str) -> dict[str, int]:
        """{"download": n, "upload": n}, or {} when the operator has not said."""
        value = await self.get_setting(group_id, self.SETTING_TRANSFER_LIMITS)
        if value is None:
            return {}
        try:
            raw = json.loads(value)
        except (ValueError, TypeError):
            return {}
        out: dict[str, int] = {}
        for kind in ("download", "upload"):
            if isinstance(raw.get(kind), int) and raw[kind] >= 1:
                out[kind] = raw[kind]
        return out

    async def set_transfer_limits(self, group_id: str, limits: dict[str, int],
                                  set_by: str = "") -> dict[str, int]:
        clean = {k: max(1, int(v)) for k, v in limits.items()
                 if k in ("download", "upload")}
        await self.set_setting(group_id, self.SETTING_TRANSFER_LIMITS,
                               json.dumps(clean), set_by)
        return clean

    async def enabled_apps(self, group_id: str) -> list[str]:
        value = await self.get_setting(group_id, self.SETTING_ENABLED_APPS)
        if value is None:
            apps = list(self.DEFAULT_APPS)
        else:
            try:
                apps = list(json.loads(value))
            except (ValueError, TypeError):
                apps = list(self.DEFAULT_APPS)
        if "files" not in apps:
            apps.insert(0, "files")
        return apps

    async def set_enabled_apps(self, group_id: str, apps: list[str],
                               set_by: str = "") -> list[str]:
        await self.set_setting(group_id, self.SETTING_ENABLED_APPS,
                               json.dumps(sorted(apps)), set_by)
        return apps

    # The TMDB credential and query language are one operator's budget, not a
    # per-group concern (docs/mediacenter.md §5.5) — stored under the
    # group_id="" sentinel, the same precedent as `roster.get_member("",
    # user_id)` authorizing the operator node-wide (desktop-client-v1.md
    # §6.3). Unset means "the shipped default token, TMDB's own default
    # language" — the same "absent means the old behaviour" discipline
    # enabled_apps already follows.
    #
    # Whether TMDB is used *at all*, though, is per-group (moved off the
    # node-wide sentinel below, 2026-08-24): an operator running a real media
    # library alongside test/demo groups wants outbound TMDB traffic for the
    # one group that needs it, not all of them just because one node process
    # serves both. See SETTING_TMDB_ENABLED's own per-group methods further
    # down, next to video_root.
    SETTING_TMDB_TOKEN = "tmdb_api_token"
    # A TMDB language tag (e.g. "fr-FR") — one for the whole node, same
    # reasoning as the token: one shared cache, not a per-viewer request.
    # Unset means TMDB's own default (English) rather than this node
    # guessing one.
    SETTING_TMDB_LANGUAGE = "tmdb_language"
    NODE_WIDE_GROUP_ID = ""

    async def tmdb_config(self) -> tuple[str | None, str | None]:
        """Returns (custom_token_or_None, language_or_None)."""
        token = await self.get_setting(self.NODE_WIDE_GROUP_ID, self.SETTING_TMDB_TOKEN)
        language = await self.get_setting(self.NODE_WIDE_GROUP_ID, self.SETTING_TMDB_LANGUAGE)
        return (token or None), (language or None)

    async def set_tmdb_config(self, token: str | None = None,
                              language: str | None = None, set_by: str = "") -> None:
        if token is not None:
            await self.set_setting(self.NODE_WIDE_GROUP_ID, self.SETTING_TMDB_TOKEN,
                                   token, set_by)
        if language is not None:
            await self.set_setting(self.NODE_WIDE_GROUP_ID, self.SETTING_TMDB_LANGUAGE,
                                   language, set_by)

    # ── App directories ─────────────────────────────────────────────────────
    #
    # Which folder(s) inside the group's shared roots each application uses as
    # its entry point. One storage shape for every app, keyed by the app's own
    # name, so adding an application needs no change here at all — that is the
    # whole point of the plugin architecture (docs/refactor-groups.md §1.6).
    #
    # Always a JSON list, even for an app that only ever wants one directory.
    # Two shapes for one idea is how `video_root` (scalar) and `photo_roots`
    # (list) ended up needing separate ops, separate MNP messages and separate
    # settings widgets to say the same thing.
    #
    # Empty/unset means nothing configured yet, and every app reads that as
    # "show nothing until an operator has chosen" rather than "the whole group
    # index" — the discipline video_root established, kept.
    SETTING_APP_DIRS_SUFFIX = "_directories"

    # What each app's directories used to be stored under, before they were
    # one shape. Read as a fallback so an existing node keeps working with no
    # migration step: the legacy key is never written again, and the first
    # save through the new path leaves it behind.
    # Keyed by the *registry* name the app is known by everywhere else
    # (apps.js, ALLOWED_APPS, enabled_apps) — which for Music is "music", while
    # its old setting was called `audio_root`. One identifier per app, and the
    # place the two names meet is this table and nowhere else.
    LEGACY_DIR_KEYS = {
        "video": ("video_root", "scalar"),
        "music": ("audio_root", "scalar"),
        "photo": ("photo_roots", "list"),
    }

    # The name each app's directories are *also* published under, for readers
    # that predate the list — the handshake ack's `video_root`, and the group
    # context the ack builds from. Derived from the list, never stored beside
    # it, so the two cannot disagree; the shape says how to derive it.
    CTX_ALIASES = {
        "video": ("video_root", "scalar"),
        "music": ("audio_root", "scalar"),
        "photo": ("photo_roots", "list"),
        "chat": ("chat_directory", "scalar"),
    }

    @classmethod
    def app_dirs_key(cls, app_key: str) -> str:
        return f"{app_key}{cls.SETTING_APP_DIRS_SUFFIX}"

    @classmethod
    def ctx_alias(cls, app_key: str, directories: list[str]) -> tuple[str, object] | None:
        """The (name, value) an app's directories are also published under."""
        alias = cls.CTX_ALIASES.get(app_key)
        if not alias:
            return None
        name, shape = alias
        if shape == "list":
            return name, list(directories)
        return name, (directories[0] if directories else "")

    async def app_directories(self, group_id: str, app_key: str) -> list[str]:
        value = await self.get_setting(group_id, self.app_dirs_key(app_key))
        if value is not None:
            try:
                return [str(p) for p in json.loads(value)]
            except (ValueError, TypeError):
                return []

        legacy = self.LEGACY_DIR_KEYS.get(app_key)
        if not legacy:
            return []
        key, shape = legacy
        raw = await self.get_setting(group_id, key)
        if raw is None:
            return []
        if shape == "scalar":
            return [raw] if raw else []
        try:
            return [str(p) for p in json.loads(raw)]
        except (ValueError, TypeError):
            return []

    async def set_app_directories(self, group_id: str, app_key: str,
                                  paths: list[str], set_by: str = "") -> list[str]:
        clean = sorted({str(p).strip("/") for p in paths if str(p).strip("/")})
        await self.set_setting(group_id, self.app_dirs_key(app_key),
                               json.dumps(clean), set_by)
        return clean

    # ── Chat ────────────────────────────────────────────────────────────────

    # Whether the node fetches a page's title/preview when a member posts a
    # link. Outbound third-party traffic on the operator's connection, from a
    # message they did not write, so it is theirs to switch off — the same
    # reasoning as the per-group TMDB switch. Unset means on, because that is
    # what the node did before this existed.
    SETTING_CHAT_LINK_PREVIEW = "chat_link_preview"

    async def chat_link_preview(self, group_id: str) -> bool:
        value = await self.get_setting(group_id, self.SETTING_CHAT_LINK_PREVIEW, "1")
        return value != "0"

    async def set_chat_link_preview(self, group_id: str, enabled: bool,
                                    set_by: str = "") -> bool:
        await self.set_setting(group_id, self.SETTING_CHAT_LINK_PREVIEW,
                               "1" if enabled else "0", set_by)
        return enabled

    # Whether TMDB lookups run for this group at all — per-group, unlike the
    # token/language above: one node process can share a real media library
    # group and several test/demo groups, and outbound TMDB traffic (and API
    # quota) for the demo groups is not something turning it on for the real
    # one should imply. Unset means on, same "absent means the old
    # behaviour" discipline as everything else here — a node that predates
    # this setting keeps working exactly as before for every group.
    SETTING_TMDB_ENABLED = "tmdb_enabled"

    async def tmdb_enabled(self, group_id: str) -> bool:
        return (await self.get_setting(group_id, self.SETTING_TMDB_ENABLED, "1")) != "0"

    async def set_tmdb_enabled(self, group_id: str, enabled: bool, set_by: str = "") -> None:
        await self.set_setting(group_id, self.SETTING_TMDB_ENABLED,
                               "1" if enabled else "0", set_by)

    # MusicBrainz contact is now the node owner's hub email, resolved at
    # login (musicbrainz.py) — no roster setting needed.

    # Whether MusicBrainz lookups run for this group at all — per-group from
    # the start (unlike tmdb_enabled, which started node-wide and moved
    # per-group later once the lesson was already learned). Unset means on,
    # same "absent means the old behaviour" discipline as everything else.
    SETTING_MUSICBRAINZ_ENABLED = "musicbrainz_enabled"

    async def musicbrainz_enabled(self, group_id: str) -> bool:
        return (await self.get_setting(group_id, self.SETTING_MUSICBRAINZ_ENABLED, "1")) != "0"

    async def set_musicbrainz_enabled(self, group_id: str, enabled: bool, set_by: str = "") -> None:
        await self.set_setting(group_id, self.SETTING_MUSICBRAINZ_ENABLED,
                               "1" if enabled else "0", set_by)

    # How often the indexer's reconciliation backstop runs, and how long it
    # waits after the last change on a file before hashing it. Unset means
    # the indexer's own defaults — an existing group's behaviour must not
    # change because a node was upgraded. See indexer.py DirectoryIndexer
    # for what these actually do and why the defaults are what they are.
    SETTING_RECONCILE_INTERVAL = "reconcile_interval_secs"
    SETTING_DEBOUNCE_SECS = "debounce_secs"

    async def scan_settings(self, group_id: str) -> dict:
        # Imported here, not at module load: roster.py is loaded before the
        # indexer package during startup, and this is the only place the two
        # need each other's names.
        from meshbay_node.indexer.indexer import DirectoryIndexer

        reconcile = await self.get_setting(group_id, self.SETTING_RECONCILE_INTERVAL)
        debounce = await self.get_setting(group_id, self.SETTING_DEBOUNCE_SECS)
        return {
            "reconcile_interval_secs": (
                float(reconcile) if reconcile is not None
                else DirectoryIndexer.DEFAULT_RECONCILE_SECS),
            "debounce_secs": (
                float(debounce) if debounce is not None
                else DirectoryIndexer.DEFAULT_DEBOUNCE_SECS),
        }

    async def set_scan_settings(self, group_id: str, reconcile_interval_secs: float,
                                debounce_secs: float, set_by: str = "") -> dict:
        await self.set_setting(group_id, self.SETTING_RECONCILE_INTERVAL,
                               str(float(reconcile_interval_secs)), set_by)
        await self.set_setting(group_id, self.SETTING_DEBOUNCE_SECS,
                               str(float(debounce_secs)), set_by)
        return await self.scan_settings(group_id)

    # ── Node-wide daemon settings ───────────────────────────────────────────
    # Same pattern as TMDB config: stored under NODE_WIDE_GROUP_ID.
    # On startup, node.toml values are the defaults; the roster override
    # takes precedence at runtime. Changing a setting writes to both
    # roster.db (immediate) and node.toml (survives a DB wipe).
    SETTING_INVITE_TTL = "invite_ttl_hours"
    SETTING_PAIR_TTL = "pair_ttl_hours"
    SETTING_DEVICE_TTL = "device_request_ttl_minutes"
    SETTING_MAX_STREAMS = "max_concurrent_streams"
    SETTING_MAX_DOWNLOADS = "max_concurrent_downloads"
    SETTING_MAX_UPLOADS = "max_concurrent_uploads"
    SETTING_TRANSCODE = "transcode_incompatible_video"
    SETTING_STUN_SERVERS = "stun_servers"
    SETTING_ICE_INTERFACES = "ice_interfaces"

    async def node_settings(self, defaults: dict) -> dict:
        """Current effective settings: roster override if present, else config default."""
        import json as _json
        result = {}
        for key, setting in [
            ("invite_ttl_hours", self.SETTING_INVITE_TTL),
            ("pair_ttl_hours", self.SETTING_PAIR_TTL),
            ("device_request_ttl_minutes", self.SETTING_DEVICE_TTL),
            ("max_concurrent_streams", self.SETTING_MAX_STREAMS),
            ("max_concurrent_downloads", self.SETTING_MAX_DOWNLOADS),
            ("max_concurrent_uploads", self.SETTING_MAX_UPLOADS),
            ("transcode_incompatible_video", self.SETTING_TRANSCODE),
        ]:
            stored = await self.get_setting(self.NODE_WIDE_GROUP_ID, setting)
            if stored is not None:
                if key == "transcode_incompatible_video":
                    result[key] = stored != "0"
                else:
                    result[key] = int(stored)
            else:
                result[key] = defaults.get(key)
        for list_key, setting in [
            ("stun_servers", self.SETTING_STUN_SERVERS),
            ("ice_interfaces", self.SETTING_ICE_INTERFACES),
        ]:
            stored = await self.get_setting(self.NODE_WIDE_GROUP_ID, setting)
            if stored is not None:
                try:
                    result[list_key] = _json.loads(stored)
                except (ValueError, TypeError):
                    result[list_key] = defaults.get(list_key, [])
            else:
                result[list_key] = defaults.get(list_key, [])
        return result

    async def set_node_setting(self, key: str, value: str,
                               set_by: str = "") -> None:
        await self.set_setting(self.NODE_WIDE_GROUP_ID, key, value, set_by)

    async def create_invite(
        self,
        group_id: str,
        user_id: str,
        role: str,
        created_by: str,
        ttl: int = DEFAULT_INVITE_TTL,
        username: str = "",
    ) -> str:
        """
        Issue a one-time code. Returns it in the clear — this is the only moment it
        exists outside the operator's hands; only its hash is kept.

        Any earlier unused invite for the same person and group is dropped, so
        re-inviting supersedes rather than accumulating valid codes.
        """
        assert self._db
        await self._db.execute(
            "DELETE FROM invites WHERE group_id = ? AND user_id = ? AND used_at IS NULL",
            (group_id, user_id),
        )
        code = generate_code()
        expires = datetime.now(timezone.utc) + timedelta(seconds=ttl)
        await self._db.execute(
            "INSERT INTO invites (code_hash, group_id, user_id, username, role, "
            "created_by, created_at, expires_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
            (hash_code(code), group_id, user_id, username, role, created_by, _now(),
             expires.isoformat(timespec="seconds")),
        )
        await self._db.commit()
        return code

    async def consume_invite(self, code: str, user_id: str) -> dict | None:
        """
        Redeem a code for `user_id`, or return None.

        Single use is enforced by the UPDATE's WHERE clause: two connections racing
        the same code cannot both see `used_at IS NULL`, so exactly one wins.
        """
        assert self._db
        code_hash = hash_code(code)
        async with self._db.execute(
            "SELECT * FROM invites WHERE code_hash = ?", (code_hash,)
        ) as cur:
            row = await cur.fetchone()
        if not row:
            return None

        invite = dict(row)
        if invite["used_at"] is not None:
            return None
        # A code is valid for exactly one account, so a leaked code cannot be
        # redeemed by whoever finds it first.
        if invite["user_id"] != user_id:
            return None
        if datetime.fromisoformat(invite["expires_at"]) < datetime.now(timezone.utc):
            return None

        cur = await self._db.execute(
            "UPDATE invites SET used_at = ? WHERE code_hash = ? AND used_at IS NULL",
            (_now(), code_hash),
        )
        await self._db.commit()
        if cur.rowcount == 0:
            return None
        return invite

    async def list_invites(self, include_used: bool = False) -> list[dict]:
        assert self._db
        sql = "SELECT * FROM invites"
        if not include_used:
            sql += " WHERE used_at IS NULL"
        async with self._db.execute(sql + " ORDER BY created_at") as cur:
            return [dict(r) for r in await cur.fetchall()]

    async def purge_expired(self) -> int:
        assert self._db
        now = _now()
        cur = await self._db.execute(
            "DELETE FROM invites WHERE used_at IS NULL AND expires_at < ?",
            (now,),
        )
        removed = cur.rowcount
        # Device requests expire too, and an abandoned one left lying about is
        # a row an approver could still be shown.
        cur = await self._db.execute(
            "DELETE FROM device_requests WHERE expires_at < ?", (now,))
        removed += cur.rowcount
        await self._db.commit()
        return removed


async def open_roster(data_dir: Path) -> Roster:
    roster = Roster(data_dir / "roster.db")
    await roster.open()
    return roster


def write_code_file(data_dir: Path, code: str, expires_at: str,
                    name: str = "pair-code") -> Path:
    """
    Leave the code in a file as well as on stdout.

    An operator working over SSH may not be able to copy out of their terminal,
    and a code that can only be read off a scrolled-away screen is a dead end.
    Pairing and invitation codes go to different files so one does not overwrite
    the other.
    """
    path = data_dir / name
    path.parent.mkdir(parents=True, exist_ok=True)
    from meshbay_node.platform import chmod_private
    path.write_text(f"{code}\nexpires {expires_at}\n", encoding="utf-8", newline="\n")
    chmod_private(path)
    return path