pool: add RelayPoolBuilder

- Add `RelayPool::builder` constructor
- Change `RelayPool::new` constructor signature
- Remove `set_admit_policy` method from `Relay` and `RelayPool`
- Remove `RelayPool::__with_shared_state`

Signed-off-by: Yuki Kishimoto <[email protected]>

Author: yukibtc

CHANGELOG.md

@@ -49,6 +49,7 @@
 * nostr: change `EventBuilder::delete` arguments ([Yuki Kishimoto])
 * pool: drop `RelayFiltering` ([Yuki Kishimoto])
 * pool: remove `Relay` constructors ([Yuki Kishimoto])
+* pool: change `RelayPool::new` signature ([Yuki Kishimoto])
 * sdk: change `Client::fetch_metadata` output ([Yuki Kishimoto]) 
 
 ### Changed
@@ -98,6 +99,7 @@
 * pool: event verification cache ([Yuki Kishimoto])
 * pool: add `AdmitPolicy` trait ([Yuki Kishimoto])
 * pool: add `ReqExitPolicy::WaitForEvents` variant ([Yuki Kishimoto])
+* pool: add `RelayPoolBuilder` ([Yuki Kishimoto])
 * ffi: add Mac Catalyst support in Swift package ([Yuki Kishimoto])
 * js: add `KindStandard` enum ([Yuki Kishimoto])
 

crates/nostr-relay-pool/src/pool/builder.rs

@@ -0,0 +1,84 @@
+// Copyright (c) 2022-2023 Yuki Kishimoto
+// Copyright (c) 2023-2025 Rust Nostr Developers
+// Distributed under the MIT software license
+
+//! Relay Pool builder
+
+use std::sync::Arc;
+
+use nostr::NostrSigner;
+use nostr_database::{MemoryDatabase, NostrDatabase};
+
+use super::options::RelayPoolOptions;
+use super::RelayPool;
+use crate::policy::AdmitPolicy;
+use crate::transport::websocket::{DefaultWebsocketTransport, WebSocketTransport};
+
+/// Relay Pool builder
+#[derive(Debug, Clone)]
+pub struct RelayPoolBuilder {
+    /// WebSocket transport
+    pub websocket_transport: Arc<dyn WebSocketTransport>,
+    /// Admission policy
+    pub admit_policy: Option<Arc<dyn AdmitPolicy>>,
+    /// Relay pool options
+    pub opts: RelayPoolOptions,
+    // Private stuff
+    #[doc(hidden)]
+    pub __database: Arc<dyn NostrDatabase>,
+    #[doc(hidden)]
+    pub __signer: Option<Arc<dyn NostrSigner>>,
+}
+
+impl Default for RelayPoolBuilder {
+    fn default() -> Self {
+        Self {
+            websocket_transport: Arc::new(DefaultWebsocketTransport),
+            admit_policy: None,
+            opts: RelayPoolOptions::default(),
+            __database: Arc::new(MemoryDatabase::default()),
+            __signer: None,
+        }
+    }
+}
+
+impl RelayPoolBuilder {
+    /// New default builder
+    #[inline]
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Set a WebSocket transport
+    #[inline]
+    pub fn websocket_transport<T>(mut self, transport: T) -> Self
+    where
+        T: WebSocketTransport + 'static,
+    {
+        self.websocket_transport = Arc::new(transport);
+        self
+    }
+
+    /// Admission policy
+    #[inline]
+    pub fn admit_policy<T>(mut self, policy: T) -> Self
+    where
+        T: AdmitPolicy + 'static,
+    {
+        self.admit_policy = Some(Arc::new(policy));
+        self
+    }
+
+    /// Set options
+    #[inline]
+    pub fn opts(mut self, opts: RelayPoolOptions) -> Self {
+        self.opts = opts;
+        self
+    }
+
+    /// Build relay pool
+    #[inline]
+    pub fn build(self) -> RelayPool {
+        RelayPool::from_builder(self)
+    }
+}

crates/nostr-relay-pool/src/pool/inner.rs

@@ -14,7 +14,7 @@ use nostr_database::prelude::*;
 use tokio::sync::{broadcast, RwLock};
 
 use super::options::RelayPoolOptions;
-use super::{Error, RelayPoolNotification};
+use super::{Error, RelayPoolBuilder, RelayPoolNotification};
 use crate::relay::flags::RelayServiceFlags;
 use crate::relay::options::RelayOptions;
 use crate::relay::Relay;
