1 files changed, 68 insertions, 0 deletions
diff --git a/packages/meshbay-cli/src/meshbay_cli/client/groups.py b/packages/meshbay-cli/src/meshbay_cli/client/groups.py
new file mode 100644
index 0000000..ac738f3
--- /dev/null
+++ b/packages/meshbay-cli/src/meshbay_cli/client/groups.py
@@ -0,0 +1,68 @@
+"""Opening a group: which of the nodes serving it to dial, and in which order."""
+
+import asyncio
+import logging
+import time
+
+from ..errors import NODE, CliError
+from .account import Account
+from .paths import handle
+from .session import DeviceLinkPending, GroupSession, NodeKeyChanged, connect
+
+# How often a machine waiting for its approval tries the node again. Each try is
+# one offer: well inside what the hub admits per account and per node.
+LINK_POLL = 10
+
+log = logging.getLogger(__name__)
+
+
+async def my_groups(account: Account) -> list[dict]:
+    return await account.call(account.hub.my_groups)
+
+
+async def open_group(account: Account, group: dict) -> GroupSession:
+    """
+    A session with a node serving `group`: the pinned one first, then the others
+    in the hub's order. A node whose key changed stops everything — that is an
+    alarm, not a reason to quietly try its neighbour.
+    """
+    nodes = await account.call(account.hub.group_nodes, group["id"])
+    if not nodes:
+        raise CliError(NODE, f"No node serving {handle(group)} is online.")
+    nodes = sorted(nodes, key=lambda n: account.pin(n["node_id"]) is None)
+    failures = []
+    for node in nodes:
+        try:
+            return await _connect_linking(account, group, node)
+        except (NodeKeyChanged, DeviceLinkPending):
+            raise
+        except CliError as e:
+            if e.code != NODE or len(nodes) == 1:
+                raise
+            log.info("node %s: %s", node["node_id"], e.message)
+            failures.append(f"{node['node_id']}: {e.message}")
+    raise CliError(NODE, f"No node serving {handle(group)} could be reached:\n  "
+                         + "\n  ".join(failures))
+
+
+async def _connect_linking(account: Account, group: dict, node: dict) -> GroupSession:
+    """
+    `connect`, and — when this machine is waiting to be approved as a device on
+    that node — the wait: the code is shown once (`account.on_device_code`), and
+    the node is tried again until the approval lands or the request expires. The
+    code names this node, so no other node is tried meanwhile.
+    """
+    shown = False
+    while True:
+        try:
+            return await connect(account, group["id"], node)
+        except DeviceLinkPending as pending:
+            hook = account.on_device_code
+            wait = hook(group, pending, first=not shown) if hook else False
+            shown = True
+            if not wait:
+                raise
+            if pending.expires_at <= time.time():
+                raise CliError(NODE, "The device request expired before it was approved; "
+                                     "run the command again for a new code.") from None
+            await asyncio.sleep(LINK_POLL)