path: root/packages/meshbay-cli/src/meshbay_cli/complete.py

"""
Shell completion: what the word under the cursor may become.

`meshbay-cli __complete SHELL CWORD WORD...` prints one candidate per line, for
the scripts `completion` prints. Actions, verbs and options come from the
parser, so they cannot drift from it. Remote paths — `group@owner/root/folder/`
— come from the hub (the groups) and the node (a group's folders and files),
for the account this machine remembers only: nothing is asked, no passphrase
and no device approval, and a node is opened the way that never makes keys
(`session.Rewrap`). What a group holds is cached for two minutes, so the next
Tab is instant. Anything that fails is no candidate, never an error on screen.

`__files__` on a line of its own asks the shell for its own file completion
(a local path: `put`'s sources, `-o`).
"""

import argparse
import asyncio
import contextlib
import json
import logging
import os
import sys
import time

FILES = "__files__"
CACHE_TTL = 120
DEADLINE = 8.0
GLOBAL_VALUES = {"--hub", "-u", "--user", "-p", "--password", "--password-file"}
FILE_OPTIONS = {"--password-file", "-o", "--new-password-file", "--recovery-key-file"}


def unquote(word: str, shell: str) -> str:
    """The word as typed, its quoting taken off: bash's backslashes and quotes,
    PowerShell's quotes and doubled single quote."""
    if len(word) >= 1 and word[0] == "'":
        body = word[1:-1] if len(word) > 1 and word.endswith("'") else word[1:]
        return body.replace("''", "'") if shell == "powershell" else body
    if len(word) >= 1 and word[0] == '"':
        body = word[1:-1] if len(word) > 1 and word.endswith('"') else word[1:]
        return body.replace('\\"', '"').replace('`"', '"')
    if shell == "bash":
        out, escaped = [], False
        for c in word:
            if escaped or c != "\\":
                out.append(c)
                escaped = False
            else:
                escaped = True
        return "".join(out)
    return word


def _subparsers(parser: argparse.ArgumentParser) -> dict[str, argparse.ArgumentParser]:
    sub = next(a for a in parser._actions if isinstance(a, argparse._SubParsersAction))
    return dict(sub.choices)


def _options(parser: argparse.ArgumentParser) -> dict[str, argparse.Action]:
    return {s: a for a in parser._actions for s in a.option_strings}


def plan(words: list[str], cword: int, parser: argparse.ArgumentParser):
    """
    What kind of word is under the cursor: ("words", [candidates]), ("files",),
    ("remote", action), ("groups",), ("invitations",), ("pins",), or ("none",).
    `words[0]` is the program.
    """
    from .__main__ import ACTIONS
    subs = _subparsers(parser)
    current = words[cword] if cword < len(words) else ""
    action, positionals, i = None, [], 1
    top = _options(parser)
    while i < cword:
        w = words[i]
        opts = _options(subs[action]) if action else top
        if w.startswith("-") and w != "-":
            a = opts.get(w.split("=", 1)[0])
            if a is not None and a.nargs != 0 and "=" not in w:
                i += 1                              # its value
        elif action is None and w in ACTIONS:
            action = w
        elif action is not None:
            positionals.append(w)
        i += 1

    prev = words[cword - 1] if cword - 1 >= 1 else ""
    opts = _options(subs[action]) if action else top
    taking = opts.get(prev)
    if prev.startswith("-") and taking is not None and taking.nargs != 0:
        if taking.choices:
            return ("words", [str(c) for c in taking.choices])
        return ("files",) if prev in FILE_OPTIONS else ("none",)

    if current.startswith("-"):
        names = sorted(opts) if action else sorted(top)
        return ("words", [n for n in names if n.startswith("--") or len(n) == 2])
    if action is None:
        return ("words", list(ACTIONS))

    slots = [a for a in subs[action]._actions if not a.option_strings]
    slot = _slot(slots, len(positionals), action)
    if slot is None:
        return ("none",)
    if slot.choices:
        return ("words", [str(c) for c in slot.choices])
    meta = slot.metavar or ""
    if meta == "ACTION":
        return ("words", list(ACTIONS))
    if meta == "SOURCE":
        return ("files+remote", action)
    if meta in ("GROUP",):
        return ("groups",)
    if meta in ("INVITATION", "INVITATION|LINK"):
        return ("invitations",)
    if action == "trust":
        return ("pins",)
    if meta in ("PATH", "DEST") or slot.dest == "path":
        return ("remote", action)
    return ("none",)


def _slot(slots: list[argparse.Action], n: int, action: str):
    """The positional the n-th word fills. `put SOURCE... DEST`: every word is a
    source or the destination, and either may be what is being typed."""
    if action == "put":
        return slots[0]
    for a in slots:
        if a.nargs in ("+", "*"):
            return a
        if n == 0:
            return a
        n -= 1
    return None


# ── The account's groups, and what they hold ─────────────────────────────────

def _cache_path():
    from .client.store import data_dir
    return data_dir() / "completion-cache.json"


