syncweb_core/gossip/
topic_channel.rs

1use std::marker::PhantomData;
2use std::sync::Arc;
3
4use bytes::Bytes;
5use iroh_gossip::{
6    TopicId,
7    api::{ApiError, Event, GossipSender},
8    net::Gossip,
9};
10use n0_future::{Stream, StreamExt};
11
12use crate::error::{Result, SyncwebError};
13use crate::gossip::signed_topic::SignedGossipMessage;
14
15/// A typed gossip topic that can publish and subscribe to one message type.
16///
17/// Uses `Gossip` (syncweb's wrapper around `iroh_gossip::net::Gossip`) for
18/// lifecycle management. `TopicChannel` only handles publish and
19/// per-subscription stream filtering — the underlying gossip connection,
20/// subscription, and peer bootstrap are managed by `GossipService`.
21pub struct TopicChannel<T: SignedGossipMessage + Send + Sync> {
22    gossip: Arc<Gossip>,
23    topic_id: TopicId,
24    sender: GossipSender,
25    _phantom: PhantomData<T>,
26}
27
28impl<T: SignedGossipMessage + Send + Sync> TopicChannel<T> {
29    /// Create a deterministic topic ID from a byte-string topic name, and
30    /// store a sender for publishing.
31    #[must_use]
32    pub fn new(gossip: Arc<Gossip>, topic_name: &[u8], sender: GossipSender) -> Self {
33        Self {
34            gossip,
35            topic_id: TopicId::from_bytes(*blake3::hash(topic_name).as_bytes()),
36            sender,
37            _phantom: PhantomData,
38        }
39    }
40
41    /// Publish a signed message to the topic. All subscribers receive it.
42    ///
43    /// # Errors
44    ///
45    /// Returns an error if signature verification or broadcasting fails.
46    pub async fn publish(&self, message: &T) -> Result<()> {
47        message.verify_signature()?;
48        let wire_bytes = message.to_wire_bytes()?;
49        let bytes = Bytes::from(wire_bytes);
50        self.sender
51            .broadcast(bytes)
52            .await
53            .map_err(|error| SyncwebError::operation("failed to publish gossip message", error))?;
54        Ok(())
55    }
56
57    /// Receive verified messages from an already-subscribed gossip event stream.
58    ///
59    /// The caller must already have subscribed to `self.topic_id` via
60    /// `GossipService`. This method wraps the raw event stream with
61    /// deserialization and signature verification.
62    ///
63    /// Returns a stream of verified, deserialized `T` messages. Messages that
64    /// fail deserialization or signature verification are silently dropped
65    /// (logged at debug level).
66    pub fn receive_from(
67        &self,
68        stream: impl Stream<Item = std::result::Result<Event, ApiError>> + Send + 'static,
69    ) -> impl Stream<Item = T> + Send {
70        stream.filter_map(|event_result| match event_result {
71            Ok(Event::Received(msg)) => {
72                let parsed = match T::from_wire_bytes(&msg.content) {
73                    Ok(m) => m,
74                    Err(error) => {
75                        tracing::debug!("failed to deserialize gossip message: {error}");
76                        return None;
77                    }
78                };
79                if let Err(error) = parsed.verify_signature() {
80                    tracing::debug!("gossip message signature verification failed: {error}");
81                    return None;
82                }
83                Some(parsed)
84            }
85            Ok(_) => None,
86            Err(error) => {
87                tracing::debug!("gossip stream error: {error}");
88                None
89            }
90        })
91    }
92
93    /// Collect messages matching a filter within a timeout.
94    ///
95    /// The caller must already be subscribed to the topic via `GossipService`.
96    /// `stream` is the raw gossip event stream from the subscription.
97    ///
98    /// # Errors
99    ///
100    /// Returns an empty vector on timeout (not an error).
101    pub async fn collect_for(
102        &self,
103        stream: impl Stream<Item = std::result::Result<Event, ApiError>> + Send + 'static,
104        filter: impl Fn(&T) -> bool + Send + Sync + 'static,
105        timeout_duration: std::time::Duration,
106    ) -> Result<Vec<T>> {
107        let mut filtered = Box::pin(self.receive_from(stream));
108        let mut results = Vec::new();
109        let _ = tokio::time::timeout(timeout_duration, async {
110            while let Some(msg) = filtered.next().await {
111                if filter(&msg) {
112                    results.push(msg);
113                }
114            }
115        })
116        .await;
117        Ok(results)
118    }
119
120    #[must_use]
121    pub const fn topic_id(&self) -> TopicId {
122        self.topic_id
123    }
124
125    #[must_use]
126    pub fn gossip(&self) -> &Gossip {
127        &self.gossip
128    }
129}
syncweb_core/gossip/topic_channel.rs

syncweb_core/gossip/
topic_channel.rs
