chatter + cli rewrite

colony/cli.py

@@ -1,403 +1,169 @@
 import argparse
-import datetime
 import json
 import sys
 import time
+from datetime import datetime
 
-from colony.core.network.requests import CommandRequest
-from colony.core.roles.chatter.role import Chatter
 from colony.core.storages.chat import ChatStorage
-from colony.core.roles.unit.role import Unit
+from colony.settings import setting
-from colony.settings import configuration, setting
+from colony.transport import control
-from colony.transport.whisperer import Whisperer
 
-DEFAULT_TIMEOUT = 5
+CLEAR = "\033[H\033[2J"
-DEFAULT_INTERVAL = 0.5
-
-HOME = "\033[H"
-CLEAR_SCREEN = "\033[2J"
-CLEAR_LINE = "\033[K"
-CLEAR_BELOW = "\033[J"
-
-def call(host, port, command, *args, sender=None, timeout=DEFAULT_TIMEOUT, cert_file=None, keyfile=None):
-    payload = CommandRequest.form(command, *args).collapse(sender=sender)
-    resp = Whisperer(host, port, cert_file, keyfile).send(payload, timeout=timeout)
-
-    if not isinstance(resp, dict) or resp.get("status") != "ok" or "result" not in resp:
-        reason = resp.get("error") if isinstance(resp, dict) else resp
-        raise RuntimeError(f"the node answered with an error: {reason}")
-    return resp["result"]
 
 
-def fetch_cluster(host, port, timeout=DEFAULT_TIMEOUT, cert_file=None, keyfile=None):
+class CliError(Exception):
-    return call(host, port, Unit.COMMANDS.SHOW_CLUSTER, timeout=timeout, cert_file=cert_file, keyfile=keyfile)
+    pass
 
 
-def request_role(host, port, role, timeout=DEFAULT_TIMEOUT, cert_file=None, keyfile=None):
+def call(args, action, **params):
-    return call(host, port, Unit.COMMANDS.REQUEST_ROLE, role, timeout=timeout, cert_file=cert_file, keyfile=keyfile)
-
-
-def chat_history(peer_id):
-    return ChatStorage().messages(peer_id)
-
-
-def send_chat(own, target, text, timeout=DEFAULT_TIMEOUT):
-    call(target["ip"], target["port"], Chatter.COMMANDS.RECEIVE, text, sender=member_id(own), timeout=timeout)
-    ChatStorage().append(member_id(target), "out", text)
-
-
-def member_id(member):
-    return f"{member['hostname']}@{member['ip']}"
-
-
-def resolve_chat_target(payload, name):
-    members = payload["members"]
-    own = next(member for member in members if member["self"])
-    if "Chatter" not in own["roles"]:
-        raise LookupError(f"{own['hostname']} is not running Chatter. Start it with: colonyctl role add Chatter")
-
-    matches = [member for member in members if name in (address(member), member_id(member))]
-    if not matches:
-        matches = [member for member in members if member["hostname"].lower() == name.lower()]
-    chatters = [member["hostname"] for member in members if "Chatter" in member["roles"] and not member["self"]]
-    if not matches:
-        raise LookupError(f"unknown node {name!r}. Nodes running Chatter: {', '.join(chatters) or 'none'}")
-    if len(matches) > 1:
-        raise LookupError(f"{name!r} matches more than one node, use an address: {', '.join(address(member) for member in matches)}")
-
-    target = matches[0]
-    if target["self"]:
-        raise LookupError("that is the node you are connected to, pick another node")
-    if "Chatter" not in target["roles"]:
-        raise LookupError(f"{target['hostname']} is not running Chatter. Nodes running it: {', '.join(chatters) or 'none'}")
-    return target
-
-
-def format_chat_time(at):
     try:
-        moment = datetime.datetime.fromisoformat(at).astimezone()
+        reply = control.request({"action": action, "args": params}, path=args.socket, timeout=args.timeout)
-    except (TypeError, ValueError):
+    except OSError as e:
-        return "?"
+        raise CliError(f"cannot reach the node at {args.socket}: {e}")
-    if moment.date() == datetime.date.today():
+    if reply.get("status") != "ok":
-        return moment.strftime("%H:%M:%S")
+        raise CliError(f"the node answered with an error: {reply.get('error')}")
-    return moment.strftime("%Y-%m-%d %H:%M")
+    return reply["result"]
 
 
-def render_chat(messages, own_label, peer_label):
+def table(rows):
-    if not messages:
+    widths = [max(len(str(row[i])) for row in rows) for i in range(len(rows[0]))]
-        return f"No messages with {peer_label} yet."
+    return "\n".join("  ".join(str(cell).ljust(width) for cell, width in zip(row, widths)).rstrip() for row in rows)
-    return "\n".join(
-        f"{format_chat_time(message.get('at'))}  {own_label if message.get('direction') == 'out' else peer_label}: {message.get('text', '')}"
-        for message in messages
-    )
 
 
-def resolve_on_demand_role(payload, name):
+def usage(stats, name):
-    roles = payload["roles"]
+    used, total = stats.get(f"{name}_used_gb"), stats.get(f"{name}_total_gb")
-    match = next((role for role in roles if role["name"].lower() == name.lower()), None)
+    return f"{used:.1f}/{total:.1f} GB" if used is not None and total is not None else "-"
-    if match is None:
-        raise LookupError(f"unknown role {name!r}. Roles: {', '.join(role['name'] for role in roles)}")
-    if not match.get("on_demand"):
-        on_demand = [role["name"] for role in roles if role.get("on_demand")]
-        raise LookupError(
-            f"{match['name']} is assigned automatically, only on-demand roles can be added by hand"
-            + (f" ({', '.join(on_demand)})" if on_demand else "")
-        )
-    return match["name"]
 
 
-def format_age(seconds):
+def render_status(cluster):
-    seconds = int(seconds)
+    members = cluster["members"]
-    if seconds < 60:
+    nodes = [["NODE", "ADDRESS", "STATE", "CPU", "RAM", "DISK", "ROLES"]]
-        return f"{seconds}s"
-    if seconds < 3600:
-        return f"{seconds // 60}m{seconds % 60:02d}s"
-    return f"{seconds // 3600}h{seconds % 3600 // 60:02d}m"
-
-
-def table(headers, rows):
-    widths = [max(len(str(cell)) for cell in column) for column in zip(headers, *rows)]
-    lines = ["  ".join(str(cell).ljust(width) for cell, width in zip(row, widths)).rstrip() for row in [headers, *rows]]
-    return "\n".join(lines)
-
-
-def format_size(gb):
-    if gb >= 1000:
-        return f"{gb / 1024:.1f} TB"
-    if gb >= 100:
-        return f"{gb:.0f} GB"
-    return f"{gb:.1f} GB"
-
-
-def format_uptime(hours):
-    if hours < 1:
-        return f"{round(hours * 60)}m"
-    if hours < 48:
-        return f"{round(hours)}h"
-    return f"{int(hours // 24)}d {round(hours % 24)}h"
-
-
-def describe_stats(stats):
-    stats = stats or {}
-
-    def usage(used_key, total_key):
-        if used_key in stats and total_key in stats:
-            return f"{format_size(stats[used_key])} / {format_size(stats[total_key])}"
-        return "-"
-
-    cpu = "-"
-    if "cpu_percent" in stats:
-        cpu = f"{stats['cpu_percent']:.0f}%"
-        if stats.get("cpu_cores") and stats.get("cpu_threads"):
-            cpu += f" · {stats['cpu_cores']}C/{stats['cpu_threads']}T"
-
-    gpu = usage("gpu_mem_used_gb", "gpu_mem_total_gb")
-    if gpu != "-" and stats.get("gpu_count", 1) > 1:
-        gpu = f"{stats['gpu_count']}× {gpu}"
-
-    uptime = format_uptime(stats["uptime_hours"]) if "uptime_hours" in stats else "-"
-    return [cpu, usage("mem_used_gb", "mem_total_gb"), usage("disk_used_gb", "disk_total_gb"), gpu, uptime]
-
-
-def address(member):
-    return f"{member['ip']}:{member['port']}"
-
-
-def role_states(payload):
-    alive = [member for member in payload["members"] if member["alive"]]
-    states = []
-    for role in payload["roles"]:
-        holders = sorted(address(member) for member in alive if role["name"] in member["roles"])
-        required = role["required"]
-        limit = role.get("max", required)
-        if role.get("on_demand"):
-            state = "held" if holders else "off"
-        elif required is None:
-            state = "general"
-        elif len(holders) < required:
-            state = "unmet"
-        elif limit is not None and len(holders) > limit:
-            state = "duplicate"
-        else:
-            state = "held"
-        states.append({**role, "holders": holders, "state": state})
-    return states
-
-
-def address_of(key):
-    return f"{key[0]}:{key[1]}" if key else ""
-
-
-def render_status(payload):
-    members = payload["members"]
-
-    hostnames = [member["hostname"] for member in members]
-    labels = {
-        address(member): member["hostname"] if hostnames.count(member["hostname"]) == 1 else address(member)
-        for member in members
-    }
-
-    alive = [member for member in members if member["alive"]]
-    down = len(members) - len(alive)
-    self_member = next(member for member in members if member["self"])
-    title = f"Cluster as seen by {labels[address(self_member)]}: {len(alive)} node(s) alive"
-    if down:
-        title += f", {down} down"
-
-    show_declined = any(member["declined"] for member in members)
-    show_route = any(member.get("via") for member in members)
-    node_headers = ["NODE", "ADDRESS", "STATE", "SEEN", "ROLES"] + (["DECLINED"] if show_declined else []) + (["ROUTE"] if show_route else [])
-    node_rows = []
     for member in members:
