add epoch catch-up / snapshots

1 files changed, 28 insertions, 2 deletions

diff --git a/src/state.rs b/src/state.rs
index ec7996a..598585d 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -566,13 +566,39 @@ impl Host for AppState {
 
             let mut summaries = Vec::with_capacity(peers.len());
             for (id, peer) in peers {
-                let active = peer.lock().await.liveness == Liveness::Active;
-                summaries.push(PeerSummary { id, active });
+                let peer = peer.lock().await;
+                summaries.push(PeerSummary {
+                    id,
+                    active: peer.liveness == Liveness::Active,
+                    established: peer.hello_state == HelloState::Established,
+                    epoch: peer.epoch,
+                });
             }
             summaries
         }
     }
 
+    fn is_cluster_member(&self, peer: &str) -> impl Future<Output = bool> + Send {
+        let state = self.clone();
+        let peer = peer.to_owned();
+        async move {
+            let Some(ours) = state.cluster_id().await else {
+                return false;
+            };
+            crate::peer::peer_cluster(&state, &peer).await.as_deref() == Some(ours.as_str())
+        }
+    }
+
+    fn send_message(
+        &self,
+        peer: &str,
+        message: transport::ParsedMessage,
+    ) -> impl Future<Output = anyhow::Result<()>> + Send {
+        let state = self.clone();
+        let peer = peer.to_owned();
+        async move { state.transport.send_message(&peer, &message).await }
+    }
+
     fn subscribe_changes(&self) -> broadcast::Receiver<Arc<ChangeSet>> {
         self.change_events.subscribe()
     }