feat(hermes): Add handler for the SYN payload

hermes/bin/src/ipfs/doc_sync/reconciliation.rs

@@ -36,6 +36,7 @@ pub(crate) enum DocReconciliation {
 /// The context for document reconciliation process, if it is needed.
 pub(crate) struct DocReconciliationData {
     /// Root of our SMT.
+    #[allow(dead_code)]
     our_root: Blake3256,
     /// Document count in our SMT.
     our_count: u64,
@@ -60,7 +61,7 @@ pub(super) async fn start_reconciliation(
     tracing::info!(%channel, "subscribed to .dif");
 
     let syn_payload = make_syn_payload(doc_reconciliation_data, peer).await;
-    tracing::info!("SYN payload created");
+    tracing::info!(root = %syn_payload.root.to_hex(), count = %syn_payload.count, "SYN payload created");
 
     if let Err(err) = send_syn_payload(&syn_payload, app_name, channel).await {
         unsubscribe_from_dif(app_name, channel).await?;

hermes/bin/src/ipfs/doc_sync/topic_handler.rs

@@ -2,7 +2,10 @@
 
 use std::sync::Arc;
 
-use hermes_ipfs::doc_sync::payload::{self, CommonFields, DocumentDisseminationBody};
+use hermes_ipfs::doc_sync::{
+    payload::{self, CommonFields, DocumentDisseminationBody},
+    syn_payload::{self, MsgSyn},
+};
 
 use crate::ipfs::{
     doc_sync::reconciliation, task::process_broadcasted_cids,
@@ -132,6 +135,27 @@ impl TopicHandler for payload::New {
     }
 }
 
+impl TopicHandler for syn_payload::MsgSyn {
+    const TOPIC_SUFFIX: &'static str = ".syn";
+
+    async fn handle(
+        self,
+        _topic: &str,
+        _source: Option<hermes_ipfs::PeerId>,
+        context: TopicMessageContext,
+    ) -> anyhow::Result<()> {
+        let Some(_tree) = context.tree() else {
+            return Err(anyhow::anyhow!(
+                "Context for payload::New handler must contain an SMT."
+            ));
+        };
+
+        let msg: MsgSyn = MsgSyn::from(self);
+        tracing::info!(root = %msg.root.to_hex(), count = %msg.count, "Received SYN message");
+        Ok(())
+    }
+}
+
 /// Handles the IPFS messages of a specific topic.
 pub(crate) async fn handle_doc_sync_topic<'a, TH: TopicHandler + minicbor::Decode<'a, ()>>(
     message: &'a hermes_ipfs::rust_ipfs::GossipsubMessage,

hermes/bin/src/ipfs/topic_handlers.rs

@@ -4,7 +4,10 @@
 
 use std::{pin::Pin, sync::Arc};
 
-use hermes_ipfs::doc_sync::payload::{self};
+use hermes_ipfs::doc_sync::{
+    payload::{self},
+    syn_payload,
+};
 
 use super::HERMES_IPFS;
 use crate::{
@@ -131,6 +134,16 @@ pub(super) async fn topic_message_handler(
     }
 }
 
+async fn try_handlers(
+    message: &hermes_ipfs::rust_ipfs::GossipsubMessage,
+    topic: &str,
+    context: &TopicMessageContext,
+) -> Option<anyhow::Result<()>> {
+    handle_doc_sync_topic::<payload::New>(message, topic, context.clone())
+        .await
+        .or(handle_doc_sync_topic::<syn_payload::MsgSyn>(message, topic, context.clone()).await)
+}
+
 /// Handler for Doc Sync `PubSub` messages.
 #[allow(
     clippy::needless_pass_by_value,
@@ -148,7 +161,7 @@ pub(super) async fn doc_sync_topic_message_handler(
         );
     }
 
-    let result = handle_doc_sync_topic::<payload::New>(&message, &topic, context).await;
+    let result = try_handlers(&message, &topic, &context).await;
     if let Some(Err(err)) = result {
         tracing::error!("Failed to handle IPFS message: {}", err);
     }