-        label = labels[address(member)] + (" (this node)" if member["self"] else "")
+        stats = member.get("stats") or {}
-        row = [
+        nodes.append([
-            label,
+            member["hostname"] + (" (this node)" if member["self"] else ""),
-            address(member),
+            f"{member['ip']}:{member['port']}",
-            "alive" if member["alive"] else "DOWN",
+            ("alive" if member["alive"] else "DOWN") + (" via gateway" if member.get("via") else ""),
-            "-" if member["self"] else format_age(member["seen_ago"]),
+            f"{stats['cpu_percent']:.0f}%" if "cpu_percent" in stats else "-",
+            usage(stats, "mem"),
+            usage(stats, "disk"),
             " ".join(member["roles"]) or "-",
-        ]
+        ])
-        if show_declined:
-            row.append(" ".join(member["declined"]) or "-")
-        if show_route:
-            via = member.get("via")
-            if via:
-                row.append(f"via {labels.get(address_of(via), address_of(via))}")
-            else:
-                row.append("direct" if member.get("direct") or member["self"] else "unknown")
-        node_rows.append(row)
 
-    alive_addresses = {address(member) for member in alive}
+    roles = [["ROLE", "PRIORITY", "HOLDERS"]]
-    role_rows = []
+    for role in cluster["roles"]:
-    for role in role_states(payload):
+        holders = [member["hostname"] for member in members if member["alive"] and role["name"] in member["roles"]]
-        holders = role["holders"]
+        empty = "off (on demand)" if role.get("on_demand") else "UNMET"
-        if role["state"] == "general":
+        roles.append([role["name"], role["priority"], ", ".join(holders) or empty])
-            state = "ok"
-        else:
-            state = {
-                "held": "ok",
-                "off": "off (on demand)",
-                "unmet": f"UNMET ({len(holders)}/{role['required']})",
-                "duplicate": f"DUPLICATE ({len(holders)}/{role['required']})",
-            }[role["state"]]
-        if role["state"] == "general" and set(holders) == alive_addresses:
-            names = "all nodes"
-        else:
-            names = ", ".join(labels[holder] for holder in holders) or "-"
-        role_rows.append([role["name"], role["priority"], state, names])
 
-    hardware_rows = [[labels[address(member)]] + describe_stats(member.get("stats")) for member in alive]
+    return table(nodes) + "\n\n" + table(roles)
 
-    off = [role["name"] for role in role_states(payload) if role["state"] == "off"]
+
-    sections = [
+def render_chat(messages, peer):
-        title,
+    if not messages:
-        table(node_headers, node_rows),
+        return f"No messages with {peer} yet."
-        table(["ROLE", "PRIORITY", "STATE", "HOLDERS"], role_rows),
+    return "\n".join(f"{datetime.fromisoformat(m['at']).astimezone():%H:%M}  {'you' if m['direction'] == 'out' else peer}: {m['text']}" for m in messages)
-        table(["NODE", "CPU", "RAM", "DISK", "GPU", "UP"], hardware_rows),
+
+
+def find_node(cluster, name):
+    matches = [
+        member for member in cluster["members"]
+        if name.lower() in (member["hostname"].lower(), f"{member['ip']}:{member['port']}", f"{member['hostname']}@{member['ip']}".lower())
     ]
-    if off:
+    if len(matches) != 1:
-        sections.append(f"Not running (on demand): {', '.join(off)}. Start one with: colonyctl role add {off[0]}")
+        raise CliError(f"{'no node' if not matches else 'more than one node'} matches {name!r}")
-    return "\n\n".join(sections)
+    return matches[0]
+
+
+def show(args, render):
+    if args.json or args.raw or not sys.stdout.isatty():
+        print(render())
+        return
+    try:
+        while True:
+            try:
+                text = render()
+            except CliError as e:
+                text = f"colonyctl: {e}\nretrying..."
+            print(CLEAR + text + "\n\nCtrl-C to quit.", flush=True)
+            time.sleep(args.interval)
+    except KeyboardInterrupt:
+        print()
+
+
+def status(args):
+    def render():
+        cluster = call(args, "status")
+        return json.dumps(cluster, indent=2) if args.json else render_status(cluster)
+    show(args, render)
+
+
+def add_role(args, name):
+    cluster = call(args, "status")
+    on_demand = {role["name"].lower(): role["name"] for role in cluster["roles"] if role.get("on_demand")}
+    if name.lower() not in on_demand:
+        raise CliError(f"{name!r} is not an on-demand role, pick one of: {', '.join(on_demand.values())}")
+    role = on_demand[name.lower()]
+
+    reply = call(args, "add_role", role=role)
+    if not reply["accepted"]:
+        raise CliError(f"the node did not take {role}: {reply['reason']}")
+    print(f"{role} is running ({reply['reason']})")
+    if role == "WebView":
+        me = next(member for member in cluster["members"] if member["self"])
+        print(f"Web view: http://{me['ip']}:{setting('web.port')}")
+
+
+def chat(args):
+    peer = find_node(call(args, "status"), args.node)
+    peer_id = f"{peer['hostname']}@{peer['ip']}"
+
+    if args.text is not None:
+        call(args, "send_chat", peer=peer_id, text=args.text)
+        print(f"Sent to {peer['hostname']}")
+    else:
+        history = lambda: ChatStorage().messages(peer_id)
+        show(args, lambda: json.dumps(history(), indent=2) if args.json else render_chat(history(), peer["hostname"]))
 
 
 def build_parser():
-    parser = argparse.ArgumentParser(prog="colonyctl", description="Inspect and control a running colony node.")
+    common = argparse.ArgumentParser(add_help=False)
+    common.add_argument("--socket", default=str(control.CONTROL_SOCKET), help="control socket of the local node")
+    common.add_argument("--timeout", type=float, default=5)
+
+    live = argparse.ArgumentParser(add_help=False)
+    live.add_argument("--interval", type=float, default=1, help="seconds between refreshes")
+    live.add_argument("--raw", action="store_true", help="print once instead of refreshing")
+    live.add_argument("--json", action="store_true", help="print the raw data as JSON once")
+
+    parser = argparse.ArgumentParser(prog="colonyctl", description="Inspect and control the local colony node.")
     commands = parser.add_subparsers(dest="command", required=True)
 
-    connection = argparse.ArgumentParser(add_help=False)
+    commands.add_parser("status", parents=[common, live], help="show nodes and roles")
-    connection.add_argument("--host", default="127.0.0.1", help="node to ask (default: this machine)")
-    connection.add_argument("--port", type=int, default=None, help="node port (default: listener.port from config)")
-    connection.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT)
 
