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

"""
The sign-in screen, and the main screen: two panes and the transfers.

F5 copies the selection to the other pane's folder — a download or an upload,
with the rules `get` and `put` follow: an upload is decided whole before it
starts (names the node refuses stop it, names already there are asked about
once for the batch), nothing on either side is replaced without being asked,
and a name from a group is made writable here before it is written.

Delete removes the selection from a group with the rules `rm` follows: files
only, every one of them this account's to delete (uploaded by it, or this key
the node's operator) or none is, then asked once for the batch.
"""

import asyncio
import contextlib
import os
import threading
from pathlib import Path

from meshbay_common.paths import long_path
from rich.text import Text
from textual.app import ComposeResult
from textual.binding import Binding
from textual.containers import Horizontal, Vertical
from textual.screen import Screen
from textual.widgets import Button, Footer, Input, Static

from ..client.account import Account
from ..client.names import UnsafeName, local_path
from ..client.paths import handle
from ..client.pool import Pool
from ..client.remove import delete_file, may_delete
from ..client.store import FILE_KEY_NOTICE, KEY_FILE
from ..client.transfers import PART_SUFFIX, TransferError, create_dir, download, upload
from ..client.tree import Tree
from ..client.upload_plan import plan_upload, refused_names, taken_names
from ..errors import CliError
from ..output import safe
from .panes import LocalPane, RemotePane
from .widgets import Choice, TransferStrip

PER_GROUP = 2


class SignInScreen(Screen):
    DEFAULT_CSS = """
    SignInScreen { align: center middle; }
    SignInScreen > Vertical { width: 64; height: auto; border: thick $accent;
                              padding: 1 2; background: $surface; }
    SignInScreen Input { margin-bottom: 1; }
    SignInScreen #error { color: $error; height: auto; }
    """

    def compose(self) -> ComposeResult:
        ctx = self.app.ctx
        with Vertical():
            yield Static(Text("Sign in to MeshBay"))
            yield Input(value=ctx.hub_url, placeholder="hub", id="hub")
            yield Input(value=ctx.username or ctx.stored_username or "",
                        placeholder="username", id="user")
            yield Input(placeholder="passphrase", password=True, id="pass")
            yield Button("Sign in", id="go", variant="primary")
            yield Static("", id="error", markup=False)

    def on_mount(self) -> None:
        empty = next((i for i in self.query(Input) if not i.value), None)
        (empty or self.query_one("#pass", Input)).focus()

    def on_input_submitted(self, event: Input.Submitted) -> None:
        self._submit()

    def on_button_pressed(self, event: Button.Pressed) -> None:
        self._submit()

    def _submit(self) -> None:
        hub = self.query_one("#hub", Input).value.strip().rstrip("/")
        user = self.query_one("#user", Input).value.strip()
        passphrase = self.query_one("#pass", Input).value
        if not (hub and user and passphrase):
            self.query_one("#error", Static).update("Hub, username and passphrase.")
            return
        self.query_one("#error", Static).update("Signing in…")
        self.run_worker(self._sign_in(hub, user, passphrase), exclusive=True)

    async def _sign_in(self, hub_url: str, user: str, passphrase: str) -> None:
        app = self.app
        try:
            await app.use_hub(hub_url)
            account = Account(app.hub, user, lambda: passphrase)
            await account.hub.check_version()
            await account.sign_in(remember=True)
            kind = await account.remember(app.ctx.store)
        except CliError as e:
            self.query_one("#error", Static).update(e.message)
            return
        if kind == KEY_FILE:
            app.notify(FILE_KEY_NOTICE, timeout=10, markup=False)
        app.start(account)


