syncweb_core/node/
discovery.rs

1use distributed_topic_tracker::{
2    AutoDiscoveryGossip, Config, DefaultSecretRotation, RecordPublisher, RotationHandle, Topic, TopicId,
3};
4use ed25519_dalek::SigningKey;
5use iroh::{Endpoint, PublicKey};
6use iroh_docs::NamespaceId;
7use iroh_gossip::net::Gossip;
8use std::{collections::HashMap, sync::Arc};
9use tokio::sync::Mutex;
10
11use crate::error::{Result, SyncwebError};
12
13#[derive(Clone)]
14pub struct TopicTracker {
15    gossip: Gossip,
16    signing_key: SigningKey,
17    topics: Arc<Mutex<HashMap<[u8; 32], Topic>>>,
18}
19
20impl TopicTracker {
21    #[must_use]
22    pub fn new(gossip: &Gossip, endpoint: &Endpoint) -> Self {
23        Self {
24            gossip: gossip.clone(),
25            signing_key: SigningKey::from_bytes(&endpoint.secret_key().to_bytes()),
26            topics: Arc::new(Mutex::new(HashMap::new())),
27        }
28    }
29
30    /// # Errors
31    ///
32    /// Returns an error if joining the topic fails.
33    pub async fn announce(&self, namespace_id: NamespaceId) -> Result<()> {
34        self.topic(namespace_id).await?;
35        Ok(())
36    }
37
38    /// # Errors
39    ///
40    /// Returns an error if the topic cannot be joined or gossip neighbors cannot be retrieved.
41    pub async fn find_peers(&self, namespace_id: NamespaceId) -> Result<Vec<PublicKey>> {
42        let topic = self.topic(namespace_id).await?;
43        let receiver = topic
44            .gossip_receiver()
45            .await
46            .map_err(|error| SyncwebError::operation("failed to get discovery receiver", error))?;
47        Ok(receiver
48            .neighbors()
49            .await
50            .map_err(|error| SyncwebError::operation("failed to find discovery peers", error))?
51            .into_iter()
52            .collect())
53    }
54
55    async fn topic(&self, namespace_id: NamespaceId) -> Result<Topic> {
56        let key = *namespace_id.as_bytes();
57        let existing = self.topics.lock().await.get(&key).cloned();
58        if let Some(topic) = existing {
59            return Ok(topic);
60        }
61
62        let topic_id = TopicId::from_hash(&key);
63        let publisher = RecordPublisher::builder(topic_id, self.signing_key.clone(), namespace_id.as_bytes().to_vec())
64            .config(Config::default())
65            .secret_rotation(RotationHandle::new(DefaultSecretRotation))
66            .build();
67        let topic = self
68            .gossip
69            .subscribe_and_join_with_auto_discovery_no_wait(publisher)
70            .await
71            .map_err(|error| SyncwebError::operation("failed to join discovery topic", error))?;
72        self.topics.lock().await.insert(key, topic.clone());
73        Ok(topic)
74    }
75}
syncweb_core/node/discovery.rs

syncweb_core/node/
discovery.rs