-    status = commands.add_parser("status", parents=[connection], help="show members, roles and hardware (live in a terminal)")
+    role = commands.add_parser("role", help="manage roles").add_subparsers(dest="role_command", required=True)
-    status.add_argument("--interval", type=float, default=DEFAULT_INTERVAL, help="seconds between refreshes")
+    role.add_parser("add", parents=[common], help="run an on-demand role on this node").add_argument("name")
-    status.add_argument("--raw", action="store_true", help="print once and exit instead of refreshing")
-    status.add_argument("--json", action="store_true", help="print the raw data as JSON once and exit")
 
-    role = commands.add_parser("role", help="manage roles on a node").add_subparsers(dest="role_command", required=True)
+    commands.add_parser("webview", parents=[common], help="same as: role add WebView")
-    add = role.add_parser("add", parents=[connection], help="ask the node to run an on-demand role")
-    add.add_argument("name", help="role name, e.g. WebView (case does not matter)")
 
-    commands.add_parser("webview", parents=[connection], help="ask the node to run the web view (same as: role add WebView)")
+    chat = commands.add_parser("chat", parents=[common, live], help="show the chat with a node, or send it a message")
-
+    chat.add_argument("node", help="hostname, ip:port or hostname@ip")
-    chat = commands.add_parser("chat", parents=[connection], help="show the chat with a node (live in a terminal), or send it a message; the history is read from this machine's storages/chat")
+    chat.add_argument("text", nargs="?", help="message to send")
-    chat.add_argument("node", help="hostname, ip:port or hostname@ip of the node to chat with")
-    chat.add_argument("text", nargs="?", help="message to send; without it the history is shown")
-    chat.add_argument("--interval", type=float, default=DEFAULT_INTERVAL, help="seconds between refreshes")
-    chat.add_argument("--raw", action="store_true", help="print the history once and exit instead of refreshing")
-    chat.add_argument("--json", action="store_true", help="print the history as JSON once and exit")
     return parser
 
 
-def draw(out, text):
+def main(argv=None):
-    out.write(HOME + text.replace("\n", CLEAR_LINE + "\n") + CLEAR_LINE + CLEAR_BELOW)
-    out.flush()
-
-
-def watch(host, port, args, out):
-    out.write(CLEAR_SCREEN)
-    try:
-        while True:
-            try:
-                text = render_status(fetch_cluster(host, port, timeout=args.timeout))
-            except (OSError, ValueError, RuntimeError) as e:
-                text = f"colonyctl: cannot get status from {host}:{port}: {e}\nretrying..."
-            footer = f"Every {args.interval:g}s. Updated {time.strftime('%H:%M:%S')}. Ctrl-C to quit."
-            draw(out, f"{text}\n\n{footer}")
-            time.sleep(args.interval)
-    except KeyboardInterrupt:
-        out.write("\n")
-        return 0
-
-
-def watch_chat(args, port, target, own, out):
-    peer_label, own_label = target["hostname"], f"{own['hostname']} (you)"
-    out.write(CLEAR_SCREEN)
-    try:
-        while True:
-            try:
-                text = render_chat(chat_history(member_id(target)), own_label, peer_label)
-            except (OSError, ValueError, RuntimeError) as e:
-                text = f"colonyctl: cannot read the chat: {e}\nretrying..."
-            footer = f"Chat with {peer_label}. Every {args.interval:g}s. Updated {time.strftime('%H:%M:%S')}. Ctrl-C to quit."
-            footer += f'\nSend with: colonyctl chat {target["hostname"]} "message"'
-            draw(out, f"{text}\n\n{footer}")
-            time.sleep(args.interval)
-    except KeyboardInterrupt:
-        out.write("\n")
-        return 0
-
-
-def chat_command(args, port, out):
-    try:
-        payload = fetch_cluster(args.host, port, timeout=args.timeout)
-        target = resolve_chat_target(payload, args.node)
-        own = next(member for member in payload["members"] if member["self"])
-        if args.text is not None:
-            send_chat(own, target, args.text, timeout=args.timeout)
-        elif not args.raw and not args.json and out.isatty():
-            return watch_chat(args, port, target, own, out)
-        else:
-            history = chat_history(member_id(target))
-    except LookupError as e:
-        print(f"colonyctl: {e}", file=sys.stderr)
-        return 1
-    except (OSError, ValueError, RuntimeError) as e:
-        print(f"colonyctl: cannot reach {args.host}:{port}: {e}", file=sys.stderr)
-        return 1
-
-    if args.text is not None:
-        print(f"Sent to {target['hostname']}", file=out)
-    elif args.json:
-        print(json.dumps(history, indent=2), file=out)
-    else:
-        print(render_chat(history, f"{own['hostname']} (you)", target["hostname"]), file=out)
-    return 0
-
-
-def add_role(args, port, name, out):
-    try:
-        payload = fetch_cluster(args.host, port, timeout=args.timeout)
-        role = resolve_on_demand_role(payload, name)
-        reply = request_role(args.host, port, role, timeout=args.timeout)
-    except LookupError as e:
-        print(f"colonyctl: {e}", file=sys.stderr)
-        return 1
-    except (OSError, ValueError, RuntimeError) as e:
-        print(f"colonyctl: cannot reach {args.host}:{port}: {e}", file=sys.stderr)
-        return 1
-
-    node = next(member for member in payload["members"] if member["self"])
-    if not reply.get("accepted"):
-        print(f"colonyctl: {node['hostname']} did not take {role}: {reply.get('reason')}", file=sys.stderr)
-        return 1
-
-    print(f"{role} is running on {node['hostname']} ({reply.get('reason')})", file=out)
-    if role == "WebView":
-        print(f"Web view: http://{node['ip']}:{setting('web.port')}", file=out)
-    return 0
-
-
-def main(argv=None, out=None):
     args = build_parser().parse_args(argv)
-    out = sys.stdout if out is None else out
-    port = args.port if args.port is not None else configuration.listener.port
-
-    if args.command == "role":
-        return add_role(args, port, args.name, out)
-    if args.command == "webview":
-        return add_role(args, port, "WebView", out)
-    if args.command == "chat":
-        return chat_command(args, port, out)
-
-    if not args.raw and not args.json and out.isatty():
-        return watch(args.host, port, args, out)
-
     try:
-        payload = fetch_cluster(args.host, port, timeout=args.timeout)
+        if args.command == "status":
-    except (OSError, ValueError, RuntimeError) as e:
+            status(args)
-        print(f"colonyctl: cannot get status from {args.host}:{port}: {e}", file=sys.stderr)
+        elif args.command == "role":
+            add_role(args, args.name)
+        elif args.command == "webview":
+            add_role(args, "WebView")
+        elif args.command == "chat":
+            chat(args)
+    except CliError as e:
+        print(f"colonyctl: {e}", file=sys.stderr)
         return 1
-
-    print(json.dumps(payload, indent=2) if args.json else render_status(payload), file=out)
     return 0

colony/core/control.py

