dashboard: subscribed nodes' messages in the console

Each tick's messages of a subscribed node, listed in the console with
this node's own, from its own stream.

Auth plan M3 (live data from every reachable node).

2 files changed, 136 insertions, 71 deletions

diff --git a/src/api/dashboard/dashboard.html b/src/api/dashboard/dashboard.html
index efbf892..3f02a64 100644
--- a/src/api/dashboard/dashboard.html
+++ b/src/api/dashboard/dashboard.html
@@ -774,11 +774,25 @@ function renderStats() {
     // Drawn once: a message doesn't change, and drawing it anew would scroll it back up.
     if (stats.dataset.message === String(message.seq)) return;
     stats.dataset.message = String(message.seq);
-    what.textContent = message.type + " " + WAYS[message.way] + " " + name(message.peer);
+    what.textContent = message.node
+      ? message.type + " " + name(message.to) + " ← " + name(message.from)
+      : message.type + " " + WAYS[message.way] + " " + name(message.peer);
     const field = (key, value) => el("div", { class: "leaf" },
       el("span", { class: "key" }, key + ": "), value);
     title.replaceChildren("Message: ", what);
     retitle(title, title.textContent);
+    // A subscribed node's: logged there, the message itself not read from it.
+    if (message.node) {
+      document.getElementById("stats").replaceChildren(
+        field("at", clockTime(new Date(message.at))),
+        field("from", message.from),
+        field("to", message.to),
+        field("logged by", message.node),
+        field("type", message.type),
+        field("bytes", String(message.bytes)),
+        el("div", { class: "note" }, "Written out only on the node that logged it (its console)."));
+      return;
+    }
     document.getElementById("stats").replaceChildren(
       field("at", clockTime(new Date(message.at))),
       field(message.way === "received" ? "from" : "to", message.peer),
@@ -1224,7 +1238,8 @@ function renderGraph() {
   const marking = new Map();
   for (const [message, how] of [[pointedMessage, "pointed"], [shownMessage, "shown"]]) {
     if (message === null) continue;
-    for (const key of [state.node.id, message.peer, pairKey(state.node.id, message.peer)]) {
+    const [a, b] = message.node ? [message.from, message.to] : [state.node.id, message.peer];
+    for (const key of [a, b, pairKey(a, b)]) {
       marking.set(key, how);
     }
   }
@@ -2292,9 +2307,12 @@ function follow(id, delay = 2000) {
   const member = state && state.graph.members[id];
   if (!member || !member.dashboard) return fail("it publishes no dashboard (--dashboard-url)");
   if (!narrowToken) return fail(tokenWhy);
+  // Its endpoints, beside its page (`…/admin/dashboard` → `…/admin/dashboard/…`).
+  const base = member.dashboard.split("#")[0].replace(/\/+$/, "");
+  const at = path => new URL(path, base).href;
   let url;
   try {
-    url = new URL("dashboard/links/own/events", member.dashboard.split("#")[0].replace(/\/+$/, "")).href;
+    url = at("dashboard/links/own/events");
   } catch (_) {
     return fail("its dashboard's address isn't one");
   }
@@ -2310,25 +2328,19 @@ function follow(id, delay = 2000) {
       render();
     }
   }).catch(error => fail(error.message || String(error)));
-  // Its traffic, for its messages on the map: while it goes, quietly; a failure of its
-  // own leaves the links be.
-  let trafficUrl;
-  try {
-    trafficUrl = new URL("dashboard/traffic/own/events", url).href;
-  } catch (_) {
-    return;
-  }
-  const followTraffic = (wait = 2000) => {
+  // Its messages, for the console and the map: while they come, quietly; a failure of
+  // their own leaves the links be.
+  const messagesUrl = at("dashboard/messages/own/events");
+  const followMessages = (wait = 2000) => {
     if (controller.signal.aborted) return;
-    read(trafficUrl, controller.signal, () => { entry.trafficLive = true; wait = 2000; },
-      (kind, data) => { if (kind === "traffic") remoteTrafficArrived(id, entry, JSON.parse(data)); })
+    read(messagesUrl, controller.signal, () => { entry.messagesLive = true; wait = 2000; },
+      (kind, data) => { if (kind === "messages") remoteMessagesArrived(id, entry, JSON.parse(data)); })
       .catch(() => {
-        entry.trafficLive = false;
-        entry.traffic = null;
-        if (!controller.signal.aborted) setTimeout(() => followTraffic(Math.min(60000, wait * 2)), wait);
+        entry.messagesLive = false;
+        if (!controller.signal.aborted) setTimeout(() => followMessages(Math.min(60000, wait * 2)), wait);
       });
   };
-  followTraffic();
+  followMessages();
 }
 // Server-sent events from `url`, read with this page's narrow token (`EventSource`
 // can't send one): `opened` once answered, `event(kind, data)` for each; rejects when it
@@ -2644,8 +2656,9 @@ graph.addEventListener("mousemove", event => { if (event.target === graph) hover
 graph.addEventListener("mouseleave", () => hover(null));
 
 // --- Messages on the map: this node's traffic with each peer (`/peers`, `traffic`),
-// and each subscribed node's with its peers (`remoteTrafficArrived`), as it grew since
-// the last update, drawn as dots running along the link the way the messages went. Counts come with each update (every 2 s), so the dots are spread
+// as it grew since the last update, and each subscribed node's messages with its peers
+// (`remoteMessagesArrived`), drawn as dots running along the link the way the messages
+// went. Counts come with each update (every 2 s), so the dots are spread
 // over the next 2 s, a few at most per link and way: a picture of the traffic, not a
 // record of it. All at one speed (`flowSpeed`), so over a longer link they arrive
 // later. Kept above everything else on the map (`flows`, after what `show` places). ---
@@ -2839,24 +2852,49 @@ function dotsNow(gap) {
     !calm.matches && !document.hidden &&
     !document.getElementById("map").classList.contains("collapsed");
 }
-// A subscribed node's traffic (`followTraffic`), as `trafficArrived` this node's: its
-// messages with this node are this node's to draw, and what it received from another
-// subscribed node that one's (as what it sent).
-function remoteTrafficArrived(id, entry, traffic) {
+// A subscribed node's messages (`follow`), those of a tick: listed in the console and
+// run along its links, as this node's own. Its messages with this node are this node's
+// to show, and what it received from another subscribed node that one's (as what it
+// sent): each message once. Each named by both ends (`from`, `to`) and the node that
+// logged it (`node`), its number and exchange made its own (`node:seq`).
+function remoteMessagesArrived(id, entry, list) {
   if (!state) return;
-  const current = new Map();
-  for (const [peer, counts] of Object.entries(traffic)) {
-    current.set(peer, { sent: messages(counts.sent), received: messages(counts.received) });
+  const self = state.node.id;
+  const wall = Date.now(), gap = wall - (entry.messagesAt || 0);
+  entry.messagesAt = wall;
+  const kept = [];
+  for (const message of list) {
+    if (message.way === "refused" || message.peer === self) continue;
+    if (message.way === "received" && message.peer !== id && subscriptions.has(message.peer) &&
+        (subscribed.get(message.peer) || {}).messagesLive) continue;
+    const [from, to] = message.way === "sent" ? [id, message.peer] : [message.peer, id];
+    kept.push({ ...message, seq: id + ":" + message.seq, node: id, from, to,
+      related: message.related ? id + ":" + message.related : null });
   }
-  const wall = Date.now(), gap = wall - (entry.trafficAt || 0);
-  entry.trafficAt = wall;
-  if (entry.traffic && dotsNow(gap)) {
-    const self = state.node.id;
-    addFlows(id, current, entry.traffic, (way, peer) => peer === self ||
-      (way === "received" && peer !== id && subscriptions.has(peer) &&
-        (subscribed.get(peer) || {}).trafficLive));
+  if (kept.length === 0) return;
+  if (dotsNow(gap)) {
+    const now = performance.now();
+    const links = new Map();
+    for (const message of kept) {
+      if (!linksShown && isLinks(message.counted)) continue;
+      const key = message.from + " " + message.to;
+      const link = links.get(key) || { from: message.from, to: message.to, kinds: {}, total: 0 };
+      const kind = link.kinds[message.counted] || (link.kinds[message.counted] = { messages: 0, bytes: 0 });
+      kind.messages += 1;
+      kind.bytes += message.bytes;
+      link.total += 1;
+      links.set(key, link);
+    }
+    for (const { from, to, kinds, total } of links.values()) {
+      const shown = dotsFor(kinds, total);
+      shown.forEach((dot, i) => {
+        flows.push({ from, to, start: now + i * FLOW_SPREAD_MS / shown.length,
+          kind: dot.kind, r: dotRadius(dot.bytes) });
+      });
+    }
+    if (flows.length > 0 && flowFrame === null) flowFrame = requestAnimationFrame(drawFlows);
   }
-  entry.traffic = current;
+  messagesArrived(kept);
 }
 
 // --- The console's rates: messages and bytes per second sent and received, per peer
@@ -2977,15 +3015,27 @@ function peerOf(key) {
   return a === self ? b : b === self ? a : null;
 }
 // A console line's tint for the node or link pointed at on the map, or in focus.
-function peerClasses(peer) {
-  return (peer === peerOf(focused) ? " peer-focused" : "") +
-    (peer === peerOf(hovered) ? " peer-hovered" : "");
+// Whether a message (its line's `data-…`: `peer` for this node's, `from` and `to` for a
+// subscribed node's) is one with the node or over the link `key` stands for.
+function about(data, key) {
+  if (key === null || !state) return false;
+  if (!data.peer) {
+    if (!data.from) return false;
+    return isPair(key) ? key === pairKey(data.from, data.to)
+      : key !== state.node.id && (data.from === key || data.to === key);
+  }
+  const peer = peerOf(key);
+  return peer !== null && data.peer === peer;
+}
+function peerClasses(message) {
+  const data = { peer: message.node ? "" : message.peer, from: message.from, to: message.to };
+  return (about(data, focused) ? " peer-focused" : "") +
+    (about(data, hovered) ? " peer-hovered" : "");
 }
 function markPeerMessages() {
-  const focusedPeer = peerOf(focused), hoveredPeer = peerOf(hovered);
   for (const line of document.getElementById("console").children) {
-    line.classList.toggle("peer-focused", focusedPeer !== null && line.dataset.peer === focusedPeer);
-    line.classList.toggle("peer-hovered", hoveredPeer !== null && line.dataset.peer === hoveredPeer);
+    line.classList.toggle("peer-focused", about(line.dataset, focused));
+    line.classList.toggle("peer-hovered", about(line.dataset, hovered));
   }
   filterConsole();
 }
@@ -3023,16 +3073,21 @@ function filterConsole() {
 }
 // A line: when, which way, with whom, the type (in its kind's colour) and the size.
 function consoleLine(message) {
+  // A subscribed node's (`remoteMessagesArrived`): its receiver, and whom from.
+  const between = message.node ? name(message.to) + " ← " + name(message.from) + " "
+    : WAYS[message.way] + " " + name(message.peer) + " ";
   const line = el("div", { "data-seq": String(message.seq),
     "data-related": message.related || "",
-    "data-peer": message.peer,
+    "data-peer": message.node ? "" : message.peer,
+    "data-from": message.from || "", "data-to": message.to || "",
     class: (shownMessage && shownMessage.seq === message.seq ? "focused" : "") +
       (isLinks(message.counted) ? " links-kind" : "") +
       (relatedMarked !== null && message.related === relatedMarked ? " related" : "") +
-      peerClasses(message.peer),
-    title: message.way === "refused" ? "refused: the transport wouldn't send it" : "" },
+      peerClasses(message),
+    title: message.way === "refused" ? "refused: the transport wouldn't send it"
+      : message.node ? "logged by " + name(message.node) + ", subscribed to" : "" },
     el("span", { class: "when" }, clockTime(new Date(message.at)) + " "),
-    WAYS[message.way] + " " + name(message.peer) + " ",
+    between,
     el("span", { style: "color: " + kindColour(message.counted) }, message.type),
     " " + size(message.bytes));
   // On press, as the peers' rows: lines come and go under the pointer.
diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs
index 1d7dbf7..7d940c9 100644
--- a/src/api/dashboard/mod.rs
+++ b/src/api/dashboard/mod.rs
@@ -540,8 +540,8 @@ pub(crate) fn router(state: AppState) -> Router<AppState> {
         .route("/admin/dashboard/links/own", get(own_links_now))
         .route("/admin/dashboard/links/own/events", get(own_links_events))
         .route(
-            "/admin/dashboard/traffic/own/events",
-            get(own_traffic_events),
+            "/admin/dashboard/messages/own/events",
+            get(own_messages_events),
         )
         .layer(Extension(Hub::new(state, closing)))
 }
@@ -598,31 +598,28 @@ async fn own_links_events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoRespo
     )
 }
 
-/// This node's traffic with each peer: by node ID, what it sent and received, by kind
-/// (messages and bytes counted since it started, as `/peers` has them).
-async fn own_traffic(state: &AppState) -> JsonValue {
-    let peers = state.peers_snapshot().await;
-    let traffic: Map<String, JsonValue> = peers
-        .iter()
-        .map(|(peer, _)| {
-            let counted = state
-                .traffic
-                .node(peer, &state.transport.identities_of(peer));
-            (
-                peer.clone(),
-                json!({ "sent": counted.sent, "received": counted.received }),
-            )
-        })
-        .collect();
-    JsonValue::Object(traffic)
+/// Stops looking at the message log when dropped (`own_messages_events`).
+struct Watching(Arc<Hub>);
+
+impl Drop for Watching {
+    fn drop(&mut self) {
+        self.0.state.traffic.log.watch(false);
+    }
 }
 
-/// `GET /admin/dashboard/traffic/own/events`: server-sent events, `traffic` with this
-/// node's traffic with each peer (`own_traffic`) every `TICK`, for a page of another
-/// member subscribed to this node to draw its messages on the map, as a page of this
-/// node draws its own from `/peers`; `offline` when the node stops, which ends it.
-async fn own_traffic_events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoResponse {
+/// `GET /admin/dashboard/messages/own/events`: server-sent events, `messages` with the
+/// messages this node sent and received since the last (every `TICK`, those of the
+/// tick, none too, as the page's own `messages` events have them, without each message
+/// written out), for a page of another member subscribed to this node: its console lists them,
+/// and its map runs them along this node's links. `offline` when the node stops, which
+/// ends it. The message log is kept while it is open.
+async fn own_messages_events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoResponse {
+    let log = &hub.state.traffic.log;
+    log.watch(true);
+    let watching = Watching(Arc::clone(&hub));
+    let mut logged = log.latest();
     let stream = async_stream::stream! {
+        let _watching = watching;
         let mut every = tokio::time::interval(TICK);
         loop {
             tokio::select! {
@@ -632,8 +629,21 @@ async fn own_traffic_events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoRes
                     break;
                 }
             }
-            let traffic = own_traffic(&hub.state).await;
-            yield Ok(Event::default().event("traffic").data(traffic.to_string()));
+            // Every tick, none too: the page times its dots by them.
+            let mut messages = hub.state.traffic.log.after(logged, MESSAGES_AT_MOST);
+            if let Some(last) = messages.last() {
+                logged = last.seq;
+                by_node(&hub.state, &mut messages).await;
+            }
+            let messages: Vec<JsonValue> = messages
+                .into_iter()
+                .filter_map(|message| {
+                    let mut json = serde_json::to_value(message).ok()?;
+                    json.as_object_mut()?.remove("text");
+                    Some(json)
+                })
+                .collect();
+            yield Ok(Event::default().event("messages").data(JsonValue::Array(messages).to_string()));
         }
     };
     (