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).

1 files changed, 36 insertions, 26 deletions

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()));
         }
     };
     (