@@ -0,0 +1,34 @@
+class ControlActions:
+    def __init__(self, network):
+        self.network = network
+        self.actions = {
+            "status": self.status,
+            "add_role": self.add_role,
+            "send_chat": self.send_chat,
+        }
+
+    def run(self, payload):
+        action = payload.get("action") if isinstance(payload, dict) else None
+        handler = self.actions.get(action)
+        if handler is None:
+            raise ValueError(f"Unknown action {action!r}")
+
+        args = payload.get("args", {})
+        if not isinstance(args, dict):
+            raise ValueError("Action args must be a dict")
+        return handler(**args)
+
+    def _role(self, name):
+        role = self.network.get_self().get_role_instances().get(name)
+        if role is None:
+            raise LookupError(f"{name} is not running on this node")
+        return role
+
+    def status(self):
+        return self._role("Unit").show_cluster()
+
+    def add_role(self, role):
+        return self._role("Unit").request_role(role)
+
+    def send_chat(self, peer, text):
+        self._role("Chatter").send(peer, text)

colony/core/roles/chatter/role.py

@@ -2,6 +2,7 @@ from enum import Enum
 
 from colony.core.roles.role import Role, Command
 from colony.core.storages.chat import ChatStorage
+from colony.settings import setting
 
 MAX_MESSAGE_LENGTH = 4000
 
@@ -16,6 +17,7 @@ class Chatter(Role):
         }
 
         self.storage = storage or ChatStorage()
+        self.max_messages = setting("chat.max_messages")
 
         super().__init__(commands, context)
 
@@ -28,14 +30,22 @@ class Chatter(Role):
         if response.get("status") != "ok":
             raise ValueError(f"{peer_id} refused the message")
 
-        self.storage.append(peer_id, "out", text)
+        self.storage.append(peer_id, "out", text, self.max_messages)
 
     def receive(self, text, caller_id):
         if not isinstance(text, str) or not text.strip() or len(text) > MAX_MESSAGE_LENGTH:
             raise ValueError("Invalid message")
 
-        self.storage.append(caller_id, "in", text)
+        self.storage.append(caller_id, "in", text, self.max_messages)
         return True
 
     def history(self, peer_id, after=0):
-        return self.storage.messages(peer_id)[after:]
+        return [message for message in self.storage.messages(peer_id) if message["seq"] > after]
+
+    def inbox(self):
+        peers = {}
+        for peer_id in self.storage.peers():
+            messages = self.storage.messages(peer_id)
+            incoming = [message for message in messages if message["direction"] == "in"]
+            peers[peer_id] = {"incoming": messages[-1]["received"] if messages else 0, "last": incoming[-1] if incoming else None}
+        return peers

colony/core/roles/webview/app.py

@@ -1,3 +1,4 @@
+import hmac
 import logging
 from pathlib import Path
 
@@ -8,13 +9,29 @@ STATIC_DIR = Path(__file__).parent / "static"
 logging.getLogger("werkzeug").setLevel(logging.WARNING)
 
 
-def create_app(get_status, get_chatter):
+def create_app(get_status, get_chatter, credentials=None):
     app = Flask(__name__, static_folder=str(STATIC_DIR), static_url_path="/static")
 
     def no_store(response):
         response.headers["Cache-Control"] = "no-store"
         return response
 
+    def authorized():
+        auth = request.authorization
+        if auth is None or auth.type != "basic":
+            return False
+        username, password = credentials
+        return hmac.compare_digest((auth.username or "").encode(), username.encode()) & \
+            hmac.compare_digest((auth.password or "").encode(), password.encode())
+
+    @app.before_request
+    def require_login():
+        if credentials is not None and not authorized():
+            response = jsonify(error="login required")
+            response.status_code = 401
+            response.headers["WWW-Authenticate"] = 'Basic realm="Colony", charset="UTF-8"'
+            return response
+
     @app.get("/")
     def index():
         return app.send_static_file("index.html")
@@ -60,4 +77,12 @@ def create_app(get_status, get_chatter):
             return jsonify(error=f"{body['peer']} cannot be reached"), 502
         return jsonify(sent=True)
 
+    @app.get("/api/chat/inbox")
+    def chat_inbox():
+        chatter = get_chatter()
+        if chatter is None:
+            return jsonify(error="Chatter is not running on this node"), 409
+
+        return no_store(jsonify(chatter.inbox()))
+
     return app

colony/core/roles/webview/role.py

@@ -21,7 +21,8 @@ class WebView(Role):
 
     def __init__(self, context):
         host, port = setting("web.host"), setting("web.port")
-        self.app = create_app(self._status, self._chatter)
+        username, password = setting("web.username"), setting("web.password")
+        self.app = create_app(self._status, self._chatter, (username, password) if password else None)
         try:
             self._server = make_server(host, port, self.app, threaded=True)
         except SystemExit:
@@ -143,6 +144,8 @@ class WebView(Role):
             threading.Thread(target=self._refresh_loop, daemon=True, name="WebViewTopology").start()
 
         self.logger.info(f"Web view at http://{self.context.self().ip}:{self.port}")
+        if not setting("web.password"):
+            self.logger.warning("The web view has no password, set web.password to require one")
 
     def stop(self):
         super().stop()

colony/core/roles/webview/static/app.js

@@ -299,15 +299,21 @@
     const texts = { name: svg("text", { class: "l-name" }), roles: [svg("text", { class: "l-roles" }), svg("text", { class: "l-roles" })], caption: svg("text", { class: "l-caption" }) };
     [title, halo, shape, dot, texts.name, texts.roles[0], texts.roles[1], texts.caption].forEach((part) => group.appendChild(part));
 
+    const badge = svg("g", { class: "badge" });
+    const badgeText = svg("text", { "text-anchor": "middle", y: 3.5 });
+    badge.appendChild(svg("circle", { r: 9 }));
+    badge.appendChild(badgeText);
+
     /* Hardware: three bars with their numbers, and a footer. */
     const stats = svg("g", { class: "stats" });
     const rows = Array.from({ length: C.STAT_COUNT }, () => statBar(stats, C.STAT_W));
     const footer = svg("text", { class: "st-foot" });
     stats.appendChild(footer);
     group.appendChild(stats);
+    group.appendChild(badge);
 
     layers.nodes.appendChild(group);
-    return { group: group, title: title, halo: halo, shape: shape, dot: dot, texts: texts, stats: stats, rows: rows, footer: footer, gateway: node.gateway };
+    return { group: group, title: title, halo: halo, shape: shape, dot: dot, texts: texts, stats: stats, rows: rows, footer: footer, badge: badge, badgeText: badgeText, gateway: node.gateway };
   }
 
   function drawNodes() {
@@ -364,11 +370,28 @@
       put(record.footer, "y", C.STAT_COUNT * C.STAT_ROW + 10);
     });
     prune(nodeRecords, present);
+    drawUnread();
+  }
+
+  function drawUnread() {
+    let total = 0;
+    nodeRecords.forEach((record, key) => {
+      const info = model.byKey.get(key);
+      const count = info ? unreadOf(info.id) : 0;
+      total += count;
+      const corner = record.node.r * 0.8;
+      put(record.badge, "transform", "translate(" + corner + "," + -corner + ")");
+      put(record.badge, "display", count ? "inline" : "none");
+      putText(record.badgeText, count > 9 ? "9+" : String(count));
+    });
+    document.title = total ? "(" + total + ") Colony" : "Colony";
+    drawBell(total);
   }
 
   /* The tooltip and the screen reader's text carry the numbers too, so they follow them. */
   function describe(record, info) {
-    const text = C.describeNode(model, info, routes);
+    const unread = unreadOf(info.id);
+    const text = C.describeNode(model, info, routes) + (unread ? "\n" + C.plural(unread, "unread message") : "");
     putText(record.title, text);
     put(record.group, "aria-label", text.replace(/\n/g, ". ") + ". Show routes from this node.");
   }
