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

"""
`chat`: a group's conversation — the last messages, or one more posted.

`chat` alone opens the interactive chat (`tui/chat.py`): the groups, one
group's conversation IRC-style, back to the groups. `chat GROUP` prints the
last ten, oldest first; `-N` (as `head -N`) or `-n N`
asks for another count. `chat GROUP -m TEXT` posts TEXT (`-m -` reads it from
stdin), sealed and signed as the application seals and signs it. Each message
is shown with its time and the name its sender sealed in it; one that cannot
be read is shown as such, never left out.

`chat GROUP --follow` prints the last messages, then each new one as the node
pushes it, until Ctrl+C. A connection that drops — or that stops answering the
ping sent every half minute, as a sleeping laptop's does — is opened again,
and what was said meanwhile is read from the history: each message is printed
once, whichever way it came (the node stores a device and nonce pair once).
With `--json`, one object per line.
"""

import asyncio
import contextlib
import json
import re
import sys

from ..client.chat import chat_keys, history, open_row, post
from ..client.groups import my_groups, open_group
from ..client.paths import group_arg, handle, resolve_group
from ..client.roster import RosterCheck
from ..errors import NODE, USAGE, CliError
from ..output import date, err, human_size, safe, sentence, terminal
from .context import Context

HELP = ("chat: alone, interactively; with a group, its last 10 messages (-N for N), "
        "or post with -m")

_COUNT = re.compile(r"^-(\d+)$")
# --follow: how often the connection is asked whether it is alive, how long it
# has to answer, and the waits between attempts to open it again.
PING_EVERY, PING_TIMEOUT = 30, 10
RECONNECT_WAITS = (2, 5, 10, 30)
# The one verdict worth a line (roster.py): the application's words.
KEY_CHANGED = ("! This account is using a key you have not seen before: a new device "
               "added elsewhere, or the node naming the wrong person.")


def add_parser(sub, parent) -> None:
    p = sub.add_parser("chat", parents=[parent], help=HELP, description=sentence(HELP)
                       + ". Messages are end-to-end encrypted and signed; the node "
                         "stores what it cannot read.")
    p.add_argument("group", metavar="GROUP", nargs="?",
                   help="the group, as `groups` lists it; without it, the interactive "
                        "chat: choose a group, talk, leave it, choose another")
    p.add_argument("-n", "--last", type=int, default=10, metavar="N",
                   help="how many of the last messages to show (also -N, as -100)")
    p.add_argument("-m", "--message", metavar="TEXT",
                   help="post TEXT instead of reading; - reads it from stdin")
    p.add_argument("-f", "--follow", action="store_true",
                   help="after the last messages, print new ones as they arrive, until "
                        "Ctrl+C; with --json, one object per line")


def rewrite_count(argv: list[str], actions) -> list[str]:
    """`chat -100 GROUP` as `chat --last 100 GROUP`: argparse reads -100 as a
    negative number, not as an option. Only when the action is `chat`."""
    action = next((i for i, a in enumerate(argv) if a in actions), None)
    if action is None or argv[action] != "chat":
        return argv
    out = argv[:action + 1]
    for a in argv[action + 1:]:
        m = _COUNT.match(a)
        out += ["--last", m.group(1)] if m else [a]
    return out


def _text(message: str) -> str:
    text = sys.stdin.read() if message == "-" else message
    text = text.rstrip("\n")
    if not text.strip():
        raise CliError(USAGE, "Nothing to post.")
    return text


async def run(ctx: Context) -> int:
    a = ctx.args
    if a.last < 1:
        raise CliError(USAGE, "Ask for one message at least.")
    if a.follow and a.message is not None:
        raise CliError(USAGE, "Post with -m, or follow with -f; not both at once.")
    if a.group is None:
        if a.message is not None or a.follow:
            raise CliError(USAGE, "Name the group to post to.")
        if not (terminal(sys.stdin) and terminal(sys.stdout)):
            raise CliError(USAGE, "The interactive chat needs a terminal; name a group "
                                  "to read it, or to post with -m.")
        from ..tui.app import run_interactive
        from ..tui.chat import ChatGroupsScreen
        # As with no argument at all: the sign-in screen checks the hub.
        return await run_interactive(ctx, ChatGroupsScreen)
    text = _text(a.message) if a.message is not None else None
    await ctx.hub.check_version()
    account = ctx.account()
    session = None
    try:
        group = resolve_group(await my_groups(account), group_arg(a.group))
        session = await open_group(account, group)
        keys = await chat_keys(session)
        if text is not None:
            await post(session, keys, text, account.username)
            if a.json:
                print(json.dumps({"posted": True, "group": handle(group)}))
            elif not a.quiet:
                err(f"Posted to {safe(handle(group))}.")
            return 0
        check = RosterCheck(account, session)
        messages = [await open_row(r, keys, group["id"], check)
                    for r in await history(session, a.last)]
        if a.follow:
            follower = Follower(ctx, account, group, session, keys, check)
            for m in messages:
                follower.emit(m)
            if not messages and not a.quiet:
                err(f"No message in {safe(handle(group))} yet; waiting for one "
                    "(Ctrl+C to stop).")
            session = None                      # the follower owns it now
            await follower.run()
            return 0
    finally:
        if session is not None:
            await session.close()
        await account.close()

    if a.json:
        json.dump([as_json(m) for m in messages], sys.stdout, indent=1, ensure_ascii=False)
        print()
        return 0
    if not messages and not a.quiet:
        err(f"No message in {safe(handle(group))} yet.")
    for m in messages:
        print_message(m)
    return 0


