1use anyhow::{Context, Result};
2use axum::Router;
3use hashtree_core::Cid;
4use nostr::nips::nip19::ToBech32;
5use nostr::Keys;
6use std::collections::HashSet;
7use std::path::PathBuf;
8use std::sync::Arc;
9use tokio::net::TcpListener;
10use tokio::sync::{Mutex, Notify};
11use tokio::task::JoinHandle;
12use tower_http::cors::CorsLayer;
13
14use crate::config::{ensure_keys, ensure_keys_in, parse_npub, pubkey_bytes, Config};
15use crate::eviction::{spawn_background_eviction_task, BACKGROUND_EVICTION_INTERVAL};
16use crate::nostr_relay::{NostrRelay, NostrRelayConfig};
17use crate::server::{AppState, HashtreeServer};
18use crate::socialgraph;
19use crate::storage::HashtreeStore;
20
21#[cfg(feature = "p2p")]
22use crate::webrtc::{ContentStore, PeerClassifier, WebRTCManager, WebRTCState};
23#[cfg(not(feature = "p2p"))]
24use crate::WebRTCState;
25
26#[cfg(feature = "p2p")]
27struct PeerRouterRuntime {
28 shutdown: Arc<tokio::sync::watch::Sender<bool>>,
29 join: JoinHandle<()>,
30 peer_state_persist: JoinHandle<()>,
31}
32
33struct BackgroundSyncRuntime {
34 service: Arc<crate::sync::BackgroundSync>,
35 join: Option<JoinHandle<()>>,
36}
37
38impl Drop for BackgroundSyncRuntime {
39 fn drop(&mut self) {
40 self.service.shutdown();
41 if let Some(join) = self.join.take() {
42 join.abort();
43 }
44 }
45}
46
47struct BackgroundMirrorRuntime {
48 service: Arc<crate::nostr_mirror::BackgroundNostrMirror>,
49 join: Option<JoinHandle<()>>,
50}
51
52impl Drop for BackgroundMirrorRuntime {
53 fn drop(&mut self) {
54 self.service.shutdown();
55 if let Some(join) = self.join.take() {
56 join.abort();
57 }
58 }
59}
60
61struct BackgroundServicesRuntime {
62 crawler: Option<socialgraph::crawler::SocialGraphTaskHandles>,
63 mirror: Option<BackgroundMirrorRuntime>,
64 sync: Option<BackgroundSyncRuntime>,
65}
66
67impl Drop for BackgroundServicesRuntime {
68 fn drop(&mut self) {
69 if let Some(handles) = self.crawler.as_ref() {
70 let _ = handles.shutdown_tx.send(true);
71 }
72 if let Some(runtime) = self.mirror.as_ref() {
73 runtime.service.shutdown();
74 }
75 if let Some(runtime) = self.sync.as_ref() {
76 runtime.service.shutdown();
77 }
78 }
79}
80
81impl BackgroundServicesRuntime {
82 fn status(&self) -> EmbeddedBackgroundServicesStatus {
83 EmbeddedBackgroundServicesStatus {
84 crawler_active: self.crawler.is_some(),
85 mirror_active: self.mirror.is_some(),
86 sync_active: self.sync.is_some(),
87 }
88 }
89}
90
91struct EmbeddedServerRuntime {
92 shutdown: Arc<Notify>,
93 join: Option<JoinHandle<()>>,
94}
95
96pub struct EmbeddedServerController {
97 runtime: Mutex<Option<EmbeddedServerRuntime>>,
98}
99
100impl EmbeddedServerController {
101 pub fn new(shutdown: Arc<Notify>, join: JoinHandle<()>) -> Self {
102 Self {
103 runtime: Mutex::new(Some(EmbeddedServerRuntime {
104 shutdown,
105 join: Some(join),
106 })),
107 }
108 }
109
110 pub async fn shutdown(&self) {
111 let mut runtime = self.runtime.lock().await;
112 let Some(mut runtime) = runtime.take() else {
113 return;
114 };
115
116 runtime.shutdown.notify_waiters();
117 if let Some(mut join) = runtime.join.take() {
118 match tokio::time::timeout(std::time::Duration::from_secs(3), &mut join).await {
119 Ok(Ok(())) => {}
120 Ok(Err(err)) => {
121 tracing::warn!("Embedded server task ended with join error: {}", err)
122 }
123 Err(_) => {
124 tracing::warn!("Timed out waiting for embedded server shutdown");
125 join.abort();
126 }
127 }
128 }
129 }
130}
131
132#[derive(Debug, Clone, Copy, PartialEq, Eq)]
133pub struct EmbeddedBackgroundServicesStatus {
134 pub crawler_active: bool,
135 pub mirror_active: bool,
136 pub sync_active: bool,
137}
138
139pub struct EmbeddedBackgroundServicesController {
140 keys: Keys,
141 data_dir: PathBuf,
142 store: Arc<HashtreeStore>,
143 graph_store_concrete: Arc<socialgraph::SocialGraphStore>,
144 graph_store: Arc<dyn socialgraph::SocialGraphBackend>,
145 spambox: Option<Arc<dyn socialgraph::SocialGraphBackend>>,
146 webrtc_state: Option<Arc<WebRTCState>>,
147 runtime: Mutex<BackgroundServicesRuntime>,
148}
149
150impl EmbeddedBackgroundServicesController {
151 const MIRROR_PUBLISH_RELAY_PRIORITY: &[&str] = &[
152 "wss://nos.lol",
153 "wss://temp.iris.to",
154 "wss://vault.iris.to",
155 "wss://relay.damus.io",
156 ];
157 const MIRROR_PUBLISH_RELAY_BLOCKLIST: &[&str] =
158 &["wss://graph-relay.iris.to", "wss://upload.iris.to/nostr"];
159
160 fn mirror_publish_relays(active_relays: &[String], _bind_address: &str) -> Vec<String> {
161 let mut seen = HashSet::new();
162 let active_relays = active_relays
163 .iter()
164 .filter(|relay| seen.insert((*relay).clone()))
165 .cloned()
166 .collect::<Vec<_>>();
167 if active_relays.is_empty() {
168 return Vec::new();
169 }
170 let filtered = active_relays
171 .iter()
172 .filter(|relay| !Self::MIRROR_PUBLISH_RELAY_BLOCKLIST.contains(&relay.as_str()))
173 .cloned()
174 .collect::<Vec<_>>();
175 if filtered.is_empty() {
176 return active_relays;
177 }
178
179 let mut selected = Vec::new();
180 let mut selected_set = HashSet::new();
181 for relay in Self::MIRROR_PUBLISH_RELAY_PRIORITY {
182 if filtered.iter().any(|active| active == relay) {
183 selected.push((*relay).to_string());
184 selected_set.insert((*relay).to_string());
185 }
186 }
187 for relay in filtered {
188 if selected_set.insert(relay.clone()) {
189 selected.push(relay);
190 }
191 }
192
193 selected
194 }
195
196 pub fn new(
197 keys: Keys,
198 data_dir: PathBuf,
199 store: Arc<HashtreeStore>,
200 graph_store_concrete: Arc<socialgraph::SocialGraphStore>,
201 graph_store: Arc<dyn socialgraph::SocialGraphBackend>,
202 spambox: Option<Arc<dyn socialgraph::SocialGraphBackend>>,
203 webrtc_state: Option<Arc<WebRTCState>>,
204 ) -> Self {
205 Self {
206 keys,
207 data_dir,
208 store,
209 graph_store_concrete,
210 graph_store,
211 spambox,
212 webrtc_state,
213 runtime: Mutex::new(BackgroundServicesRuntime {
214 crawler: None,
215 mirror: None,
216 sync: None,
217 }),
218 }
219 }
220
221 pub async fn status(&self) -> EmbeddedBackgroundServicesStatus {
222 self.runtime.lock().await.status()
223 }
224
225 pub async fn shutdown(&self) {
226 let mut runtime = self.runtime.lock().await;
227 Self::shutdown_crawler(&mut runtime.crawler).await;
228 Self::shutdown_mirror(&mut runtime.mirror).await;
229 Self::shutdown_sync(&mut runtime.sync).await;
230 }
231
232 async fn shutdown_crawler(crawler: &mut Option<socialgraph::crawler::SocialGraphTaskHandles>) {
233 let Some(handles) = crawler.take() else {
234 return;
235 };
236
237 let _ = handles.shutdown_tx.send(true);
238
239 let mut crawl_handle = handles.crawl_handle;
240 match tokio::time::timeout(std::time::Duration::from_secs(3), &mut crawl_handle).await {
241 Ok(Ok(())) => {}
242 Ok(Err(err)) => tracing::warn!("Crawler task ended with join error: {}", err),
243 Err(_) => {
244 tracing::warn!("Timed out waiting for crawler task shutdown");
245 crawl_handle.abort();
246 }
247 }
248
249 let mut local_list_handle = handles.local_list_handle;
250 match tokio::time::timeout(std::time::Duration::from_secs(3), &mut local_list_handle).await
251 {
252 Ok(Ok(())) => {}
253 Ok(Err(err)) => tracing::warn!("Local list task ended with join error: {}", err),
254 Err(_) => {
255 tracing::warn!("Timed out waiting for local list task shutdown");
256 local_list_handle.abort();
257 }
258 }
259 }
260
261 async fn shutdown_sync(sync: &mut Option<BackgroundSyncRuntime>) {
262 let Some(mut runtime) = sync.take() else {
263 return;
264 };
265
266 runtime.service.shutdown();
267 if let Some(mut join) = runtime.join.take() {
268 match tokio::time::timeout(std::time::Duration::from_secs(3), &mut join).await {
269 Ok(Ok(())) => {}
270 Ok(Err(err)) => {
271 tracing::warn!("Background sync task ended with join error: {}", err)
272 }
273 Err(_) => {
274 tracing::warn!("Timed out waiting for background sync shutdown");
275 join.abort();
276 }
277 }
278 }
279 }
280
281 async fn shutdown_mirror(mirror: &mut Option<BackgroundMirrorRuntime>) {
282 let Some(mut runtime) = mirror.take() else {
283 return;
284 };
285
286 runtime.service.shutdown();
287 if let Some(mut join) = runtime.join.take() {
288 match tokio::time::timeout(std::time::Duration::from_secs(3), &mut join).await {
289 Ok(Ok(())) => {}
290 Ok(Err(err)) => {
291 tracing::warn!("Background mirror task ended with join error: {}", err)
292 }
293 Err(_) => {
294 tracing::warn!("Timed out waiting for background mirror shutdown");
295 join.abort();
296 }
297 }
298 }
299 }
300
301 fn nostr_mirror_config(
302 config: &Config,
303 active_relays: &[String],
304 ) -> crate::nostr_mirror::NostrMirrorConfig {
305 crate::nostr_mirror::NostrMirrorConfig {
306 relays: active_relays.to_vec(),
307 publish_relays: Self::mirror_publish_relays(active_relays, &config.server.bind_address),
308 blossom_write_servers: config.blossom.all_write_servers(),
309 max_follow_distance: config
310 .nostr
311 .mirror_max_follow_distance
312 .unwrap_or(config.nostr.social_graph_crawl_depth),
313 overmute_threshold: config.nostr.overmute_threshold,
314 require_negentropy: config.nostr.negentropy_only,
315 kinds: config.nostr.mirror_kinds.clone(),
316 history_sync_author_chunk_size: config.nostr.history_sync_author_chunk_size.max(1),
317 history_sync_per_author_event_limit: config
318 .nostr
319 .history_sync_per_author_event_limit
320 .max(1),
321 missing_profile_backfill_batch_size: config.nostr.history_sync_author_chunk_size.max(1),
322 history_sync_on_reconnect: config.nostr.history_sync_on_reconnect,
323 full_text_note_history_follow_distance: config
324 .nostr
325 .full_text_note_history_follow_distance,
326 full_text_note_history_max_relay_pages: config
327 .nostr
328 .full_text_note_history_max_relay_pages,
329 ..crate::nostr_mirror::NostrMirrorConfig::default()
330 }
331 }
332
333 pub async fn apply_config(&self, config: &Config) -> Result<EmbeddedBackgroundServicesStatus> {
334 let mut runtime = self.runtime.lock().await;
335
336 Self::shutdown_crawler(&mut runtime.crawler).await;
337 Self::shutdown_mirror(&mut runtime.mirror).await;
338 Self::shutdown_sync(&mut runtime.sync).await;
339
340 if !config.server.mode.background_services_enabled() {
341 return Ok(runtime.status());
342 }
343
344 let active_relays = config.nostr.active_relays();
345
346 if config.nostr.enabled
347 && config.nostr.social_graph_crawl_depth > 0
348 && !active_relays.is_empty()
349 {
350 runtime.crawler = Some(socialgraph::crawler::spawn_social_graph_tasks(
351 self.graph_store.clone(),
352 self.keys.clone(),
353 active_relays.clone(),
354 config.nostr.social_graph_crawl_depth,
355 self.spambox.clone(),
356 self.data_dir.clone(),
357 ));
358
359 let service = Arc::new(
360 crate::nostr_mirror::BackgroundNostrMirror::new(
361 Self::nostr_mirror_config(config, &active_relays),
362 self.store.clone(),
363 self.graph_store_concrete.clone(),
364 Some(
365 nostr_sdk::Keys::parse(&self.keys.secret_key().to_bech32()?)
366 .context("Failed to parse keys for background nostr mirror")?,
367 ),
368 )
369 .await
370 .context("Failed to create background nostr mirror")?,
371 );
372 let service_for_task = service.clone();
373 let join = tokio::task::spawn_blocking(move || {
374 let runtime = tokio::runtime::Builder::new_current_thread()
375 .enable_all()
376 .build()
377 .expect("build background nostr mirror runtime");
378 runtime.block_on(async {
379 if let Err(err) = service_for_task.run().await {
380 tracing::error!("Background nostr mirror error: {:#}", err);
381 }
382 });
383 });
384 runtime.mirror = Some(BackgroundMirrorRuntime {
385 service,
386 join: Some(join),
387 });
388 }
389
390 if config.sync.enabled && !active_relays.is_empty() {
391 let has_pinned_refs = self
392 .store
393 .list_pinned_refs()
394 .map(|refs| !refs.is_empty())
395 .unwrap_or(false);
396 let has_tracked_authors = self
397 .store
398 .list_tracked_authors()
399 .map(|authors| !authors.is_empty())
400 .unwrap_or(false);
401 let should_sync = config.sync.sync_own
402 || config.sync.sync_followed
403 || has_pinned_refs
404 || has_tracked_authors;
405 if !should_sync {
406 return Ok(runtime.status());
407 }
408
409 let sync_config = crate::sync::SyncConfig {
410 sync_own: config.sync.sync_own,
411 sync_followed: config.sync.sync_followed,
412 relays: active_relays,
413 max_concurrent: config.sync.max_concurrent,
414 webrtc_timeout_ms: config.sync.webrtc_timeout_ms,
415 blossom_timeout_ms: config.sync.blossom_timeout_ms,
416 };
417
418 let sync_keys = nostr_sdk::Keys::parse(&self.keys.secret_key().to_bech32()?)
419 .context("Failed to parse keys for sync")?;
420 let service = Arc::new(
421 crate::sync::BackgroundSync::new(
422 sync_config,
423 self.store.clone(),
424 sync_keys,
425 self.webrtc_state.clone(),
426 )
427 .await
428 .context("Failed to create background sync service")?,
429 );
430 let contacts_file = self.data_dir.join("contacts.json");
431 let service_for_task = service.clone();
432 let join = tokio::spawn(async move {
433 if let Err(err) = service_for_task.run(contacts_file).await {
434 tracing::error!("Background sync error: {}", err);
435 }
436 });
437 runtime.sync = Some(BackgroundSyncRuntime {
438 service,
439 join: Some(join),
440 });
441 }
442
443 Ok(runtime.status())
444 }
445}
446
447#[cfg(feature = "p2p")]
448pub struct EmbeddedPeerRouterController {
449 keys: Keys,
450 data_dir: PathBuf,
451 state: Arc<WebRTCState>,
452 store: Arc<dyn ContentStore>,
453 peer_classifier: PeerClassifier,
454 nostr_relay: Arc<NostrRelay>,
455 runtime: Mutex<Option<PeerRouterRuntime>>,
456}
457
458#[cfg(feature = "p2p")]
459impl EmbeddedPeerRouterController {
460 pub fn new(
461 keys: Keys,
462 data_dir: PathBuf,
463 state: Arc<WebRTCState>,
464 store: Arc<dyn ContentStore>,
465 peer_classifier: PeerClassifier,
466 nostr_relay: Arc<NostrRelay>,
467 ) -> Self {
468 Self {
469 keys,
470 data_dir,
471 state,
472 store,
473 peer_classifier,
474 nostr_relay,
475 runtime: Mutex::new(None),
476 }
477 }
478
479 pub fn state(&self) -> Arc<WebRTCState> {
480 self.state.clone()
481 }
482
483 pub async fn apply_config(&self, config: &Config) -> Result<bool> {
484 let mut runtime = self.runtime.lock().await;
485 if let Some(runtime_handle) = runtime.take() {
486 if let Err(err) =
487 crate::p2p_common::persist_peer_state(&self.data_dir, &self.state).await
488 {
489 tracing::warn!("Failed to persist mesh peer state before router restart: {err:#}");
490 }
491 let _ = runtime_handle.shutdown.send(true);
492 runtime_handle.peer_state_persist.abort();
493 let mut join = runtime_handle.join;
494 match tokio::time::timeout(std::time::Duration::from_secs(3), &mut join).await {
495 Ok(Ok(())) => {}
496 Ok(Err(err)) => {
497 tracing::warn!("Peer router task ended with join error: {}", err);
498 }
499 Err(_) => {
500 tracing::warn!("Timed out waiting for peer router shutdown");
501 join.abort();
502 }
503 }
504 }
505
506 self.state.reset_runtime_state().await;
507 if let Err(err) = crate::p2p_common::load_peer_state(&self.data_dir, &self.state).await {
508 tracing::warn!("Failed to load persisted mesh peer state: {err:#}");
509 }
510
511 if !crate::p2p_common::peer_router_enabled(config) {
512 return Ok(false);
513 }
514
515 let webrtc_config = crate::p2p_common::default_webrtc_config(config);
516 let mut manager = if config.server.mode.hash_get_enabled() {
517 WebRTCManager::new_with_state_and_store_and_classifier(
518 self.keys.clone(),
519 webrtc_config,
520 self.state.clone(),
521 self.store.clone(),
522 self.peer_classifier.clone(),
523 )
524 } else {
525 let mut manager =
526 WebRTCManager::new_with_state(self.keys.clone(), webrtc_config, self.state.clone());
527 manager.set_peer_classifier(self.peer_classifier.clone());
528 manager
529 };
530 manager
531 .set_nostr_relay(self.nostr_relay.clone() as hashtree_network::SharedMeshRelayClient);
532 let shutdown = manager.shutdown_signal();
533 let join = tokio::spawn(async move {
534 if let Err(err) = manager.run().await {
535 tracing::error!("Peer router error: {}", err);
536 }
537 });
538 let peer_state_persist = crate::p2p_common::spawn_peer_state_persist_task(
539 self.data_dir.clone(),
540 self.state.clone(),
541 );
542 *runtime = Some(PeerRouterRuntime {
543 shutdown,
544 join,
545 peer_state_persist,
546 });
547 Ok(true)
548 }
549
550 pub async fn shutdown(&self) {
551 let mut runtime = self.runtime.lock().await;
552 let Some(runtime_handle) = runtime.take() else {
553 return;
554 };
555
556 if let Err(err) = crate::p2p_common::persist_peer_state(&self.data_dir, &self.state).await {
557 tracing::warn!("Failed to persist mesh peer state during router shutdown: {err:#}");
558 }
559 let _ = runtime_handle.shutdown.send(true);
560 runtime_handle.peer_state_persist.abort();
561 let mut join = runtime_handle.join;
562 match tokio::time::timeout(std::time::Duration::from_secs(3), &mut join).await {
563 Ok(Ok(())) => {}
564 Ok(Err(err)) => tracing::warn!("Peer router task ended with join error: {}", err),
565 Err(_) => {
566 tracing::warn!("Timed out waiting for peer router shutdown");
567 join.abort();
568 }
569 }
570
571 self.state.reset_runtime_state().await;
572 }
573}
574
575pub struct EmbeddedDaemonController {
576 server_controller: Arc<EmbeddedServerController>,
577 fips_handle: Option<Arc<crate::fips_transport::DaemonFipsHandle>>,
578 #[cfg(feature = "experimental-decentralized-pubsub")]
579 nostr_pubsub_handle: Option<Arc<crate::fips_transport::DaemonNostrPubsubHandle>>,
580 #[cfg(feature = "p2p")]
581 peer_router_controller: Option<Arc<EmbeddedPeerRouterController>>,
582 background_services_controller: Option<Arc<EmbeddedBackgroundServicesController>>,
583}
584
585impl EmbeddedDaemonController {
586 #[cfg(feature = "p2p")]
587 pub fn new(
588 server_controller: Arc<EmbeddedServerController>,
589 fips_handle: Option<Arc<crate::fips_transport::DaemonFipsHandle>>,
590 #[cfg(feature = "experimental-decentralized-pubsub")] nostr_pubsub_handle: Option<
591 Arc<crate::fips_transport::DaemonNostrPubsubHandle>,
592 >,
593 peer_router_controller: Option<Arc<EmbeddedPeerRouterController>>,
594 background_services_controller: Option<Arc<EmbeddedBackgroundServicesController>>,
595 ) -> Self {
596 Self {
597 server_controller,
598 fips_handle,
599 #[cfg(feature = "experimental-decentralized-pubsub")]
600 nostr_pubsub_handle,
601 #[cfg(feature = "p2p")]
602 peer_router_controller,
603 background_services_controller,
604 }
605 }
606
607 #[cfg(not(feature = "p2p"))]
608 pub fn new(
609 server_controller: Arc<EmbeddedServerController>,
610 fips_handle: Option<Arc<crate::fips_transport::DaemonFipsHandle>>,
611 #[cfg(feature = "experimental-decentralized-pubsub")] nostr_pubsub_handle: Option<
612 Arc<crate::fips_transport::DaemonNostrPubsubHandle>,
613 >,
614 background_services_controller: Option<Arc<EmbeddedBackgroundServicesController>>,
615 ) -> Self {
616 Self {
617 server_controller,
618 fips_handle,
619 #[cfg(feature = "experimental-decentralized-pubsub")]
620 nostr_pubsub_handle,
621 background_services_controller,
622 }
623 }
624
625 pub async fn shutdown(&self) {
626 self.server_controller.shutdown().await;
627 #[cfg(feature = "experimental-decentralized-pubsub")]
628 if let Some(handle) = self.nostr_pubsub_handle.as_ref() {
629 handle.shutdown();
630 }
631 if let Some(handle) = self.fips_handle.as_ref() {
632 handle.shutdown();
633 }
634 if let Some(controller) = self.background_services_controller.as_ref() {
635 controller.shutdown().await;
636 }
637 #[cfg(feature = "p2p")]
638 if let Some(controller) = self.peer_router_controller.as_ref() {
639 controller.shutdown().await;
640 }
641 }
642}
643
644pub struct EmbeddedDaemonOptions {
645 pub config: Config,
646 pub data_dir: PathBuf,
647 pub config_dir: Option<PathBuf>,
648 pub bind_address: String,
649 pub relays: Option<Vec<String>>,
650 pub initial_tree_roots: Vec<(String, Cid)>,
651 pub extra_routes: Option<Router<AppState>>,
652 pub cors: Option<CorsLayer>,
653}
654
655pub struct EmbeddedDaemonInfo {
656 pub addr: String,
657 pub port: u16,
658 pub npub: String,
659 pub store: Arc<HashtreeStore>,
660 pub daemon_controller: Arc<EmbeddedDaemonController>,
661 #[allow(dead_code)]
662 pub webrtc_state: Option<Arc<WebRTCState>>,
663 #[cfg(feature = "p2p")]
664 #[allow(dead_code)]
665 pub peer_router_controller: Option<Arc<EmbeddedPeerRouterController>>,
666 #[allow(dead_code)]
667 pub background_services_controller: Option<Arc<EmbeddedBackgroundServicesController>>,
668}
669
670pub async fn start_embedded(opts: EmbeddedDaemonOptions) -> Result<EmbeddedDaemonInfo> {
671 let _ = rustls::crypto::ring::default_provider().install_default();
672
673 let mut config = opts.config;
674 config.server.bind_address = opts.bind_address.clone();
675 if let Some(relays) = opts.relays {
676 config.nostr.relays = relays;
677 config.nostr.enabled = embedded_nostr_enabled_after_relay_override(&config);
678 }
679
680 let max_size_bytes = config.storage.max_size_gb * 1024 * 1024 * 1024;
681 let nostr_db_max_bytes = config
682 .nostr
683 .db_max_size_gb
684 .saturating_mul(1024 * 1024 * 1024);
685 let spambox_db_max_bytes = config
686 .nostr
687 .spambox_max_size_gb
688 .saturating_mul(1024 * 1024 * 1024);
689
690 let store = Arc::new(HashtreeStore::with_embedded_options(
691 &opts.data_dir,
692 config.storage.s3.as_ref(),
693 max_size_bytes,
694 )?);
695
696 let (keys, _was_generated) = if let Some(config_dir) = opts.config_dir.as_ref() {
697 ensure_keys_in(config_dir, Some(&opts.data_dir), Some(&config))?
698 } else {
699 ensure_keys()?
700 };
701 let pk_bytes = pubkey_bytes(&keys);
702 let npub = keys
703 .public_key()
704 .to_bech32()
705 .context("Failed to encode npub")?;
706
707 let mut allowed_pubkeys: HashSet<String> = HashSet::new();
708 allowed_pubkeys.insert(hex::encode(pk_bytes));
709 for npub_str in &config.nostr.allowed_npubs {
710 if let Ok(pk) = parse_npub(npub_str) {
711 allowed_pubkeys.insert(hex::encode(pk));
712 } else {
713 tracing::warn!("Invalid npub in allowed_npubs: {}", npub_str);
714 }
715 }
716
717 let graph_store = socialgraph::open_embedded_social_graph_store_with_storage(
718 &opts.data_dir,
719 store.store_arc(),
720 Some(nostr_db_max_bytes),
721 )
722 .context("Failed to initialize social graph store")?;
723 graph_store.set_profile_index_overmute_threshold(config.nostr.overmute_threshold);
724
725 let social_graph_root_bytes = if let Some(ref root_npub) = config.nostr.socialgraph_root {
726 parse_npub(root_npub).unwrap_or(pk_bytes)
727 } else {
728 pk_bytes
729 };
730 socialgraph::set_social_graph_root(&graph_store, &social_graph_root_bytes);
731 socialgraph::sync_local_list_files_force(graph_store.as_ref(), &opts.data_dir, &keys)
732 .context("Failed to sync local social graph lists")?;
733 let fips_peer_ids = crate::fips_transport::fips_peer_ids_from_pubkeys(
734 socialgraph::get_follows(graph_store.as_ref(), &pk_bytes),
735 );
736 let social_graph_store: Arc<dyn socialgraph::SocialGraphBackend> = graph_store.clone();
737
738 let social_graph = Arc::new(socialgraph::SocialGraphAccessControl::new(
739 Arc::clone(&social_graph_store),
740 config.nostr.max_write_distance,
741 allowed_pubkeys.clone(),
742 ));
743
744 let nostr_relay_config = NostrRelayConfig {
745 spambox_db_max_bytes,
746 ..Default::default()
747 };
748 let nostr_relay = if config.nostr.enabled {
749 let mut public_event_pubkeys = HashSet::new();
750 public_event_pubkeys.insert(hex::encode(pk_bytes));
751 Some(Arc::new(
752 NostrRelay::new(
753 Arc::clone(&social_graph_store),
754 opts.data_dir.clone(),
755 public_event_pubkeys,
756 Some(social_graph.clone()),
757 nostr_relay_config,
758 )
759 .map(|relay| {
760 relay.with_historical_nostr_index(store.store_arc(), opts.data_dir.clone())
761 })
762 .context("Failed to initialize Nostr relay")?,
763 ))
764 } else {
765 None
766 };
767
768 let crawler_spambox = if config.nostr.enabled && spambox_db_max_bytes != 0 {
769 let spam_dir = opts.data_dir.join("socialgraph_spambox");
770 match socialgraph::open_embedded_social_graph_store_at_path(
771 &spam_dir,
772 Some(spambox_db_max_bytes),
773 ) {
774 Ok(store) => Some(store),
775 Err(err) => {
776 tracing::warn!("Failed to open social graph spambox for crawler: {}", err);
777 None
778 }
779 }
780 } else {
781 None
782 };
783 let crawler_spambox_backend = crawler_spambox
784 .clone()
785 .map(|store| store as Arc<dyn socialgraph::SocialGraphBackend>);
786
787 #[cfg(feature = "p2p")]
788 let (webrtc_state, peer_router_controller): (
789 Option<Arc<WebRTCState>>,
790 Option<Arc<EmbeddedPeerRouterController>>,
791 ) = if let Some(nostr_relay) = nostr_relay.clone() {
792 let router_config = crate::p2p_common::default_webrtc_config(&config);
793 let peer_classifier = crate::p2p_common::build_peer_classifier(
794 opts.data_dir.clone(),
795 Arc::clone(&social_graph_store),
796 );
797 let cashu_payment_client =
798 if config.cashu.default_mint.is_some() || !config.cashu.accepted_mints.is_empty() {
799 match crate::cashu_helper::CashuHelperClient::discover(opts.data_dir.clone()) {
800 Ok(client) => {
801 Some(Arc::new(client) as Arc<dyn crate::cashu_helper::CashuPaymentClient>)
802 }
803 Err(err) => {
804 tracing::warn!(
805 "Cashu settlement helper unavailable; paid retrieval stays disabled: {}",
806 err
807 );
808 None
809 }
810 }
811 } else {
812 None
813 };
814 let cashu_mint_metadata =
815 if config.cashu.default_mint.is_some() || !config.cashu.accepted_mints.is_empty() {
816 let metadata_path = crate::webrtc::cashu_mint_metadata_path(&opts.data_dir);
817 match crate::webrtc::CashuMintMetadataStore::load(metadata_path) {
818 Ok(store) => Some(store),
819 Err(err) => {
820 tracing::warn!(
821 "Failed to load Cashu mint metadata; falling back to in-memory state: {}",
822 err
823 );
824 Some(crate::webrtc::CashuMintMetadataStore::in_memory())
825 }
826 }
827 } else {
828 None
829 };
830
831 let state = Arc::new(WebRTCState::new_with_routing_and_cashu(
832 router_config.request_selection_strategy,
833 router_config.request_fairness_enabled,
834 router_config.request_dispatch,
835 std::time::Duration::from_millis(router_config.message_timeout_ms),
836 crate::webrtc::CashuRoutingConfig::from(&config.cashu),
837 cashu_payment_client,
838 cashu_mint_metadata,
839 ));
840 let controller = Arc::new(EmbeddedPeerRouterController::new(
841 keys.clone(),
842 opts.data_dir.clone(),
843 state.clone(),
844 Arc::clone(&store) as Arc<dyn ContentStore>,
845 peer_classifier,
846 nostr_relay.clone(),
847 ));
848 controller.apply_config(&config).await?;
849 (Some(state), Some(controller))
850 } else {
851 (None, None)
852 };
853
854 #[cfg(not(feature = "p2p"))]
855 let webrtc_state: Option<Arc<crate::webrtc::WebRTCState>> = None;
856
857 let background_services_controller = Arc::new(EmbeddedBackgroundServicesController::new(
858 keys.clone(),
859 opts.data_dir.clone(),
860 Arc::clone(&store),
861 graph_store.clone(),
862 Arc::clone(&social_graph_store),
863 crawler_spambox_backend,
864 webrtc_state.clone(),
865 ));
866
867 let upstream_blossom = config.blossom.all_read_servers();
868 let blossom_replica_queue_bytes = crate::server::bounded_upload_queue_bytes(
869 config
870 .blossom
871 .replicate_queue_mb
872 .saturating_mul(1024 * 1024),
873 );
874 let active_nostr_relays = config.nostr.active_relays();
875 let fips_handle = crate::fips_transport::start_daemon_fips_transport(
876 &config,
877 &keys,
878 Arc::clone(&store),
879 fips_peer_ids,
880 )
881 .await?
882 .map(Arc::new);
883 let nostr_provider =
884 crate::fips_transport::start_daemon_nostr_provider(&config, fips_handle.as_deref()).await?;
885 #[cfg(feature = "experimental-decentralized-pubsub")]
886 let nostr_pubsub_handle = crate::fips_transport::start_daemon_nostr_pubsub(
887 &config,
888 fips_handle.as_deref(),
889 Arc::clone(&store),
890 nostr_relay.clone(),
891 )
892 .await?;
893
894 let mut server = HashtreeServer::new(Arc::clone(&store), opts.bind_address.clone())
895 .with_server_mode(config.server.mode)
896 .with_hash_get_enabled(config.server.mode.hash_get_enabled())
897 .with_fetch_from_fips_peers(config.server.fetch_from_fips_peers)
898 .with_allowed_pubkeys(allowed_pubkeys.clone())
899 .with_max_upload_bytes((config.blossom.max_upload_mb as usize) * 1024 * 1024)
900 .with_public_writes(config.server.public_writes)
901 .with_public_plaintext_reads(config.server.public_plaintext_reads)
902 .with_require_random_untrusted_ingest(config.blossom.require_random_untrusted_ingest)
903 .with_optimistic_blossom_uploads(config.blossom.optimistic_uploads)
904 .with_upstream_blossom(upstream_blossom)
905 .with_blossom_upload_replicas(
906 config.blossom.replicate_servers.clone(),
907 blossom_replica_queue_bytes,
908 keys.clone(),
909 )
910 .with_nostr_relay_urls(active_nostr_relays)
911 .with_cached_tree_roots(opts.initial_tree_roots)
912 .with_social_graph(social_graph)
913 .with_socialgraph_snapshot(
914 Arc::clone(&social_graph_store),
915 social_graph_root_bytes,
916 config.server.socialgraph_snapshot_public,
917 );
918 if let Some(nostr_relay) = nostr_relay {
919 server = server.with_nostr_relay(nostr_relay);
920 }
921 if let Some(provider) = nostr_provider {
922 server = server.with_nostr_provider(provider);
923 }
924
925 if crate::p2p_common::peer_router_enabled(&config) {
926 if let Some(ref state) = webrtc_state {
927 server = server.with_webrtc_peers(state.clone());
928 }
929 }
930 if let Some(ref fips_handle) = fips_handle {
931 server = server.with_fips_transport(fips_handle.transport.clone());
932 }
933
934 if let Some(extra) = opts.extra_routes {
935 server = server.with_extra_routes(extra);
936 }
937 if let Some(cors) = opts.cors {
938 server = server.with_cors(cors);
939 }
940
941 spawn_background_eviction_task(
942 Arc::clone(&store),
943 BACKGROUND_EVICTION_INTERVAL,
944 "embedded daemon",
945 );
946
947 let listener = TcpListener::bind(&opts.bind_address).await?;
948 let local_addr = listener.local_addr()?;
949 let actual_addr = format!("{}:{}", local_addr.ip(), local_addr.port());
950
951 let server_shutdown = Arc::new(Notify::new());
952 let server_shutdown_for_task = Arc::clone(&server_shutdown);
953 let server_join = tokio::spawn(async move {
954 if let Err(e) = server
955 .run_with_listener_until(listener, async move {
956 server_shutdown_for_task.notified().await;
957 })
958 .await
959 {
960 tracing::error!("Embedded daemon server error: {}", e);
961 }
962 });
963 let server_controller = Arc::new(EmbeddedServerController::new(server_shutdown, server_join));
964 background_services_controller.apply_config(&config).await?;
965 #[cfg(feature = "p2p")]
966 let daemon_controller = Arc::new(EmbeddedDaemonController::new(
967 server_controller,
968 fips_handle.clone(),
969 #[cfg(feature = "experimental-decentralized-pubsub")]
970 nostr_pubsub_handle.clone(),
971 peer_router_controller.clone(),
972 Some(background_services_controller.clone()),
973 ));
974 #[cfg(not(feature = "p2p"))]
975 let daemon_controller = Arc::new(EmbeddedDaemonController::new(
976 server_controller,
977 fips_handle.clone(),
978 #[cfg(feature = "experimental-decentralized-pubsub")]
979 nostr_pubsub_handle.clone(),
980 Some(background_services_controller.clone()),
981 ));
982
983 tracing::info!(
984 "Embedded daemon started on {}, identity {}",
985 actual_addr,
986 npub
987 );
988
989 Ok(EmbeddedDaemonInfo {
990 addr: actual_addr,
991 port: local_addr.port(),
992 npub,
993 store,
994 daemon_controller,
995 webrtc_state,
996 #[cfg(feature = "p2p")]
997 peer_router_controller,
998 background_services_controller: Some(background_services_controller),
999 })
1000}
1001
1002fn embedded_nostr_enabled_after_relay_override(config: &Config) -> bool {
1003 config.nostr.decentralized_pubsub || !config.nostr.relays.is_empty()
1004}
1005
1006#[cfg(test)]
1007mod tests {
1008 use super::{
1009 embedded_nostr_enabled_after_relay_override, EmbeddedBackgroundServicesController,
1010 };
1011 use crate::config::Config;
1012
1013 #[test]
1014 fn mirror_publish_relays_orders_known_root_publish_relays_first() {
1015 let relays = EmbeddedBackgroundServicesController::mirror_publish_relays(
1016 &[
1017 "wss://graph-relay.iris.to".to_string(),
1018 "wss://relay.example".to_string(),
1019 "wss://relay.primal.net".to_string(),
1020 "wss://relay.damus.io".to_string(),
1021 "wss://temp.iris.to".to_string(),
1022 "wss://vault.iris.to".to_string(),
1023 "wss://upload.iris.to/nostr".to_string(),
1024 ],
1025 "0.0.0.0:8080",
1026 );
1027 assert_eq!(
1028 relays,
1029 vec![
1030 "wss://temp.iris.to".to_string(),
1031 "wss://vault.iris.to".to_string(),
1032 "wss://relay.damus.io".to_string(),
1033 "wss://relay.example".to_string(),
1034 "wss://relay.primal.net".to_string(),
1035 ]
1036 );
1037 }
1038
1039 #[test]
1040 fn mirror_publish_relays_do_not_add_non_active_publish_targets() {
1041 let relays = EmbeddedBackgroundServicesController::mirror_publish_relays(
1042 &[
1043 "wss://graph-relay.iris.to".to_string(),
1044 "wss://relay.example".to_string(),
1045 ],
1046 "0.0.0.0:8080",
1047 );
1048 assert_eq!(relays, vec!["wss://relay.example".to_string()]);
1049 }
1050
1051 #[test]
1052 fn mirror_publish_relays_falls_back_to_active_relays_when_all_are_blocklisted() {
1053 let relays = EmbeddedBackgroundServicesController::mirror_publish_relays(
1054 &[
1055 "wss://graph-relay.iris.to".to_string(),
1056 "wss://upload.iris.to/nostr".to_string(),
1057 ],
1058 "0.0.0.0:8080",
1059 );
1060 assert_eq!(
1061 relays,
1062 vec![
1063 "wss://graph-relay.iris.to".to_string(),
1064 "wss://upload.iris.to/nostr".to_string(),
1065 ]
1066 );
1067 }
1068
1069 #[test]
1070 fn nostr_mirror_config_allows_disabling_full_note_paging() {
1071 let mut config = Config::default();
1072 config.nostr.full_text_note_history_max_relay_pages = 0;
1073
1074 let mirror_config = EmbeddedBackgroundServicesController::nostr_mirror_config(
1075 &config,
1076 &["wss://relay.example".to_string()],
1077 );
1078
1079 assert_eq!(mirror_config.full_text_note_history_max_relay_pages, 0);
1080
1081 config.nostr.full_text_note_history_max_relay_pages = 64;
1082 let mirror_config = EmbeddedBackgroundServicesController::nostr_mirror_config(
1083 &config,
1084 &["wss://relay.example".to_string()],
1085 );
1086
1087 assert_eq!(mirror_config.full_text_note_history_max_relay_pages, 64);
1088 }
1089
1090 #[test]
1091 fn nostr_mirror_config_can_limit_mirror_distance_independently() {
1092 let mut config = Config::default();
1093 config.nostr.social_graph_crawl_depth = 6;
1094 config.nostr.mirror_max_follow_distance = Some(2);
1095
1096 let mirror_config = EmbeddedBackgroundServicesController::nostr_mirror_config(
1097 &config,
1098 &["wss://relay.example".to_string()],
1099 );
1100
1101 assert_eq!(mirror_config.max_follow_distance, 2);
1102
1103 config.nostr.mirror_max_follow_distance = None;
1104 let mirror_config = EmbeddedBackgroundServicesController::nostr_mirror_config(
1105 &config,
1106 &["wss://relay.example".to_string()],
1107 );
1108
1109 assert_eq!(mirror_config.max_follow_distance, 6);
1110 }
1111
1112 #[test]
1113 fn embedded_empty_relays_keep_nostr_enabled_for_decentralized_pubsub() {
1114 let mut config = Config::default();
1115 config.nostr.relays = Vec::new();
1116 config.nostr.decentralized_pubsub = false;
1117 assert!(!embedded_nostr_enabled_after_relay_override(&config));
1118
1119 config.nostr.decentralized_pubsub = true;
1120 assert!(embedded_nostr_enabled_after_relay_override(&config));
1121
1122 config.nostr.decentralized_pubsub = false;
1123 config.nostr.relays = vec!["wss://relay.example".to_string()];
1124 assert!(embedded_nostr_enabled_after_relay_override(&config));
1125 }
1126}