@@ -851,7 +874,7 @@
   });
 
   const [chatPanel, chatTitle, chatStatus, chatLog, chatEmpty, chatError, chatForm, chatInput, chatSend, chatClose, chatLive] = ["chat", "chat-title", "chat-status", "chat-log", "chat-empty", "chat-error", "chat-form", "chat-input", "chat-send", "chat-close", "chat-live"].map((id) => document.getElementById(id));
-  let chat = { peer: null, count: 0, loaded: false, loading: false };
+  let chat = { peer: null, after: 0, loaded: false, loading: false };
 
   const chatTarget = () => (model && selectionLocks() && selection.id !== model.selfKey ? model.byKey.get(selection.id) : null);
   const chatTime = (at) => {
@@ -864,11 +887,11 @@
     chatPanel.hidden = !peer;
     stage.classList.toggle("chatting", Boolean(peer));
     if (!peer) {
-      chat = { peer: null, count: 0, loaded: false, loading: false };
+      chat = { peer: null, after: 0, loaded: false, loading: false };
       return;
     }
     if (chat.peer !== peer.id) {
-      chat = { peer: peer.id, hostname: peer.hostname, count: 0, loaded: false, loading: false };
+      chat = { peer: peer.id, hostname: peer.hostname, after: 0, loaded: false, loading: false };
       chatTitle.textContent = "Chat with " + peer.hostname;
       chatLog.replaceChildren();
       chatEmpty.hidden = true;
@@ -876,9 +899,15 @@
       chatInput.value = "";
     }
     const state = C.chatState(model, peer.key);
-    chatStatus.textContent = state.text;
+    const ignored = isIgnored(peer.id);
-    chatStatus.hidden = !state.text;
+    const status = [ignored ? "You are ignoring " + peer.hostname + " in this browser: no notifications or unread counts for them." : "", state.text].filter(Boolean).join(" ");
+    chatStatus.textContent = status;
+    chatStatus.hidden = !status;
     chatForm.hidden = !state.send;
+    chatIgnore.hidden = !inbox || !state.read;
+    chatIgnore.textContent = ignored ? "Unignore" : "Ignore";
+    chatIgnore.title = ignored ? "Get notified about " + peer.hostname + " again" : "Stop notifications and unread counts for " + peer.hostname;
+    chatIgnore.setAttribute("aria-pressed", ignored ? "true" : "false");
     if (state.read) loadChat();
   }
 
@@ -903,14 +932,15 @@
     if (current.loading || !current.peer) return;
     current.loading = true;
     try {
-      const response = await fetch("/api/chat?peer=" + encodeURIComponent(current.peer) + "&after=" + current.count, { cache: "no-store" });
+      const response = await fetch("/api/chat?peer=" + encodeURIComponent(current.peer) + "&after=" + current.after, { cache: "no-store" });
       const body = await response.json().catch(() => ({}));
       if (!response.ok) throw new Error(body.error || "the node answered with HTTP " + response.status);
       if (chat !== current) return;
       paintChatMessages(body.messages, current.loaded);
-      current.count += body.messages.length;
+      if (body.messages.length) current.after = body.messages[body.messages.length - 1].seq;
       current.loaded = true;
-      chatEmpty.hidden = current.count > 0;
+      if (!document.hidden) markSeen(current.peer);
+      chatEmpty.hidden = chatLog.children.length > 0;
       if (chatError.dataset.from === "load") chatError.hidden = true;
     } catch (error) {
       if (chat === current) {
@@ -954,6 +984,229 @@
     }
   });
 
