feat(hermes): Add subscription to .new topic (#706)

* subscribe topic new

Signed-off-by: bkioshn <bkioshn@gmail.com>

* remove wrong doc

Signed-off-by: bkioshn <bkioshn@gmail.com>

* remove .syn subscription

Signed-off-by: bkioshn <bkioshn@gmail.com>

* fix ipfs subscribe test

Signed-off-by: bkioshn <bkioshn@gmail.com>

* use default channel name

Signed-off-by: bkioshn <bkioshn@gmail.com>

---------

Signed-off-by: bkioshn <bkioshn@gmail.com>

Author: bkioshn

hermes/apps/athena/modules/doc-sync/src/lib.rs

@@ -35,11 +35,16 @@ use shared::{
 /// Doc Sync component - thin wrapper calling host-side implementation.
 struct Component;
 
+/// Default channel name for doc-sync operations
+const DOC_SYNC_CHANNEL: &str = "documents";
+
 impl exports::hermes::init::event::Guest for Component {
     /// Initialize the module.
     fn init() -> bool {
         log::init(log::LevelFilter::Trace);
         info!(target: "doc_sync::init", "Doc sync module initialized");
+        // Create the channel during initialization
+        SyncChannel::new(DOC_SYNC_CHANNEL);
         true
     }
 }
@@ -125,9 +130,6 @@ fn json_response(
     })
 }
 
-/// Default channel name for doc-sync operations
-const DOC_SYNC_CHANNEL: &str = "documents";
-
 /// API for posting documents to IPFS `PubSub` channels.
 pub mod channel {
     use super::{DOC_SYNC_CHANNEL, DocData, SyncChannel, hermes};

hermes/bin/src/runtime_extensions/hermes/doc_sync/host.rs

@@ -5,7 +5,6 @@ use stringzilla::stringzilla::Sha256;
 use wasmtime::component::Resource;
 
 use crate::{
-    ipfs::hermes_ipfs_subscribe,
     runtime_context::HermesRuntimeContext,
     runtime_extensions::{
         bindings::hermes::{
@@ -32,9 +31,6 @@ const SHA2_256_CODE: u64 = 0x12;
 /// See: <https://github.com/ipld/cid-cbor/>
 const CID_CBOR_TAG: u64 = 42;
 
-/// Default `PubSub` topic for doc-sync channel
-const DOC_SYNC_TOPIC: &str = "doc-sync/documents";
-
 /// Wrapper for `hermes_ipfs::Cid` to implement `minicbor::Encode` for it.
 struct Cid(hermes_ipfs::Cid);
 
@@ -90,11 +86,6 @@ impl HostSyncChannel for HermesRuntimeContext {
     ///
     /// - `name`: The Name of the channel to Open.  Creates if it doesn't exist, otherwise
     ///   joins it.
-    ///
-    /// **Returns**
-    ///
-    /// - `ok(network)`: A resource network, if successfully create network resource.
-    /// - `error(create-network-error)`: If creating network resource failed.
     fn new(
         &mut self,
         name: ChannelName,
@@ -128,11 +119,17 @@ impl HostSyncChannel for HermesRuntimeContext {
             }
         }
 
-        if let Err(err) = hermes_ipfs_subscribe(self.app_name(), name) {
+        // When the channel is created, subscribe to .new <base>.<topic>
+        if let Err(err) = self.pubsub_subscribe(format!("{name}.new")) {
+            // FIXME - Do we want to remove the entry from the map here?
             DOC_SYNC_STATE.remove(&resource);
-            return Err(wasmtime::Error::msg(format!("Subscription failed: {err}",)));
+            return Err(wasmtime::Error::msg(format!(
+                "Subscription to {name}.new failed: {err}",
+            )));
         }
 
+        tracing::info!("Created Doc Sync Channel: {name}");
+
         Ok(wasmtime::component::Resource::new_own(resource))
     }
 
@@ -168,7 +165,7 @@ impl HostSyncChannel for HermesRuntimeContext {
     /// 3. Publish to `PubSub` (`pubsub_publish`)
     fn post(
         &mut self,
-        _self_: Resource<SyncChannel>,
+        self_: Resource<SyncChannel>,
         doc: DocData,
     ) -> wasmtime::Result<Result<DocLoc, Errno>> {
         tracing::info!("📤 Posting {} bytes to doc-sync channel", doc.len());
@@ -202,18 +199,27 @@ impl HostSyncChannel for HermesRuntimeContext {
         //
         // Since Step 1 (add + pin) already succeeded, the document is safely stored
         // in IPFS. We treat "no peers" as a warning rather than a fatal error.
-        let topic = DOC_SYNC_TOPIC.to_string();
 
-        // Subscribe to the topic first (required for Gossipsub - you must be subscribed
-        // to a topic before you can publish to it)
-        match self.pubsub_subscribe(topic.clone())? {
-            Ok(_) => tracing::info!("✓ Subscribed to topic: {}", topic),
+        let channel_name = DOC_SYNC_STATE
+            .get(&self_.rep())
+            .ok_or_else(|| wasmtime::Error::msg("Channel not found"))?
+            .value()
+            .clone();
+
+        let topic_new = format!("{channel_name}.new");
+
+        // The channel should already be subscribed to the `.new` topic (subscription
+        // is performed in `new()`). Invoking the subscription again to ensure
+        // the topic is active, because Gossipsub enforces that peers must subscribe
+        // to a topic before they are permitted to publish on it.
+        match self.pubsub_subscribe(topic_new.clone())? {
+            Ok(_) => tracing::info!("✓ Subscribed to topic: {topic_new}"),
             Err(e) => tracing::warn!("⚠ Subscribe warning: {:?}", e),
         }
 
         // Attempt to publish to PubSub
-        if let Ok(()) = self.pubsub_publish(topic.clone(), doc)? {
-            tracing::info!("✓ Step 3/3: Published to PubSub → {}", topic);
+        if let Ok(()) = self.pubsub_publish(topic_new.clone(), doc)? {
+            tracing::info!("✓ Step 3/3: Published to PubSub → {topic_new}",);
         } else {
             // Non-fatal: PubSub requires peer nodes to be subscribed to the topic.
             // In a single-node environment, this is expected to fail with
@@ -223,8 +229,7 @@ impl HostSyncChannel for HermesRuntimeContext {
                 "⚠ Step 3/3: PubSub publish skipped (no peer nodes subscribed to topic)"
             );
             tracing::warn!(
-                "   Note: Gossipsub requires other nodes subscribing to '{}' to work",
-                topic
+                "   Note: Gossipsub requires other nodes subscribing to '{topic_new}' to work",
             );
             tracing::info!("   Document is successfully stored in IPFS from Step 1");
         }

hermes/bin/tests/integration/tests/serial/ipfs_subscribe.rs

@@ -27,7 +27,7 @@ fn ipfs_subscribe() {
 
     assert!(utils::assert::app_logs_contain(
         &temp_dir,
-        "\"pubsub_topic\":\"ipfs_channel\""
+        "\"pubsub_topic\":\"ipfs_channel.new\""
     ));
 
     // Uncomment the line below if you want to inspect the details

wasm/wasi/wit/deps/hermes-doc-sync/api.wit

@@ -34,11 +34,6 @@ interface api {
         /// **Parameters**
         ///
         /// - `name`: The Name of the channel to Open.  Creates if it doesn't exist, otherwise joins it.
-        ///
-        /// **Returns**
-        ///
-        /// - `ok(network)`: A resource network, if successfully create network resource.
-        /// - `error(create-network-error)`: If creating network resource failed.
         constructor(name: channel-name);
 
         /// Close Doc Sync Channel