class MainScreen(Screen):
    BINDINGS = [
        Binding("f5", "copy", "Copy"),
        Binding("delete", "delete", "Delete"),
        Binding("q", "app.quit", "Quit"),
        Binding("tab", "app.focus_next", "Switch", show=False),
    ]

    DEFAULT_CSS = """
    MainScreen > Horizontal { height: 1fr; }
    """

    def __init__(self, account: Account):
        super().__init__()
        self.account = account
        self.pool = Pool(account)
        self._trees: dict[str, Tree] = {}
        self._gates: dict[str, asyncio.Semaphore] = {}
        self._watching: set[str] = set()

    def compose(self) -> ComposeResult:
        with Horizontal():
            yield LocalPane(Path.cwd(), id="local")
            yield RemotePane(self, id="remote")
        yield TransferStrip(id="transfers")
        yield Footer()

    def on_mount(self) -> None:
        self.query_one(LocalPane).table.focus()

    @property
    def local(self) -> LocalPane:
        return self.query_one(LocalPane)

    @property
    def remote(self) -> RemotePane:
        return self.query_one(RemotePane)

    @property
    def strip(self) -> TransferStrip:
        return self.query_one(TransferStrip)

    # ── Groups ───────────────────────────────────────────────────────────────

    async def tree(self, group: dict) -> Tree:
        """The group's index, asked once and then kept current by the node's pushes."""
        if group["id"] not in self._trees:
            session = await self.pool.session(group)
            self._trees[group["id"]] = Tree.from_index(await session.index())
            if group["id"] not in self._watching:
                self._watching.add(group["id"])
                session.ch.on_push("index_delta", lambda _msg, g=group: self._changed(g))
        return self._trees[group["id"]]

    def forget_tree(self, group: dict) -> None:
        self._trees.pop(group["id"], None)

    def _changed(self, group: dict) -> None:
        """Something changed in the group: the index is asked for again, and the
        pane redrawn if it is showing that group."""
        self.forget_tree(group)
        remote = self.remote
        if remote.group is not None and remote.group["id"] == group["id"]:
            remote.load()

    # ── F5 ───────────────────────────────────────────────────────────────────

    def action_copy(self) -> None:
        focused = self.app.focused
        if focused is not None and self.remote in focused.ancestors_with_self:
            self.run_worker(self._download(), group="copy")
        else:
            self.run_worker(self._upload(), group="copy")

    def _gate(self, group: dict) -> asyncio.Semaphore:
        return self._gates.setdefault(group["id"], asyncio.Semaphore(PER_GROUP))

    async def _ask(self, message: str, options: list[tuple[str, str]]) -> str:
        return await self.app.push_screen(Choice(message, options), wait_for_dismiss=True)

    async def _download(self) -> None:
        remote, local = self.remote, self.local
        rows = remote.selection()
        if remote.group is None or not rows:
            self.app.notify("Open a group and choose what to download.", timeout=4)
            return
        group, dest = remote.group, local.path
        try:
            tree = await self.tree(group)
        except CliError as e:
            self.app.notify(e.message, severity="error", markup=False)
            return
        plan: list[tuple[dict, Path]] = []
        dirs: list[Path] = []
        problems: list[str] = []
        for row in rows:
            path = row.data["path"]
            top = path.rsplit("/", 1)[-1]
            if not tree.root_available(path):
                problems.append(f"{top}: its folder on the node is unavailable")
                continue
            if not row.is_dir:
                try:
                    plan.append((row.data["entry"], local_path(dest, [top])[0]))
                except UnsafeName as e:
                    problems.append(str(e))
                continue
            for folder, items in tree.walk(path):
                rel = [top] + [p for p in folder[len(path):].strip("/").split("/") if p]
                try:
                    dirs.append(local_path(dest, rel)[0])
                    plan += [(i.entry, local_path(dest, rel + [i.name])[0])
                             for i in items if not i.is_dir]
                except UnsafeName as e:
                    problems.append(str(e))
        for p in problems:
            self.app.notify(p, severity="warning", markup=False)
        existing = [t for _, t in plan if os.path.exists(long_path(t))
                    and not os.path.exists(long_path(t.with_name(t.name + PART_SUFFIX)))]
        overwrite = False
        if existing:
            names = ", ".join(safe(t.name) for t in existing[:5]) + (
                f" and {len(existing) - 5} more" if len(existing) > 5 else "")
            answer = await self._ask(
                f"{len(existing)} file(s) already in {dest}: {names}.",
                [("skip", "Skip them"), ("overwrite", "Overwrite"), ("cancel", "Cancel")])
            if answer == "cancel":
                return
            overwrite = answer == "overwrite"
            if not overwrite:
                plan = [(e, t) for e, t in plan if t not in existing]
        for d in dirs:
            await asyncio.to_thread(os.makedirs, long_path(d), exist_ok=True)
        remote.marks.clear()
        remote.redraw()
        session = await self.pool.session(group)
        await asyncio.gather(*(self._one_download(group, session, e, t, overwrite)
                               for e, t in plan))
        local.action_refresh()

    async def _one_download(self, group, session, entry: dict, target: Path,
                            overwrite: bool) -> None:
        job = self.strip.add("↓", entry["name"], int(entry["size"]))
        async with self._gate(group):
            self.strip.update_job(job, state="running")
            try:
                await download(
                    session, entry, target, overwrite=overwrite,
                    progress=lambda n: self.strip.update_job(job, done=job.done + n),
                    on_queued=lambda n: self.strip.update_job(job, state="queued",
                                                              note=f"({n} ahead)"))
                self.strip.update_job(job, state="done",
                                      note="" if target.name == entry["name"]
                                      else f"saved as {target.name}")
            except TransferError as e:
                self.strip.update_job(job, state="failed", note=e.message)

    async def _upload(self) -> None:
        remote, local = self.remote, self.local
        rows = local.selection()
        if not rows:
            return
        if remote.group is None or not remote.path:
            self.app.notify("Open a folder of a group on the right to upload into.",
                            timeout=4)
            return
        group, folder = remote.group, remote.path
        try:
            tree = await self.tree(group)
        except CliError as e:
            self.app.notify(e.message, severity="error", markup=False)
            return
        root = tree.roots.get(folder.split("/", 1)[0], {})
        if not root.get("writable") or not root.get("available", True):
            self.app.notify(f"{safe(handle(group))}/{safe(folder)} does not accept "
                            "uploads (read-only or unavailable).", severity="error",
                            markup=False)
            return
        try:
            dirs, files, problems = plan_upload([str(local.path / r.name) for r in rows],
                                                recursive=True)
        except CliError as e:
            self.app.notify(e.message, severity="error", markup=False)
            return
        problems += refused_names([f.rel[-1] for f in files], dirs)
        if problems:
            await self._ask("Nothing was sent:\n" + "\n".join(safe(p) for p in problems),
                            [("ok", "OK")])
            return
        tops = sorted({d[0] for d in dirs} | {f.rel[0] for f in files})
        taken = taken_names(tree, folder, tops)
        if taken:
            answer = await self._ask(
                f"Already in {safe(folder)}: {', '.join(safe(t) for t in sorted(taken))}. "
                "The node never replaces a file.",
                [("skip", "Skip them"), ("cancel", "Cancel")])
            if answer != "skip":
                return
            dirs = [d for d in dirs if d[0] not in taken]
            files = [f for f in files if f.rel[0] not in taken]
        local.marks.clear()
        local.redraw()
        session = await self.pool.session(group)
        remote_dir = {(): folder}
        try:
            for d in dirs:
                remote_dir[tuple(d)] = await create_dir(session, remote_dir[tuple(d[:-1])],
                                                        d[-1])
        except TransferError as e:
            self.app.notify(e.message, severity="error", markup=False)
            return
        await asyncio.gather(*(self._one_upload(group, session, f,
                                                remote_dir[tuple(f.rel[:-1])])
                               for f in files))
        self._changed(group)

    async def _one_upload(self, group, session, item, folder: str) -> None:
        name = item.rel[-1]
        job = self.strip.add("↑", name, item.size)
        async with self._gate(group):
            self.strip.update_job(job, state="running")
            lock = threading.Lock()
            try:
                with open(long_path(item.local), "rb") as fh:
                    def read(offset: int, length: int) -> bytes:
                        with lock:
                            fh.seek(offset)
                            return fh.read(length)
                    stored = await upload(
                        session, read, item.size, name, folder,
                        progress=lambda n: self.strip.update_job(job, done=job.done + n),
                        on_queued=lambda n: self.strip.update_job(job, state="queued",
                                                                  note=f"({n} ahead)"))
                self.strip.update_job(job, state="done", note="" if stored == name
                                      else f"stored as {stored}")
            except OSError as e:
                self.strip.update_job(job, state="failed", note=e.strerror or str(e))
            except TransferError as e:
                self.strip.update_job(job, state="failed", note=e.message)

    # ── Delete ───────────────────────────────────────────────────────────────

    def action_delete(self) -> None:
        focused = self.app.focused
        if focused is None or self.remote not in focused.ancestors_with_self:
            self.app.notify("Delete removes files from a group: select them on the "
                            "right.", timeout=4)
            return
        self.run_worker(self._delete(), group="delete")

    async def _delete(self) -> None:
        remote = self.remote
        rows = remote.selection()
        if remote.group is None or not rows:
            self.app.notify("Open a group and choose the files to delete.", timeout=4)
            return
        group = remote.group
        try:
            session = await self.pool.session(group)
        except CliError as e:
            self.app.notify(e.message, severity="error", markup=False)
            return
        folders = [r.name for r in rows if r.is_dir]
        not_yours = [r.name for r in rows if not r.is_dir
                     and not may_delete(session, r.data["entry"], self.account.user_id)]
        if folders or not_yours:
            why = (["Folders are not deleted here: " + ", ".join(map(safe, folders))]
                   if folders else [])
            if not_yours:
                why.append("You did not upload these, and this machine's key is not the "
                           "node's operator: " + ", ".join(map(safe, not_yours)))
            await self._ask("Nothing was deleted.\n" + "\n".join(why), [("ok", "OK")])
            return
        names = ", ".join(safe(r.name) for r in rows[:5]) + (
            f" and {len(rows) - 5} more" if len(rows) > 5 else "")
        answer = await self._ask(
            f"Delete {len(rows)} file(s) from {safe(handle(group))} for every member? "
            f"{names}", [("delete", "Delete"), ("cancel", "Cancel")])
        if answer != "delete":
            return
        remote.marks.clear()
        deleted = 0
        for row in rows:
            try:
                await delete_file(session, row.data["entry"]["id"])
                deleted += 1
            except CliError as e:
                self.app.notify(f"{safe(row.name)}: {e.message}", severity="error",
                                markup=False, timeout=8)
        if deleted:
            self.app.notify(f"Deleted {deleted} file(s).", timeout=4)
        self._changed(group)

    async def on_unmount(self) -> None:
        with contextlib.suppress(Exception):
            await self.pool.close()


class DeviceCode:
    """`account.on_device_code` for the interactive mode: the code in a
    notification that stays until the approval lands; the node is retried meanwhile."""

    def __init__(self, app):
        self.app = app

    def __call__(self, group: dict, pending, *, first: bool) -> bool:
        if first:
            self.app.notify(
                f"This machine is not yet a device of yours on the node hosting "
                f"{safe(handle(group))}. Approve it from the MeshBay application or a "
                f"browser: the group's Settings → \"Your devices on this node\" → "
                f"enter {pending.link_code}; or on a machine already linked: "
                f"meshbay-cli devices approve {safe(handle(group))} {pending.link_code}. "
                "Waiting…",
                title="Approve this machine", timeout=600, markup=False)
        return True