+  /* ------------------------------------------------------------ notifications */
+
+  const SEEN_KEY = "colony.chat.seen.v1";
+  const IGNORED_KEY = "colony.chat.ignored.v1";
+  const TOAST_MS = 8000;
+  const TOAST_LIMIT = 3;
+  const [notifyButton, toastBox, chatIgnore, bellWrap, bell, bellCount, inboxPanel, inboxList, inboxEmpty, inboxRead] = ["notify", "toasts", "chat-ignore", "bell-wrap", "bell", "bell-count", "inbox-panel", "inbox-list", "inbox-empty", "inbox-read"].map((id) => document.getElementById(id));
+  let inbox = null;
+  let inboxText = "";
+  let inboxBusy = false;
+  let inboxSignature = "";
+  let seen = loadStored(SEEN_KEY);
+  const ignoredPeers = new Set(Object.keys(loadStored(IGNORED_KEY) || {}));
+
+  function loadStored(key) {
+    try {
+      const value = JSON.parse(window.localStorage.getItem(key));
+      return value && typeof value === "object" ? value : null;
+    } catch (error) {
+      return null;
+    }
+  }
+
+  function saveStored(key, value) {
+    try {
+      window.localStorage.setItem(key, JSON.stringify(value));
+    } catch (error) {
+      /* private mode: this is forgotten with the page */
+    }
+  }
+
+  const saveSeen = () => saveStored(SEEN_KEY, seen);
+  const isIgnored = (peerId) => ignoredPeers.has(peerId);
+  const unreadOf = (peerId) => {
+    const entry = inbox && !isIgnored(peerId) ? inbox[peerId] : null;
+    return entry ? Math.max(0, entry.incoming - ((seen && seen[peerId]) || 0)) : 0;
+  };
+  const peerName = (peerId) => {
+    const node = model && model.nodes.find((entry) => entry.id === peerId);
+    return node ? node.hostname : peerId.split("@")[0];
+  };
+
+  function markSeen(peerId) {
+    const entry = inbox && inbox[peerId];
+    if (!entry || !seen || seen[peerId] === entry.incoming) return;
+    seen[peerId] = entry.incoming;
+    saveSeen();
+    if (model) drawUnread();
+  }
+
+  function drawBell(total) {
+    bellWrap.hidden = !inbox;
+    if (!inbox) toggleInbox(false);
+    bellCount.hidden = !total;
+    bellCount.textContent = total > 99 ? "99+" : String(total);
+    bell.classList.toggle("unread", total > 0);
+    bell.setAttribute("aria-label", total ? "Messages, " + C.plural(total, "unread message") : "Messages");
+    drawInbox();
+  }
+
+  function drawInbox() {
+    if (inboxPanel.hidden || !inbox) return;
+    const conversations = Object.keys(inbox)
+      .filter((peerId) => inbox[peerId].last && !isIgnored(peerId))
+      .sort((a, b) => (inbox[b].last.at || "").localeCompare(inbox[a].last.at || ""));
+    const signature = JSON.stringify(conversations.map((peerId) => [peerId, peerName(peerId), unreadOf(peerId), inbox[peerId].last]));
+    if (signature === inboxSignature) return;
+    inboxSignature = signature;
+    inboxList.replaceChildren();
+    conversations.forEach((peerId) => {
+      const unread = unreadOf(peerId);
+      const last = inbox[peerId].last;
+      const row = html("button", undefined, "inbox-row" + (unread ? " unread" : ""));
+      row.type = "button";
+      row.appendChild(html("span", peerName(peerId), "inbox-name"));
+      const time = html("time", chatTime(last.at), "inbox-time");
+      time.dateTime = last.at;
+      row.appendChild(time);
+      row.appendChild(html("span", last.text, "inbox-text"));
+      if (unread) row.appendChild(html("span", unread > 99 ? "99+" : String(unread), "inbox-count"));
+      row.setAttribute("aria-label", peerName(peerId) + (unread ? ", " + C.plural(unread, "unread message") : "") + ": " + last.text);
+      row.addEventListener("click", () => openChat(peerId));
+      const item = html("li");
+      item.appendChild(row);
+      inboxList.appendChild(item);
+    });
+    inboxEmpty.hidden = conversations.length > 0;
+    inboxRead.disabled = !conversations.some((peerId) => unreadOf(peerId));
+  }
+
+  function toggleInbox(open) {
+    if (open === !inboxPanel.hidden) return;
+    inboxPanel.hidden = !open;
+    inboxSignature = "";
+    bell.setAttribute("aria-expanded", open ? "true" : "false");
+    if (open) {
+      toastBox.replaceChildren();
+      updateNotifyButton();
+      drawInbox();
+    }
+  }
+
+  function markAllRead() {
+    Object.keys(inbox).forEach((peerId) => (seen[peerId] = inbox[peerId].incoming));
+    saveSeen();
+    drawUnread();
+    drawStats();
+  }
+
+  function openChat(peerId) {
+    const node = model && model.nodes.find((entry) => entry.id === peerId);
+    toggleInbox(false);
+    if (!node) return;
+    focusKey = node.key;
+    if (!isPinned(node.key)) select({ kind: "node", id: node.key });
+    chatInput.focus();
+  }
+
+  function showToast(peerId, name, text) {
+    const toast = html("button", undefined, "toast");
+    toast.type = "button";
+    toast.appendChild(html("strong", name));
+    toast.appendChild(html("span", text));
+    toast.addEventListener("click", () => {
+      toast.remove();
+      openChat(peerId);
+    });
+    toastBox.appendChild(toast);
+    while (toastBox.children.length > TOAST_LIMIT) toastBox.firstChild.remove();
+    setTimeout(() => toast.remove(), TOAST_MS);
+  }
+
+  function notifyMessage(peerId, message, count) {
+    const name = peerName(peerId);
+    const text = count > 1 ? C.plural(count, "new message") + ", last: " + message.text : message.text;
+    if (inboxPanel.hidden) showToast(peerId, name, text);
+    if (!("Notification" in window) || Notification.permission !== "granted" || (!document.hidden && document.hasFocus())) return;
+    const notification = new Notification("Message from " + name, { body: text, tag: "colony-chat-" + peerId });
+    notification.addEventListener("click", () => {
+      window.focus();
+      openChat(peerId);
+      notification.close();
+    });
+  }
+
+  function handleInbox(next) {
+    const previous = inbox;
+    inbox = next;
+    if (!seen) {
+      seen = {};
+      Object.keys(next).forEach((peerId) => (seen[peerId] = next[peerId].incoming));
+      saveSeen();
+    }
+    Object.keys(next).forEach((peerId) => {
+      const entry = next[peerId];
+      const before = previous ? (previous[peerId] || { incoming: 0 }).incoming : entry.incoming;
+      if (entry.incoming <= before || isIgnored(peerId) || !entry.last) return;
+      if (chat.peer === peerId && !document.hidden) return;
+      notifyMessage(peerId, entry.last, entry.incoming - before);
+    });
+    if (model) {
+      drawUnread();
+      drawStats();
+      updateChat();
+    }
+  }
+
+  async function pollInbox() {
+    if (inboxBusy) return;
+    inboxBusy = true;
+    try {
+      const response = await fetch("/api/chat/inbox", { cache: "no-store" });
+      if (response.status === 409 && inbox) {
+        inboxText = "";
+        handleInbox(null);
+      }
+      if (!response.ok) return;
+      const text = await response.text();
+      if (text === inboxText) return;
+      inboxText = text;
+      handleInbox(JSON.parse(text));
+    } catch (error) {
+      /* the status refresh already shows when the node cannot be reached */
+    } finally {
+      inboxBusy = false;
+    }
+  }
+
+  function updateNotifyButton() {
+    notifyButton.hidden = !("Notification" in window) || !window.isSecureContext || Notification.permission !== "default";
+  }
+
+  bell.addEventListener("click", () => toggleInbox(inboxPanel.hidden));
+  inboxRead.addEventListener("click", markAllRead);
+  bellWrap.addEventListener("keydown", (event) => {
+    if (event.key !== "Escape" || inboxPanel.hidden) return;
+    event.preventDefault();
+    toggleInbox(false);
+    bell.focus();
+  });
+  document.addEventListener("pointerdown", (event) => {
+    if (!inboxPanel.hidden && !bellWrap.contains(event.target)) toggleInbox(false);
+  });
+
+  notifyButton.addEventListener("click", async () => {
+    await Notification.requestPermission();
+    updateNotifyButton();
+  });
+
+  chatIgnore.addEventListener("click", () => {
+    const peerId = chat.peer;
+    if (!peerId) return;
+    const ignored = !isIgnored(peerId);
+    if (ignored) ignoredPeers.add(peerId);
+    else ignoredPeers.delete(peerId);
+    saveStored(IGNORED_KEY, Object.fromEntries(Array.from(ignoredPeers, (id) => [id, true])));
+    if (!ignored) markSeen(peerId);
+    announce((ignored ? "Ignoring " : "No longer ignoring ") + chat.hostname);
+    drawUnread();
+    drawStats();
+    updateChat();
+  });
+
   chatClose.addEventListener("click", () => select(null));
   chatPanel.addEventListener("keydown", (event) => {
     if (event.key !== "Escape") return;
@@ -988,10 +1241,15 @@
   }
 
   document.addEventListener("visibilitychange", () => {
-    if (!document.hidden) refresh();
+    if (document.hidden) return;
+    refresh();
+    if (chat.peer && chat.loaded) markSeen(chat.peer);
   });
   svgRoot.setAttribute("viewBox", "0 0 " + size.w + " " + size.h);
   updateControls();
+  updateNotifyButton();
   refresh();
   setInterval(refresh, REFRESH_MS);
+  pollInbox();
+  setInterval(pollInbox, REFRESH_MS);
 })();

colony/core/roles/webview/static/index.html

@@ -16,6 +16,22 @@
     <div class="headline">
       <h1>Colony</h1>
       <p id="summary" class="summary">Loading...</p>
+      <div id="bell-wrap" class="bell-wrap" hidden>
+        <button id="bell" type="button" class="bell" aria-expanded="false" aria-controls="inbox-panel" aria-label="Messages" title="Messages">
+          <svg viewBox="0 0 20 20" aria-hidden="true"><path d="M5 14V9a5 5 0 0 1 10 0v5l1.5 1.5h-13zM8.2 17.5a2 2 0 0 0 3.6 0"/></svg>
+          <span id="bell-count" class="bell-count" hidden></span>
+        </button>
+        <div id="toasts" class="toasts" aria-live="polite"></div>
+        <section id="inbox-panel" class="inbox-panel" aria-labelledby="inbox-title" hidden>
+          <div class="inbox-head">
+            <h2 id="inbox-title">Messages</h2>
+            <button id="inbox-read" type="button">Mark all read</button>
+          </div>
+          <ol id="inbox-list" class="inbox-list"></ol>
+          <p id="inbox-empty" class="inbox-empty muted" hidden>No messages yet.</p>
+          <button id="notify" type="button" class="notify" hidden>Turn on desktop notifications</button>
+        </section>
+      </div>
     </div>
     <div id="alerts" class="alerts"></div>
     <p id="error" class="error" hidden></p>
@@ -69,6 +85,7 @@
     <aside id="chat" class="chat" aria-labelledby="chat-title" hidden>
       <div class="chat-head">
         <h2 id="chat-title"></h2>
+        <button id="chat-ignore" type="button" class="chat-ignore" aria-pressed="false" hidden>Ignore</button>
         <button id="chat-close" type="button" aria-label="Close the chat" title="Close the chat (Escape)"><svg viewBox="0 0 20 20" aria-hidden="true"><path d="M5 5l10 10M15 5L5 15"/></svg></button>
       </div>
       <p id="chat-status" class="chat-status" hidden></p>

colony/core/roles/webview/static/style.css