def print_message(m, *, flush: bool = False) -> None:
    head = f"{date(int(m.timestamp))}  {safe(m.sender)}: "
    if m.unreadable:
        print(f"{head}[this message cannot be read: {m.unreadable}]", flush=flush)
        return
    lines = safe_lines(m.text)
    if m.attachment:
        file_line = (f"[file: {safe(m.attachment['filename'])}, "
                     f"{human_size(m.attachment['size'])}]")
        lines = [file_line] if lines == [""] else [file_line, *lines]
    print(head + lines[0])
    for line in lines[1:]:
        print(" " * len(head) + line)
    if m.trust == "changed":
        print(" " * len(head) + KEY_CHANGED)
    if flush:
        sys.stdout.flush()


def as_json(m) -> dict:
    return {"id": m.id, "timestamp": m.timestamp, "sender": m.sender, "text": m.text,
            "trust": m.trust, **({"unreadable": m.unreadable} if m.unreadable else {}),
            **({"attachment": m.attachment} if m.attachment else {})}


class Follower:
    """`--follow`: one connection at a time, opened again when it dies, and
    every message printed once."""

    def __init__(self, ctx: Context, account, group: dict, session, keys, check):
        self.ctx, self.account, self.group = ctx, account, group
        self.session, self.keys, self.check = session, keys, check
        self.seen: set[str] = set()
        self.last_ts = 0.0

    def emit(self, m) -> None:
        if m.key:
            if m.key in self.seen:
                return
            self.seen.add(m.key)
        self.last_ts = max(self.last_ts, m.timestamp)
        if self.ctx.args.json:
            print(json.dumps(as_json(m), ensure_ascii=False), flush=True)
        else:
            print_message(m, flush=True)

    async def run(self) -> None:
        try:
            while True:
                await self._listen()
                await self.session.close()
                self.session = None
                if not self.ctx.args.quiet:
                    err("meshbay-cli: the connection to the node was lost; reconnecting...")
                await self._reopen()
        finally:
            if self.session is not None:
                await self.session.close()

    async def _listen(self) -> None:
        """Print what the node pushes until the connection dies."""
        s = self.session
        queue: asyncio.Queue = asyncio.Queue()
        dead = asyncio.Event()

        def on_message(msg):
            queue.put_nowait(msg)

        def on_epoch(msg):
            # Somebody was removed: the next message is sealed under a new key.
            s.ack["chat_epoch"] = int(msg.get("epoch") or 0)
            queue.put_nowait(None)

        def on_state():
            if s.pc is not None and s.pc.connectionState in ("failed", "closed"):
                dead.set()

        s.ch.on_push("chat_msg", on_message)
        s.ch.on_push("chat_epoch_ack", on_epoch)
        if s.pc is not None:
            s.pc.on("connectionstatechange", on_state)
        pinger = asyncio.ensure_future(self._ping(dead))
        closed = asyncio.ensure_future(s.ch.closed.wait())
        gone = asyncio.ensure_future(dead.wait())
        try:
            while True:
                getter = asyncio.ensure_future(queue.get())
                done, _ = await asyncio.wait({getter, closed, gone},
                                             return_when=asyncio.FIRST_COMPLETED)
                if getter not in done:
                    getter.cancel()
                    return
                msg = getter.result()
                if msg is None or int(msg.get("epoch") or 0) not in self.keys.by_epoch:
                    with contextlib.suppress(CliError):
                        self.keys = await chat_keys(s)
                if msg is None:
                    continue
                msg = {**msg, "timestamp": msg.get("timestamp") or 0}
                self.emit(await open_row(msg, self.keys, self.group["id"], self.check))
        finally:
            for task in (pinger, closed, gone):
                task.cancel()
            s.ch.off_push("chat_msg", on_message)
            s.ch.off_push("chat_epoch_ack", on_epoch)

    async def _ping(self, dead: asyncio.Event) -> None:
        """A connection that stopped answering looks open to both ends."""
        while True:
            await asyncio.sleep(PING_EVERY)
            try:
                await self.session.ch.request({"type": "ping", "token": "follow"}, "pong",
                                              timeout=PING_TIMEOUT)
            except CliError:
                dead.set()
                return

    async def _reopen(self) -> None:
        """A new connection, then what was said while there was none."""
        attempt = 0
        while True:
            try:
                self.session = await open_group(self.account, self.group)
                self.keys = await chat_keys(self.session)
                self.check = RosterCheck(self.account, self.session)
                rows = await history(self.session, 200)
                break
            except CliError as e:
                if self.session is not None:
                    await self.session.close()
                    self.session = None
                if e.code != NODE:
                    raise
                await asyncio.sleep(RECONNECT_WAITS[min(attempt, len(RECONNECT_WAITS) - 1)])
                attempt += 1
        for row in rows:
            m = await open_row(row, self.keys, self.group["id"], self.check)
            if m.timestamp >= self.last_ts:
                self.emit(m)


def safe_lines(text: str) -> list[str]:
    """A message's lines, each escaped: a line break stays a line break, every
    other control character is shown, never sent to the terminal."""
    return [safe(line) for line in text.split("\n")] or [""]