def _cache_read() -> dict:
    try:
        return json.loads(_cache_path().read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return {}


def _cache_write(cache: dict) -> None:
    with contextlib.suppress(OSError):
        path = _cache_path()
        path.parent.mkdir(parents=True, exist_ok=True)
        tmp = path.with_suffix(".tmp")
        tmp.write_text(json.dumps(cache), encoding="utf-8")
        os.replace(tmp, path)


def _fresh(entry: dict | None) -> bool:
    return bool(entry) and time.time() - entry.get("at", 0) < CACHE_TTL


async def _account():
    from .client.account import Account, no_passphrase
    from .client.hub import Hub
    from .client.store import Store
    store = Store()
    stored = store.peek()
    if not stored.get("hub"):
        return None
    hub = Hub(os.environ.get("MESHBAY_HUB") or stored["hub"])
    try:
        account = Account.open(hub, None, no_passphrase, store)
    except Exception:
        await hub.aclose()
        return None
    if not account.remembered:
        await hub.aclose()
        return None
    return account


async def _groups(account, cache: dict) -> list[dict]:
    key = f"groups:{account.hub.url}:{account.username}"
    if _fresh(cache.get(key)):
        return cache[key]["rows"]
    from .client.groups import my_groups
    rows = [{k: g.get(k) for k in ("id", "name", "owner_username")}
            for g in await my_groups(account)]
    cache[key] = {"at": time.time(), "rows": rows}
    return rows


async def _tree(account, group: dict, cache: dict):
    from .client.session import Rewrap, connect
    from .client.tree import Tree
    key = f"index:{group['id']}"
    if not _fresh(cache.get(key)):
        nodes = await account.call(account.hub.group_nodes, group["id"])
        nodes = sorted(nodes, key=lambda n: account.pin(n["node_id"]) is None)
        index = None
        for node in nodes:
            try:
                session = await connect(account, group["id"], node, rewrap=Rewrap())
            except Exception:
                continue
            try:
                index = await session.index()
            finally:
                await session.close()
            break
        if index is None:
            return None
        cache[key] = {"at": time.time(), "index": {
            "roots": [{k: r.get(k) for k in ("name", "available", "writable")}
                      for r in index.get("roots", [])],
            "dirs": index.get("dirs", []),
            "entries": [{k: e.get(k) for k in ("id", "name", "path", "size")}
                        for e in index.get("entries", [])]}}
    return Tree.from_index(cache[key]["index"])


async def remote(prefix: str, action: str) -> list[str]:
    from .client.paths import encode_name, handle, resolve_group
    account = await _account()
    if account is None:
        return []
    cache = _cache_read()
    try:
        groups = await _groups(account, cache)
        if "/" not in prefix:
            low = prefix.lower()
            return [handle(g) + "/" for g in groups if handle(g).lower().startswith(low)]
        spec, _, rest = prefix.partition("/")
        group = resolve_group(groups, spec.replace("%2F", "/").replace("%25", "%"))
        tree = await _tree(account, group, cache)
        if tree is None:
            return []
        folder, _, partial = rest.rpartition("/")
        folder_path = folder.replace("%2F", "/").replace("%25", "%")
        kids = tree.children(folder_path) if (not folder or tree.is_dir(folder_path)) else []
        base = f"{spec}/{folder}/" if folder else f"{spec}/"
        out = []
        for kid in kids:
            if not kid.name.lower().startswith(partial.lower()):
                continue
            if not kid.is_dir and action in ("mkdir", "put"):
                continue                     # a destination is a folder
            out.append(base + encode_name(kid.name) + ("/" if kid.is_dir else ""))
        return sorted(out, key=str.lower)
    finally:
        _cache_write(cache)
        await account.close()
        await account.hub.aclose()


async def _invitations() -> list[str]:
    from .client.paths import handle
    account = await _account()
    if account is None:
        return []
    try:
        return [handle(g) for g in await account.call(account.hub.invitations)]
    finally:
        await account.close()
        await account.hub.aclose()


def _pins() -> list[str]:
    from .client.store import Store
    store = Store()
    with contextlib.suppress(Exception):
        if store.load():
            return sorted(store.state.get("pins", {}))
    return []


def candidates(shell: str, cword: int, words: list[str]) -> list[str]:
    from .__main__ import build_parser
    words = [unquote(w, shell) for w in words]
    current = words[cword] if cword < len(words) else ""
    kind = plan(words, cword, build_parser())

    def matching(items):
        return [c for c in items if c.startswith(current)]

    if kind[0] == "words":
        return matching(kind[1])
    if kind[0] == "files":
        return [FILES]
    if kind[0] == "pins":
        return matching(_pins())
    if kind[0] in ("remote", "files+remote", "groups", "invitations"):
        async def go():
            if kind[0] == "invitations":
                return matching(await _invitations())
            found = await remote(current, kind[1] if len(kind) > 1 else "")
            if kind[0] == "groups":
                return [c[:-1] for c in found if c.count("/") == 1]
            return found
        try:
            # The client's loop: on Windows the default one never opens a node.
            from .__main__ import loop_factory
            found = asyncio.run(asyncio.wait_for(go(), DEADLINE), loop_factory=loop_factory())
        except Exception:
            found = []
        return ([FILES] if kind[0] == "files+remote" else []) + found
    return []


def main(argv: list[str]) -> int:
    """`__complete SHELL CWORD WORD...`: never fails, never writes to stderr."""
    logging.disable(logging.CRITICAL)
    try:
        shell, cword, words = argv[0], int(argv[1]), argv[2:]
        out = candidates(shell, cword, words)
    except Exception:
        out = []
    sys.stdout.write("".join(c + "\n" for c in out))
    return 0