@@ -242,6 +242,54 @@ p { margin: 4px 0; }
 #stage.chatting .controls { right: 364px; }
 @media (pointer: coarse) { .chat-form input, .chat-form button, .chat-head button { min-height: 44px; } .chat-head button { min-width: 44px; } }
 
+.chat-head .chat-ignore { min-width: 0; padding: 0 10px; border: 1px solid var(--line); font: inherit; font-size: 13px; color: var(--muted); }
+.chat-head .chat-ignore[aria-pressed="true"] { border-color: var(--crit); color: var(--crit); }
+.chat-head button:focus-visible { outline: 2px solid var(--text); outline-offset: 2px; }
+
+.badge { pointer-events: none; }
+.badge circle { fill: var(--crit); stroke: var(--card); stroke-width: 2; vector-effect: non-scaling-stroke; }
+.badge text { fill: #fff; font-size: 10.5px; font-weight: 700; font-variant-numeric: tabular-nums; }
+
+.bell-wrap { position: relative; z-index: 5; margin-left: auto; align-self: center; margin-top: 8px; }
+.bell { position: relative; display: inline-flex; align-items: center; justify-content: center; width: 38px; height: 38px; padding: 0; border: 1px solid var(--line); border-radius: 10px; background: var(--card); color: var(--text); cursor: pointer; }
+.bell:hover, .bell[aria-expanded="true"] { border-color: var(--muted); }
+.bell:focus-visible { outline: 2px solid var(--text); outline-offset: 2px; }
+.bell svg { width: 20px; height: 20px; fill: none; stroke: currentColor; stroke-width: 1.6; stroke-linecap: round; stroke-linejoin: round; }
+.bell.unread svg { stroke: var(--n0); }
+.bell-count { position: absolute; top: -6px; right: -6px; min-width: 19px; height: 19px; padding: 0 5px; border: 2px solid var(--bg); border-radius: 999px; background: var(--crit); color: #fff; font-size: 11px; font-weight: 700; line-height: 15px; text-align: center; font-variant-numeric: tabular-nums; }
+
+.toasts, .inbox-panel { position: absolute; top: calc(100% + 8px); right: 0; width: min(340px, calc(100vw - 24px)); }
+.toasts { display: flex; flex-direction: column; gap: 8px; pointer-events: none; }
+.toast { display: block; width: 100%; padding: 10px 14px; border: 1px solid var(--line); border-left: 4px solid var(--n0); border-radius: 10px; background: var(--card); color: var(--text); font: inherit; text-align: left; cursor: pointer; pointer-events: auto; box-shadow: 0 6px 24px color-mix(in srgb, var(--text) 14%, transparent); }
+.toast:hover { border-color: var(--muted); border-left-color: var(--n0); }
+.toast:focus-visible { outline: 2px solid var(--text); outline-offset: 2px; }
+.toast strong { display: block; font-size: 13px; }
+.toast span { display: block; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; color: var(--muted); font-size: 13px; }
+
+.inbox-panel { display: flex; flex-direction: column; max-height: min(480px, 70dvh); border: 1px solid var(--line); border-radius: 10px; background: var(--card); box-shadow: 0 10px 32px color-mix(in srgb, var(--text) 16%, transparent); }
+.inbox-head { display: flex; align-items: center; gap: 8px; padding: 8px 8px 8px 14px; border-bottom: 1px solid var(--line); }
+.inbox-head h2 { flex: 1; margin: 0; font-size: 15px; }
+.inbox-head button, .notify { height: 30px; padding: 0 10px; border: 1px solid var(--line); border-radius: 8px; background: transparent; color: var(--muted); font: inherit; font-size: 13px; cursor: pointer; }
+.inbox-head button:hover, .notify:hover { border-color: var(--muted); color: var(--text); }
+.inbox-head button:disabled { opacity: 0.5; cursor: default; }
+.inbox-head button:focus-visible, .notify:focus-visible { outline: 2px solid var(--text); outline-offset: 2px; }
+.inbox-list { flex: 1; min-height: 0; overflow-y: auto; margin: 0; padding: 4px; list-style: none; }
+.inbox-row { display: grid; grid-template-columns: 1fr auto; gap: 1px 10px; width: 100%; padding: 8px 10px; border: 0; border-radius: 8px; background: transparent; color: var(--text); font: inherit; text-align: left; cursor: pointer; }
+.inbox-row:hover { background: color-mix(in srgb, var(--text) 6%, transparent); }
+.inbox-row:focus-visible { outline: 2px solid var(--text); outline-offset: -2px; }
+.inbox-name { font-size: 14px; font-weight: 600; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; }
+.inbox-time { color: var(--muted); font-size: 12px; font-variant-numeric: tabular-nums; }
+.inbox-text { color: var(--muted); font-size: 13px; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; }
+.inbox-count { justify-self: end; min-width: 19px; height: 19px; padding: 0 5px; border-radius: 999px; background: var(--n0); color: var(--card); font-size: 11px; font-weight: 700; line-height: 19px; text-align: center; }
+.inbox-row.unread .inbox-text { color: var(--text); }
+.inbox-empty { margin: 0; padding: 16px 14px; text-align: center; }
+.notify { margin: 0 10px 10px; color: var(--n0); border-color: color-mix(in srgb, var(--n0) 50%, transparent); }
+
+@media (prefers-reduced-motion: no-preference) {
+  .toast, .inbox-panel { animation: drop-in 0.18s ease-out; }
+}
+@keyframes drop-in { from { opacity: 0; transform: translateY(-6px); } }
+
 .sr-only { position: absolute; width: 1px; height: 1px; margin: -1px; padding: 0; overflow: hidden; clip: rect(0 0 0 0); white-space: nowrap; border: 0; }
 
 /* Phones: the desktop sizes are small on a 400px screen, so the page text and touch targets grow. */

colony/core/storages/chat.py

@@ -1,7 +1,7 @@
 import datetime
 import os
 import threading
-from urllib.parse import quote
+from urllib.parse import quote, unquote
 
 from colony.core.storages.core import Storage, STORAGES_DIR
 
@@ -19,11 +19,32 @@ class ChatStorage(Storage):
         return quote(peer_id, safe="") + ".json"
 
     def messages(self, peer_id):
-        return self.open(self._file(peer_id)) or []
+        messages = self.open(self._file(peer_id)) or []
+        received = 0
+        for seq, message in enumerate(messages, 1):
+            received = message.setdefault("received", received + (message["direction"] == "in"))
+            message.setdefault("seq", seq)
+        return messages
 
-    def append(self, peer_id, direction, text):
+    def append(self, peer_id, direction, text, keep=None):
         at = datetime.datetime.now(datetime.timezone.utc).isoformat()
         with self._lock:
             messages = self.messages(peer_id)
-            messages.append({"direction": direction, "text": text, "at": at})
+            last = messages[-1] if messages else {"seq": 0, "received": 0}
+            messages.append({
+                "direction": direction,
+                "text": text,
+                "at": at,
+                "seq": last["seq"] + 1,
+                "received": last["received"] + (direction == "in"),
+            })
+            if keep:
+                messages = messages[-keep:]
             self.save(self._file(peer_id), messages)
+
+    def peers(self):
+        peers = []
+        for filename in self.ls(""):
+            peer_id = unquote(filename.removesuffix(".json"))
+            peers.append(peer_id)
+        return sorted(peers)

colony/settings.py

@@ -6,54 +6,13 @@ configuration = Dynaconf(
     load_dotenv=False,
 )
 
-DEFAULTS = {
+MISSING = object()
-    "discovery.range_start": "192.168.65.110",
-    "discovery.range_end": "192.168.65.120",
-    "discovery.discover_port": 28149,
-    "discovery.probe_timeout": 1.0,
-    "discovery.probe_workers": 64,
-    "membership.gossip_interval": 5,
-    "membership.member_timeout": 20,
-    "membership.reprobe_interval": 15,
-    "membership.fanout": 3,
-    "membership.forget_after": 600,
-    "membership.hello_timeout": 3,
-    "membership.max_members": 1024,
-    "membership.admission": "open",
-    "foreman.interval": 5,
-    "foreman.acquire_delay": 10,
-    "foreman.proposal_stagger": 2,
-    "foreman.proposal_cooldown": 30,
-    "gateway.client_fallback": True,
-    "gateway.connect_timeout": 2,
-    "gateway.direct_retry": 30,
-    "gateway.max_attempts": 4,
-    "gateway.forward_timeout": 8,
-    "gateway.max_concurrent": 16,
-    "gateway.max_payload_bytes": 4 * 1024 * 1024,
-    "gateway.allow_commands": [
-        "Gossip.Hello",
-        "Unit.ShowStats",
-        "Unit.ShowRoles",
-        "Unit.ShowHardware",
-        "Unit.ShowCluster",
-        "Unit.RequestRole",
-        "Replicator.HashSum",
-        "Replicator.Retrieve",
-        "Replicator.Replicate",
-        "Chatter.Receive",
-    ],
-    "web.host": "0.0.0.0",
-    "web.port": 8080,
-    "roles.on_demand": [],
-    "role_policy.max_roles": None,
-    "role_policy.deny": [],
-    "role_policy.min_scores": {},
-    "role_policy.decline_cooldown": 60,
-}
 
 
 def setting(key, override=None):
     if override is not None:
         return override
-    return configuration.get(key, DEFAULTS[key])
+    value = configuration.get(key, MISSING)
+    if value is MISSING:
+        raise KeyError(f"Setting {key} is not configured")
+    return value

colony/transport/control.py

@@ -0,0 +1,101 @@
+import json
+import os
+import queue
+import socket
+import threading
+from pathlib import Path
+
+from colony.core.logger.logging import get_logger
+from colony.transport.framing import send_frame, recv_frame
+
+CONTROL_SOCKET = Path(__file__).resolve().parents[2] / "data" / "colony.sock"
+
+
+def request(payload, path=CONTROL_SOCKET, timeout=10):
+    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
+        sock.settimeout(timeout)
+        sock.connect(str(path))
+        send_frame(sock, json.dumps(payload).encode())
+        return json.loads(recv_frame(sock).decode())
+
+
+class ControlQueue:
+    def __init__(self, actions):
+        self.logger = get_logger(self.__class__.__name__)
+        self.actions = actions
+        self.tasks = queue.Queue()
+
+    def submit(self, payload, timeout):
+        reply = queue.Queue(maxsize=1)
+        self.tasks.put((payload, reply))
+        return reply.get(timeout=timeout)
+
+    def run_once(self):
+        payload, reply = self.tasks.get()
+        try:
+            body = json.dumps({"status": "ok", "result": self.actions.run(payload)})
+        except Exception as e:
+            self.logger.exception("Control command failed")
+            body = json.dumps({"status": "error", "error": f"{type(e).__name__}: {e}"})
+        reply.put(body.encode())
+
+    def run_forever(self):
+        while True:
+            self.run_once()
+
+
+class ControlSocket:
+    CLIENT_TIMEOUT = 30
+
+    def __init__(self, control_queue, path=CONTROL_SOCKET):
+        self.logger = get_logger(self.__class__.__name__)
+        self.control_queue = control_queue
+        self.path = Path(path)
+        self._sock = None
+
+    def _remove_stale_socket(self):
+        if not self.path.exists():
+            return
+        with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as probe:
+            try:
+                probe.connect(str(self.path))
+            except OSError:
+                self.path.unlink()
+                return
+        raise OSError(f"another node is already running on {self.path}")
+
+    def bind(self):
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self._remove_stale_socket()
+        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        try:
+            sock.bind(str(self.path))
+            os.chmod(self.path, 0o600)
+            sock.listen()
+        except OSError:
+            sock.close()
+            raise
+        self._sock = sock
+
+    def start(self):
+        self.logger.info(f"Control socket on {self.path}")
+        while True:
+            conn, _ = self._sock.accept()
+            threading.Thread(target=self.handle, args=(conn,), daemon=True, name="HandleControl").start()
+
+    def handle(self, conn):
+        with conn:
+            try:
+                conn.settimeout(self.CLIENT_TIMEOUT)
+                payload = json.loads(recv_frame(conn).decode())
+                body = self.control_queue.submit(payload, self.CLIENT_TIMEOUT)
+            except queue.Empty:
+                body = json.dumps({"status": "error", "error": "timed out waiting for the node"}).encode()
+            except (OSError, ValueError) as e:
+                self.logger.warning(f"Bad control request: {e}")
+                return
+
+            try:
+                send_frame(conn, body)
+            except OSError as e:
+                self.logger.warning(f"Could not answer control request: {e}")

config.json

@@ -34,9 +34,14 @@
     "proposal_stagger": 2,
     "proposal_cooldown": 30
   },
+  "chat": {
+    "max_messages": 1000
+  },
   "web": {
     "host": "0.0.0.0",
-    "port": 8080
+    "port": 8080,
+    "username": "colony",
+    "password": ""
   },
   "explorer": {
     "hosts": []

docker-compose.yml

@@ -35,8 +35,12 @@ services:
       COLONY_FOREMAN__PROPOSAL_STAGGER: 2
       COLONY_FOREMAN__PROPOSAL_COOLDOWN: 30
 
+      COLONY_CHAT__MAX_MESSAGES: 1000
+
       COLONY_WEB__HOST: 0.0.0.0
       COLONY_WEB__PORT: 8271
+      COLONY_WEB__USERNAME: colony
+      COLONY_WEB__PASSWORD: change-me
 
       COLONY_EXPLORER__HOSTS: "[]"
 

main.py

@@ -3,6 +3,7 @@ import threading
 
 from colony.core.logger.logging import get_logger
 from colony.core.network.core import Network
+from colony.core.control import ControlActions
 from colony.core.dispatcher import Dispatcher
 from colony.core.roles.foreman.role import Foreman
 from colony.core.roles.gossip.role import Gossip, BOOTSTRAPPED
@@ -10,6 +11,7 @@ from colony.core.roles.manager import RoleManager
 from colony.core.state_manager import StateManager
 from colony.role_registry import RoleRegistry
 from colony.settings import configuration, setting
+from colony.transport.control import ControlQueue, ControlSocket
 from colony.transport.discover import Discover
 from colony.transport.listener import Listener
 from colony.utils.network import hosts_in_range
@@ -49,7 +51,20 @@ def main():
         logger.error(f"Cannot listen on {listener.host}:{listener.port}: {e}")
         sys.exit(1)
 
-    threads = []
+    control = ControlQueue(ControlActions(network))
+    control_socket = ControlSocket(control)
+    try:
+        control_socket.bind()
+    except OSError as e:
+        logger.error(f"Cannot open the control socket {control_socket.path}: {e}")
+        sys.exit(1)
+
+    t = threading.Thread(
+        target=control_socket.start,
+        daemon=True,
+        name="ControlSocket"
+    )
+    t.start()
 
     t = threading.Thread(
         target=listener.start,
@@ -57,7 +72,6 @@ def main():
         name="Listener"
     )
     t.start()
-    threads.append(t)
 
     t = threading.Thread(
         target=Discover.listen,
@@ -66,7 +80,6 @@ def main():
         name="Discover"
     )
     t.start()
-    threads.append(t)
 
     roles = role_manager.get_role_instances()
     gossip: Gossip = roles["Gossip"]
@@ -81,8 +94,7 @@ def main():
 
     role_manager.start_all()
 
-    for t in threads:
+    control.run_forever()
-        t.join()
 
 
 if __name__ == "__main__":