dashboard: document follows changes

1 files changed, 67 insertions, 8 deletions

diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs
index 99a0b83..7679787 100644
--- a/src/api/dashboard/mod.rs
+++ b/src/api/dashboard/mod.rs
@@ -56,6 +56,9 @@ struct Hub {
     latest: watch::Sender<Option<Arc<Published>>>,
     /// Each patch, to the clients connected.
     patches: broadcast::Sender<Arc<Published>>,
+    /// Where the document changed since the last tick, to the clients connected:
+    /// `{"paths": [...]}`, or `{"all": true}` if too many or not known.
+    documents: broadcast::Sender<Arc<String>>,
     control: Mutex<Control>,
     /// True once the node stops: every client gets an `offline` event, and its stream
     /// ends (so the HTTP server needn't wait for clients to go).
@@ -72,10 +75,12 @@ impl Hub {
     fn new(state: AppState, closing: watch::Receiver<bool>) -> Arc<Self> {
         let (latest, _) = watch::channel(None);
         let (patches, _) = broadcast::channel(64);
+        let (documents, _) = broadcast::channel(64);
         Arc::new(Self {
             state,
             latest,
             patches,
+            documents,
             control: Mutex::new(Control {
                 running: false,
                 wanted_until: Instant::now(),
@@ -120,6 +125,7 @@ impl Hub {
         let mut seq = 0u64;
         let mut graph = Graph::default();
         let mut graph_due = Instant::now();
+        let mut changes = self.state.change_events.subscribe();
         loop {
             if Instant::now() >= graph_due {
                 graph.refresh(&self.state).await;
@@ -146,6 +152,9 @@ impl Hub {
                 last = Some(snapshot);
             }
             tokio::time::sleep(TICK).await;
+            if let Some(changed) = document_changes(&mut changes) {
+                let _ = self.documents.send(Arc::new(changed.to_string()));
+            }
             {
                 let mut control = self.control.lock_or_recover();
                 let unwanted =
@@ -159,6 +168,41 @@ impl Hub {
     }
 }
 
+/// Paths changed in a tick listed at most; more: `all`.
+const MAX_CHANGED: usize = 256;
+
+/// Where the document changed since last asked (`Hub::documents`), if anywhere.
+fn document_changes(
+    changes: &mut broadcast::Receiver<Arc<crate::swarm::sync::ChangeSet>>,
+) -> Option<JsonValue> {
+    let mut paths = BTreeSet::new();
+    let mut all = false;
+    loop {
+        match changes.try_recv() {
+            Ok(set) => {
+                paths.extend(set.paths.iter().cloned());
+                all |= paths.len() > MAX_CHANGED;
+            }
+            Err(broadcast::error::TryRecvError::Lagged(_)) => all = true,
+            Err(_) => break,
+        }
+    }
+    if all {
+        Some(json!({ "all": true }))
+    } else if paths.is_empty() {
+        None
+    } else {
+        Some(json!({ "paths": paths }))
+    }
+}
+
+/// What a client's stream waits for.
+enum Next {
+    Patch(Result<Arc<Published>, broadcast::error::RecvError>),
+    Document(Result<Arc<String>, broadcast::error::RecvError>),
+    Closing,
+}
+
 /// The intended graph, recomputed every `LINKS_EVERY` (it costs about `members²`), when
 /// members are asked for their links too.
 #[derive(Default)]
@@ -360,11 +404,13 @@ async fn current_state(Extension(hub): Extension<Arc<Hub>>) -> Response {
 
 /// `GET /admin/dashboard/events`: Server-Sent Events: a `snapshot` first, then a
 /// `patch` whenever something changed (each with its number as the event's ID; a
-/// client that fell behind gets a `snapshot` again), and `offline` when the node
-/// stops, which ends the stream.
+/// client that fell behind gets a `snapshot` again), a `document` where the document
+/// changed (`Hub::documents`, at most every tick), and `offline` when the node stops,
+/// which ends the stream.
 async fn events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoResponse {
     // Subscribed before the snapshot is read: no patch after it is missed.
     let mut patches = hub.patches.subscribe();
+    let mut documents = hub.documents.subscribe();
     hub.want();
     let first = hub.current().await;
     let stream = async_stream::stream! {
@@ -375,13 +421,26 @@ async fn events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoResponse {
         let mut seen = first.seq;
         yield Ok(snapshot_event(&first));
         loop {
-            let received = tokio::select! {
-                received = patches.recv() => Some(received),
-                () = hub.closed() => None,
+            let next = tokio::select! {
+                received = patches.recv() => Next::Patch(received),
+                changed = documents.recv() => Next::Document(changed),
+                () = hub.closed() => Next::Closing,
             };
-            let Some(received) = received else {
-                yield Ok(offline_event());
-                break;
+            let received = match next {
+                Next::Patch(received) => received,
+                Next::Document(Ok(changed)) => {
+                    yield Ok(Event::default().event("document").data(changed.as_str()));
+                    continue;
+                }
+                Next::Document(Err(broadcast::error::RecvError::Lagged(_))) => {
+                    yield Ok(Event::default().event("document").data(r#"{"all":true}"#));
+                    continue;
+                }
+                Next::Document(Err(broadcast::error::RecvError::Closed)) => break,
+                Next::Closing => {
+                    yield Ok(offline_event());
+                    break;
+                }
             };
             match received {
                 Ok(patch) if patch.seq <= seen => {}