1 files changed, 79 insertions, 0 deletions
diff --git a/src/swarm/changes.rs b/src/swarm/changes.rs
new file mode 100644
index 0000000..0f22f52
--- /dev/null
+++ b/src/swarm/changes.rs
@@ -0,0 +1,79 @@
+//! The change publisher: after document changes, which store paths they touched
+//! (`AppState::change_events`, for the API's streams, the dashboard and remakes), and
+//! a nudge to topology when the member list changed, and to the peer worker when the
+//! forget records did.
+
+use std::sync::Arc;
+
+use tracing::{info, warn};
+
+use crate::document::changes;
+use crate::document::store::{ChangeSet, InLists};
+use crate::state::AppState;
+
+pub(crate) async fn run_change_publisher(state: AppState) {
+    let mut changed = state.change_tx.subscribe();
+    // The replica's heads (sorted, as of its last refresh: every change refreshes them).
+    let mut last_heads = state.doc.read().await.heads().to_vec();
+
+    info!("Starting change publisher");
+
+    while changed.changed().await.is_ok() {
+        let before = last_heads.clone();
+
+        let doc = Arc::clone(&state.doc);
+        let (heads, paths, conflicts) = tokio::task::spawn_blocking(move || {
+            let replica = doc.blocking_read();
+            let heads = replica.heads().to_vec();
+            if heads == before {
+                return (heads, Vec::new(), Vec::new());
+            }
+
+            // After an epoch switch the previous heads belong to the old document:
+            // report the root as changed, which every stream is related to.
+            if !changes::missing(&replica.doc, &before).is_empty() {
+                return (heads, vec![Vec::new()], Vec::new());
+            }
+
+            let mut paths = Vec::new();
+            let mut conflicts = Vec::new();
+            for changed in replica.changed_since(&before, InLists::Index) {
+                if changed.conflict {
+                    conflicts.push(changed.path.clone());
+                }
+                paths.push(changed.path);
+            }
+            (heads, paths, conflicts)
+        })
+        .await
+        .expect("change publisher task panicked");
+
+        if heads == last_heads {
+            continue;
+        }
+        last_heads = heads;
+
+        for path in &conflicts {
+            warn!(path = ?path, "Concurrent writes produced a conflict");
+        }
+
+        if paths
+            .iter()
+            .any(|path| crate::swarm::topology::affects_neighbours(path))
+        {
+            state.topology_changed.notify_one();
+        }
+        if paths
+            .iter()
+            .any(|path| crate::swarm::peer::affects_forget_records(path))
+        {
+            state
+                .forget_records_changed
+                .store(true, std::sync::atomic::Ordering::Release);
+            state.peers_due.now();
+        }
+
+        // Fails only when nobody is subscribed.
+        let _ = state.change_events.send(Arc::new(ChangeSet { paths }));
+    }
+}