@@ -47,18 +47,24 @@ impl AtomicDestroyer for InnerRelayPool {
 }
 
 impl InnerRelayPool {
-    pub fn new(opts: RelayPoolOptions, state: SharedState) -> Self {
-        let (notification_sender, _) = broadcast::channel(opts.notification_channel_size);
+    pub(super) fn from_builder(builder: RelayPoolBuilder) -> Self {
+        let (notification_sender, _) = broadcast::channel(builder.opts.notification_channel_size);
 
         Self {
-            state,
+            state: SharedState::new(
+                builder.__database,
+                builder.websocket_transport,
+                builder.__signer,
+                builder.admit_policy,
+                builder.opts.nip42_auto_authentication,
+            ),
             atomic: Arc::new(AtomicPrivateData {
                 relays: RwLock::new(HashMap::new()),
                 subscriptions: RwLock::new(HashMap::new()),
                 shutdown: AtomicBool::new(false),
             }),
             notification_sender,
-            opts,
+            opts: builder.opts,
         }
     }
 

crates/nostr-relay-pool/src/pool/mod.rs

@@ -15,17 +15,18 @@ use atomic_destructor::{AtomicDestructor, StealthClone};
 use nostr_database::prelude::*;
 use tokio::sync::{broadcast, mpsc, RwLockReadGuard};
 
+pub mod builder;
 pub mod constants;
 mod error;
 mod inner;
 pub mod options;
 mod output;
 
+pub use self::builder::RelayPoolBuilder;
 pub use self::error::Error;
 use self::inner::{InnerRelayPool, Relays};
 pub use self::options::RelayPoolOptions;
 pub use self::output::Output;
-use crate::policy::AdmitPolicy;
 use crate::relay::flags::FlagCheck;
 use crate::relay::options::{RelayOptions, ReqExitPolicy, SyncOptions};
 use crate::relay::Relay;
@@ -79,7 +80,7 @@ pub struct RelayPool {
 
 impl Default for RelayPool {
     fn default() -> Self {
-        Self::new(RelayPoolOptions::default())
+        Self::builder().build()
     }
 }
 
@@ -92,28 +93,25 @@ impl StealthClone for RelayPool {
 }
 
 impl RelayPool {
-    /// Create new `RelayPool`
+    /// Construct new default relay pool
+    ///
+    /// Use [`RelayPool::builder`] to customize it.
     #[inline]
-    pub fn new(opts: RelayPoolOptions) -> Self {
-        Self::__with_shared_state(opts, SharedState::default())
+    pub fn new() -> Self {
+        Self::default()
     }
 
+    /// New relay pool builder
     #[inline]
-    #[doc(hidden)]
-    pub fn __with_shared_state(opts: RelayPoolOptions, state: SharedState) -> Self {
-        Self {
-            inner: AtomicDestructor::new(InnerRelayPool::new(opts, state)),
-        }
+    pub fn builder() -> RelayPoolBuilder {
+        RelayPoolBuilder::default()
     }
 
-    /// Set an admission policy
     #[inline]
-    pub fn set_admit_policy<T>(&self, policy: T) -> Result<(), Error>
-    where
-        T: AdmitPolicy + 'static,
-    {
-        self.inner.state.set_admit_policy(policy)?;
-        Ok(())
+    fn from_builder(builder: RelayPoolBuilder) -> Self {
+        Self {
+            inner: AtomicDestructor::new(InnerRelayPool::from_builder(builder)),
+        }
     }
 
     /// Completely shutdown pool

crates/nostr-relay-pool/src/pool/options.rs

@@ -10,13 +10,15 @@ use super::constants::DEFAULT_NOTIFICATION_CHANNEL_SIZE;
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct RelayPoolOptions {
     pub(super) max_relays: Option<usize>,
+    pub(super) nip42_auto_authentication: bool,
     pub(super) notification_channel_size: usize,
 }
 
 impl Default for RelayPoolOptions {
     fn default() -> Self {
         Self {
             max_relays: None,
+            nip42_auto_authentication: true,
             notification_channel_size: DEFAULT_NOTIFICATION_CHANNEL_SIZE,
         }
     }
@@ -36,6 +38,15 @@ impl RelayPoolOptions {
         self
     }
 
+    /// Auto authenticate to relays (default: true)
+    ///
+    /// <https://github.com/nostr-protocol/nips/blob/master/42.md>
+    #[inline]
+    pub fn automatic_authentication(mut self, enabled: bool) -> Self {
+        self.nip42_auto_authentication = enabled;
+        self
+    }
+
     /// Notification channel size (default: [`DEFAULT_NOTIFICATION_CHANNEL_SIZE`])
     #[inline]
     pub fn notification_channel_size(mut self, size: usize) -> Self {

crates/nostr-relay-pool/src/prelude.rs

@@ -15,6 +15,7 @@ pub use nostr_database::*;
 
 // Internal modules
 pub use crate::policy::*;
+pub use crate::pool::builder::*;
 pub use crate::pool::constants::*;
 pub use crate::pool::options::*;
 pub use crate::pool::{self, *};

crates/nostr-relay-pool/src/relay/inner.rs

@@ -999,7 +999,7 @@ impl InnerRelay {
         }
 
         // Check event admission policy
-        if let Some(policy) = self.state.admit_policy.get() {
+        if let Some(policy) = &self.state.admit_policy {
             if let AdmitStatus::Rejected = policy
                 .admit_event(&self.url, &subscription_id, &event)
                 .await?

crates/nostr-relay-pool/src/relay/mod.rs

@@ -37,7 +37,6 @@ pub use self::options::{
 };
 pub use self::stats::RelayConnectionStats;
 pub use self::status::RelayStatus;
-use crate::policy::AdmitPolicy;
 use crate::shared::SharedState;
 use crate::transport::websocket::{BoxSink, BoxStream};
 
@@ -157,16 +156,6 @@ impl Relay {
         }
     }
 
-    /// Set an admission policy
-    #[inline]
-    pub fn set_admit_policy<T>(&self, policy: T) -> Result<(), Error>
-    where
-        T: AdmitPolicy + 'static,
-    {
-        self.inner.state.set_admit_policy(policy)?;
-        Ok(())
-    }
-
     /// Get relay url
     #[inline]
     pub fn url(&self) -> &RelayUrl {

crates/nostr-relay-pool/src/shared.rs

@@ -7,7 +7,7 @@ use std::fmt;
 use std::hash::{Hash, Hasher};
 use std::num::NonZeroUsize;
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::{Arc, Mutex, OnceLock};
+use std::sync::{Arc, Mutex};
 
 use lru::LruCache;
 use nostr::prelude::IntoNostrSigner;
@@ -16,9 +16,7 @@ use nostr_database::{IntoNostrDatabase, MemoryDatabase, NostrDatabase};
 use tokio::sync::RwLock;
 
 use crate::policy::AdmitPolicy;
-use crate::transport::websocket::{
-    DefaultWebsocketTransport, IntoWebSocketTransport, WebSocketTransport,
-};
+use crate::transport::websocket::{DefaultWebsocketTransport, WebSocketTransport};
 
 // LruCache pre-allocate, so keep this at a reasonable value.
 // A good value may be <= 128k, considering that stored values are the 64-bit hashes of the event IDs.
@@ -27,8 +25,6 @@ const MAX_VERIFICATION_CACHE_SIZE: usize = 128_000;
 #[derive(Debug)]
 pub enum SharedStateError {
     SignerNotConfigured,
-    /// Admit policy already set
-    AdmitPolicyAlreadySet,
     MutexPoisoned,
 }
 
@@ -38,29 +34,26 @@ impl fmt::Display for SharedStateError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             Self::SignerNotConfigured => write!(f, "signer not configured"),
-            Self::AdmitPolicyAlreadySet => write!(f, "admission policy already set"),
             Self::MutexPoisoned => write!(f, "mutex poisoned"),
         }
     }
 }
 
-// TODO: reduce atomic operations
 #[derive(Debug, Clone)]
 pub struct SharedState {
     pub(crate) database: Arc<dyn NostrDatabase>,
     pub(crate) transport: Arc<dyn WebSocketTransport>,
     signer: Arc<RwLock<Option<Arc<dyn NostrSigner>>>>,
     nip42_auto_authentication: Arc<AtomicBool>,
     verification_cache: Arc<Mutex<LruCache<u64, ()>>>,
-    pub(crate) admit_policy: OnceLock<Arc<dyn AdmitPolicy>>,
-    // TODO: add a semaphore to limit number of concurrent websocket connections attempts?
+    pub(crate) admit_policy: Option<Arc<dyn AdmitPolicy>>,
 }
 
 impl Default for SharedState {
     fn default() -> Self {
         Self::new(
             MemoryDatabase::new().into_nostr_database(),
-            DefaultWebsocketTransport.into_transport(),
+            Arc::new(DefaultWebsocketTransport),
             None,
             None,
             true,
@@ -86,33 +79,10 @@ impl SharedState {
             signer: Arc::new(RwLock::new(signer)),
             nip42_auto_authentication: Arc::new(AtomicBool::new(nip42_auto_authentication)),
             verification_cache: Arc::new(Mutex::new(LruCache::new(max_verification_cache_size))),
-            admit_policy: match admit_policy {
-                Some(policy) => OnceLock::from(policy),
-                None => OnceLock::new(),
-            },
+            admit_policy,
         }
     }
 
-    /// Set a custom transport
-    pub fn custom_transport<T>(mut self, transport: T) -> Self
-    where
-        T: IntoWebSocketTransport,
-    {
-        self.transport = transport.into_transport();
-        self
-    }
-
-    /// Set an admission policy
-    #[inline]
-    pub(crate) fn set_admit_policy<T>(&self, policy: T) -> Result<(), SharedStateError>
-    where
-        T: AdmitPolicy + 'static,
-    {
-        self.admit_policy
-            .set(Arc::new(policy))
-            .map_err(|_| SharedStateError::AdmitPolicyAlreadySet)
-    }
-
     /// Check if auto authentication to relays is enabled
     #[inline]
     pub fn is_auto_authentication_enabled(&self) -> bool {