MeshTerm/meshterm/tools/channels.py

# SPDX-License-Identifier: Apache-2.0
"""The ``channels`` tool: create, join, and manage mesh channels.

Interactively it opens a full-screen channel manager (see :mod:`meshterm.ui.channels`) — a
first-class, phone-app-style experience for creating private channels, adding public ``#``
channels, joining with a key, importing a scanned ``meshcore://`` link, and sharing any
channel as a QR code. On the CLI it exposes ``list``, ``add``, ``join``, ``import``, and
``share`` subcommands for scripted use.

Channels are also *listed* by the ``chat`` tool for picking a conversation; this tool owns
everything to do with configuring the slots themselves.
"""

from __future__ import annotations

from typing import TYPE_CHECKING, Any

import typer

from ..context import AppContext
from ..core import exitcodes
from ..core.channels import (
    channel_hash,
    derive_secret,
    is_public_channel,
    normalize_secret,
    parse_share_url,
    random_secret,
    share_url,
)
from .base import Tool, ToolResult, register

if TYPE_CHECKING:  # pragma: no cover - typing only
    from ..ui.report import Facts, Listing


@register
class ChannelsTool(Tool):
    """Create, join, share, and manage the device's mesh channels."""

    name = "channels"
    title = "Channels"
    icon = "📻"
    help = "Create, join, and share mesh channels"
    category = "Message"
    order = 20

    async def run(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
        """Open the channel manager (menu) or run a scripted action (CLI).

        Args:
            ctx: Shared application context.
            params: A ``cli_action`` with its arguments (CLI), or empty for the menu.

        Returns:
            A :class:`ToolResult` recording what happened (the menu path shows nothing).
        """
        action = params.get("cli_action")
        if action is not None:
            return await self._run_cli(ctx, action, params)

        from ..ui.channels import manage_channels

        # No message: the manager acknowledged its work on the way out, which meant telling
        # the reader that changes they had just watched land were applied. The count still
        # goes to the run log through ``summary``, which is the record of what an invocation
        # did rather than anything shown.
        return ToolResult(summary={"changes": await manage_channels(ctx)})

    # -- CLI --------------------------------------------------------------------

    async def _run_cli(self, ctx: AppContext, action: str, params: dict[str, Any]) -> ToolResult:
        """Dispatch a scripted CLI action."""
        if action == "add":
            return await self._cli_add(ctx, params)
        if action == "join":
            return await self._cli_join(ctx, params)
        if action == "import":
            return await self._cli_import(ctx, params)
        if action == "share":
            return await self._cli_share(ctx, params)
        if action == "clear":
            return await self._cli_clear(ctx, params)
        return await self._cli_list(ctx)

    async def _cli_list(self, ctx: AppContext) -> ToolResult:
        """List the configured channel slots."""
        from ..core.channel_probe import read_channel_slots

        device = await ctx.device()
        slots = await read_channel_slots(device)
        return ToolResult(
            summary={"channels": len(slots)},
            report=(_channel_listing(slots),),
            exit_code=exitcodes.OK if slots else exitcodes.NO_RESULT,
        )

    async def _cli_add(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
        """Add a public, private-random, or explicitly-keyed channel on a slot."""
        from ..ui.channels import write_channel

        device = await ctx.device()
        idx = int(params["index"])
        name = str(params["name"])
        if name.startswith("#"):  # public: firmware derives the key from the name
            await write_channel(ctx, device, idx, name, None)
            secret = derive_secret(name)
        else:
            secret = normalize_secret(params["secret"]) if params.get("secret") else random_secret()
            await write_channel(ctx, device, idx, name, secret)
        await self._show_qr(ctx, name, share_url(name, secret))
        return ToolResult(
            summary={"index": idx, "name": name},
            report=(_written(idx, name, secret),),
        )

    async def _cli_join(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
        """Join a channel from a name and an explicit key."""
        from ..ui.channels import write_channel

        device = await ctx.device()
        idx = int(params["index"])
        name = str(params["name"])
        secret = normalize_secret(str(params["secret"]))
        await write_channel(ctx, device, idx, name, secret)
        return ToolResult(
            summary={"index": idx, "name": name},
            report=(_written(idx, name, secret, shown=False),),
        )

    async def _cli_import(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
        """Import a channel from a ``meshcore://channel/add`` link."""
        parsed = parse_share_url(str(params["url"]))
        if parsed is None:
            raise typer.BadParameter("not a valid meshcore:// channel link")
        name, secret = parsed
        from ..ui.channels import write_channel

        device = await ctx.device()
        idx = int(params["index"])
        await write_channel(ctx, device, idx, name, secret)
        return ToolResult(
            summary={"index": idx, "name": name},
            report=(_written(idx, name, secret, shown=False),),
        )

    async def _cli_share(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
        """Print a channel's share link, with its QR code in the menu.

        The one command that names a channel's key deliberately, so it is the one place
        the secret and the share URL leave the app at all — every other channel document
        carries the identity and stops there.
        """
        from ..core.channel_probe import read_channel_slots

        device = await ctx.device()
        idx = int(params["index"])
        slot = next((s for s in await read_channel_slots(device) if s.idx == idx), None)
        if slot is None:
            return ToolResult(
                summary={"index": idx, "shared": False},
                report=(_written(idx, None, None, shown=False),),
                exit_code=exitcodes.NO_RESULT,
            )
        await self._show_qr(ctx, slot.name, share_url(slot.name, slot.secret))
        return ToolResult(
            summary={"index": idx, "shared": True},
            report=(_written(idx, slot.name, slot.secret),),
        )

    async def _cli_clear(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
        """Clear a channel slot, removing the channel from the device.

        The scriptable face of the channel manager's *Clear this slot*: an empty name
        makes the firmware read the slot as unused. The ``--yes`` gate lives on the CLI
        command (a private channel's key is lost with the slot unless it's saved
        elsewhere), mirroring the interactive flow's danger confirmation.
        """
        from ..core.channel_probe import read_channel_slots
        from ..ui.channels import write_channel

        device = await ctx.device()
        idx = int(params["index"])
        slot = next((s for s in await read_channel_slots(device) if s.idx == idx), None)
        if slot is None:
            return ToolResult(
                summary={"index": idx, "cleared": False},
                report=(_cleared(idx, False),),
                exit_code=exitcodes.NO_RESULT,
            )
        await write_channel(ctx, device, idx, "", None)  # empty name => the slot reads as unused
        return ToolResult(
            summary={"index": idx, "cleared": True},
            report=(_cleared(idx, True),),
        )

    @staticmethod
    async def _show_qr(ctx: AppContext, name: str, url: str) -> None:
        """Draw a channel's share link as a QR code — in the menu, and only there.

        The QR is for a phone pointed at the screen, and there it takes the whole screen
        (:func:`~meshterm.ui.qr.share_screen`). Piped into a file it is a block of block
        characters wrapped around the one thing that is actually the answer, so a scripted
        run states the link and nothing else.
        """
        from ..ui.qr import share_screen
        from ..ui.surface import TuiUi

        if isinstance(ctx.ui, TuiUi):
            await share_screen(ctx, name=name, url=url)

    def register_cli(self, app: typer.Typer) -> None:
        """Register the ``channels`` subcommand group.

        Args:
            app: The Typer application.
        """
        from ..cli import run_tool_command

        channels_app = typer.Typer(help=self.help, no_args_is_help=True, rich_markup_mode=None)

        @channels_app.command("list", help="List the configured channel slots")
        def _list_cmd() -> None:
            run_tool_command(self, {"cli_action": "list"})

        @channels_app.command("add", help="Add a channel (# name = public; else private)")
        def _add_cmd(
            index: int = typer.Argument(..., help="Channel slot index"),
            name: str = typer.Argument(..., help="Channel name (leading # = public)"),
            secret: str | None = typer.Option(
                None, "--secret", help="32-hex-char key (private only; random if omitted)"
            ),
        ) -> None:
            run_tool_command(
                self, {"cli_action": "add", "index": index, "name": name, "secret": secret}
            )

        @channels_app.command("join", help="Join a channel with its name and key")
        def _join_cmd(
            index: int = typer.Argument(..., help="Channel slot index"),
            name: str = typer.Argument(..., help="Channel name"),
            secret: str = typer.Argument(..., help="32-hex-char (16-byte) key"),
        ) -> None:
            run_tool_command(
                self, {"cli_action": "join", "index": index, "name": name, "secret": secret}
            )

        @channels_app.command("import", help="Import a meshcore:// channel link")
        def _import_cmd(
            index: int = typer.Argument(..., help="Channel slot index"),
            url: str = typer.Argument(..., help="A meshcore://channel/add link"),
        ) -> None:
            run_tool_command(self, {"cli_action": "import", "index": index, "url": url})

        @channels_app.command("share", help="Print a channel's share link and QR code")
        def _share_cmd(
            index: int = typer.Argument(..., help="Channel slot index"),
        ) -> None:
            run_tool_command(self, {"cli_action": "share", "index": index})

        @channels_app.command("clear", help="Clear a channel slot (removes it from the device)")
        def _clear_cmd(
            index: int = typer.Argument(..., help="Channel slot index"),
            yes: bool = typer.Option(
                False, "--yes", help="Confirm removing the channel from this slot"
            ),
        ) -> None:
            if not yes:
                # A usage error, in the parser's own words and under its own status — not a
                # red line printed onto whatever is reading this command's output.
                raise typer.BadParameter(
                    "clearing a slot removes the channel (a private channel's key is lost "
                    "unless you have it saved). Re-run with --yes to confirm."
                )
            run_tool_command(self, {"cli_action": "clear", "index": index})

        app.add_typer(channels_app, name=self.name)


def _channel_listing(slots: list) -> Listing:
    """The configured slots, one record each.

    The one listing whose record *is* a shared shape rather than carrying one: a row that
    nested its only field under a ``channel`` key would make ``jq '.[].channel.name'``
    out of a listing whose every column is already the channel.
    """
    from ..ui import fields
    from ..ui.report import Listing

    return Listing(
        key="channels",
        columns=(
            fields.integer("slot", "SLOT"),
            fields.name("name", "NAME"),
            fields.word("type", "TYPE"),
            fields.hexid("hash", "HASH"),
        ),
        rows=[
            {
                "slot": slot.idx,
                "name": slot.name,
                "type": "public" if slot.is_public else "private",
                "hash": slot.hash,
            }
            for slot in slots
        ],
    )


def _written(idx: int, name: str | None, secret: bytes | None, *, shown: bool = True) -> Facts:
    """What a slot now holds, after ``add``, ``join``, ``import`` or ``share``.

    All four write or read the same thing and all four know the key they wrote, so all
    four state the same shape. The plain face prints the share URL alone where the caller
    asked to be given something to pass on (``add``, ``share``) and nothing at all where
    it did not (``join``, ``import``): those two were handed the key, and reading it back
    to them is not an answer.

    Args:
        idx: The slot index.
        name: The channel name, or ``None`` for a slot that turned out to be empty.
        secret: The 16-byte key, or ``None`` for an empty slot.
        shown: Whether the plain face prints the URL.

    Returns:
        The facts block.
    """
    from ..ui import fields
    from ..ui.fields import ChannelRef
    from ..ui.report import BARE, SILENT, Facts

    channel = None
    if name is not None and secret is not None:
        public = is_public_channel(name, secret)
        channel = ChannelRef(slot=idx, name=name, public=public, hash=channel_hash(secret))
    return Facts(
        key="channel",
        fields=(
            fields.channel("channel"),
            fields.hexid("secret", "secret"),
            fields.word("url", "url"),
        ),
        values={
            "channel": channel,
            "secret": secret.hex() if secret else None,
            "url": share_url(name, secret) if name is not None and secret is not None else None,
        },
        shape=BARE if shown else SILENT,
        bare="url",
    )


def _cleared(idx: int, cleared: bool) -> Facts:
    """What ``channels clear`` did — nothing plain, since the status already said it."""
    from ..ui import fields
    from ..ui.report import SILENT, Facts

    return Facts(
        key="cleared",
        fields=(fields.integer("slot", "slot"), fields.flag("cleared", "cleared")),
        values={"slot": idx, "cleared": cleared},
        shape=SILENT,
    )