add member registry

1 files changed, 67 insertions, 0 deletions

diff --git a/src/swarm/members.rs b/src/swarm/members.rs
new file mode 100644
index 0000000..45c8aeb
--- /dev/null
+++ b/src/swarm/members.rs
@@ -0,0 +1,67 @@
+//! The member registry: every node that has joined the swarm, under
+//! `_continuum/members/<pk>`, replicated with the document (and carried into new
+//! epochs with the rest of `_continuum`).
+//!
+//! A pure registry: it decides nothing about admission (that's the swarm key). Each
+//! node registers itself once it is ready, and re-registers if its entry disappears
+//! while it is alive. An operator's forget (`DELETE /peers/{pk}`) removes the entry
+//! together with writing the forget record; a forgotten node that turns out to be alive
+//! cancels the record and registers again.
+
+use automerge::AutoCommit;
+use serde_json::json;
+use tracing::{info, warn};
+
+use crate::document::store::{self, StoreError};
+use crate::state::{AppState, unix_millis};
+
+const META_ROOT: &str = "_continuum";
+
+fn members_dir() -> Vec<String> {
+    vec![META_ROOT.to_owned(), "members".to_owned()]
+}
+
+pub(crate) fn member_path(pk: &str) -> Vec<String> {
+    let mut path = members_dir();
+    path.push(pk.to_owned());
+    path
+}
+
+/// Public keys of all registered members, sorted.
+pub(crate) fn members(doc: &AutoCommit) -> Vec<String> {
+    let mut members = store::keys_at(doc, &members_dir()).unwrap_or_default();
+    members.sort();
+    members
+}
+
+/// Registers this node if it is ready and its entry is missing.
+pub(crate) async fn ensure_registered(state: &AppState) {
+    if !state.is_ready().await {
+        return;
+    }
+    let own = member_path(state.local_id());
+    let registered = store::read_path(&state.doc.read().await.doc, &own)
+        .ok()
+        .flatten()
+        .is_some();
+    if registered {
+        return;
+    }
+
+    let entry = json!({ "since": unix_millis() });
+    match state
+        .mutate_doc(move |doc| store::write_path(doc, &own, &entry))
+        .await
+    {
+        Ok(()) => info!("Registered in the swarm's member list"),
+        Err(error) => warn!(error = %error, "Failed to register in the member list"),
+    }
+}
+
+/// Removes `pk`'s entry (an operator forgot it). Absent is fine.
+pub(crate) fn unregister(doc: &mut AutoCommit, pk: &str) -> Result<(), StoreError> {
+    match store::delete_path(doc, &member_path(pk), false) {
+        Err(StoreError::NotFound) => Ok(()),
+        result => result,
+    }
+}