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 = "p2p")]
579 peer_router_controller: Option<Arc<EmbeddedPeerRouterController>>,
580 background_services_controller: Option<Arc<EmbeddedBackgroundServicesController>>,
581}
582
583impl EmbeddedDaemonController {
584 #[cfg(feature = "p2p")]
585 pub fn new(
586 server_controller: Arc<EmbeddedServerController>,
587 fips_handle: Option<Arc<crate::fips_transport::DaemonFipsHandle>>,
588 peer_router_controller: Option<Arc<EmbeddedPeerRouterController>>,
589 background_services_controller: Option<Arc<EmbeddedBackgroundServicesController>>,
590 ) -> Self {
591 Self {
592 server_controller,
593 fips_handle,
594 #[cfg(feature = "p2p")]
595 peer_router_controller,
596 background_services_controller,
597 }
598 }
599
600 #[cfg(not(feature = "p2p"))]
601 pub fn new(
602 server_controller: Arc<EmbeddedServerController>,
603 fips_handle: Option<Arc<crate::fips_transport::DaemonFipsHandle>>,
604 background_services_controller: Option<Arc<EmbeddedBackgroundServicesController>>,
605 ) -> Self {
606 Self {
607 server_controller,
608 fips_handle,
609 background_services_controller,
610 }
611 }
612
613 pub async fn shutdown(&self) {
614 self.server_controller.shutdown().await;
615 if let Some(handle) = self.fips_handle.as_ref() {
616 handle.shutdown();
617 }
618 if let Some(controller) = self.background_services_controller.as_ref() {
619 controller.shutdown().await;
620 }
621 #[cfg(feature = "p2p")]
622 if let Some(controller) = self.peer_router_controller.as_ref() {
623 controller.shutdown().await;
624 }
625 }
626}
627
628pub struct EmbeddedDaemonOptions {
629 pub config: Config,
630 pub data_dir: PathBuf,
631 pub config_dir: Option<PathBuf>,
632 pub bind_address: String,
633 pub relays: Option<Vec<String>>,
634 pub initial_tree_roots: Vec<(String, Cid)>,
635 pub extra_routes: Option<Router<AppState>>,
636 pub cors: Option<CorsLayer>,
637}
638
639pub struct EmbeddedDaemonInfo {
640 pub addr: String,
641 pub port: u16,
642 pub npub: String,
643 pub store: Arc<HashtreeStore>,
644 pub daemon_controller: Arc<EmbeddedDaemonController>,
645 #[allow(dead_code)]
646 pub webrtc_state: Option<Arc<WebRTCState>>,
647 #[cfg(feature = "p2p")]
648 #[allow(dead_code)]
649 pub peer_router_controller: Option<Arc<EmbeddedPeerRouterController>>,
650 #[allow(dead_code)]
651 pub background_services_controller: Option<Arc<EmbeddedBackgroundServicesController>>,
652}
653
654pub async fn start_embedded(opts: EmbeddedDaemonOptions) -> Result<EmbeddedDaemonInfo> {
655 let _ = rustls::crypto::ring::default_provider().install_default();
656
657 let mut config = opts.config;
658 config.server.bind_address = opts.bind_address.clone();
659 if let Some(relays) = opts.relays {
660 config.nostr.relays = relays;
661 config.nostr.enabled = !config.nostr.relays.is_empty();
662 }
663
664 let max_size_bytes = config.storage.max_size_gb * 1024 * 1024 * 1024;
665 let nostr_db_max_bytes = config
666 .nostr
667 .db_max_size_gb
668 .saturating_mul(1024 * 1024 * 1024);
669 let spambox_db_max_bytes = config
670 .nostr
671 .spambox_max_size_gb
672 .saturating_mul(1024 * 1024 * 1024);
673
674 let store = Arc::new(HashtreeStore::with_embedded_options(
675 &opts.data_dir,
676 config.storage.s3.as_ref(),
677 max_size_bytes,
678 )?);
679
680 let (keys, _was_generated) = if let Some(config_dir) = opts.config_dir.as_ref() {
681 ensure_keys_in(config_dir, Some(&opts.data_dir), Some(&config))?
682 } else {
683 ensure_keys()?
684 };
685 let pk_bytes = pubkey_bytes(&keys);
686 let npub = keys
687 .public_key()
688 .to_bech32()
689 .context("Failed to encode npub")?;
690
691 let mut allowed_pubkeys: HashSet<String> = HashSet::new();
692 allowed_pubkeys.insert(hex::encode(pk_bytes));
693 for npub_str in &config.nostr.allowed_npubs {
694 if let Ok(pk) = parse_npub(npub_str) {
695 allowed_pubkeys.insert(hex::encode(pk));
696 } else {
697 tracing::warn!("Invalid npub in allowed_npubs: {}", npub_str);
698 }
699 }
700
701 let graph_store = socialgraph::open_embedded_social_graph_store_with_storage(
702 &opts.data_dir,
703 store.store_arc(),
704 Some(nostr_db_max_bytes),
705 )
706 .context("Failed to initialize social graph store")?;
707 graph_store.set_profile_index_overmute_threshold(config.nostr.overmute_threshold);
708
709 let social_graph_root_bytes = if let Some(ref root_npub) = config.nostr.socialgraph_root {
710 parse_npub(root_npub).unwrap_or(pk_bytes)
711 } else {
712 pk_bytes
713 };
714 socialgraph::set_social_graph_root(&graph_store, &social_graph_root_bytes);
715 socialgraph::sync_local_list_files_force(graph_store.as_ref(), &opts.data_dir, &keys)
716 .context("Failed to sync local social graph lists")?;
717 let fips_peer_ids = crate::fips_transport::fips_peer_ids_from_pubkeys(
718 socialgraph::get_follows(graph_store.as_ref(), &pk_bytes),
719 );
720 let social_graph_store: Arc<dyn socialgraph::SocialGraphBackend> = graph_store.clone();
721
722 let social_graph = Arc::new(socialgraph::SocialGraphAccessControl::new(
723 Arc::clone(&social_graph_store),
724 config.nostr.max_write_distance,
725 allowed_pubkeys.clone(),
726 ));
727
728 let nostr_relay_config = NostrRelayConfig {
729 spambox_db_max_bytes,
730 ..Default::default()
731 };
732 let nostr_relay = if config.nostr.enabled {
733 let mut public_event_pubkeys = HashSet::new();
734 public_event_pubkeys.insert(hex::encode(pk_bytes));
735 Some(Arc::new(
736 NostrRelay::new(
737 Arc::clone(&social_graph_store),
738 opts.data_dir.clone(),
739 public_event_pubkeys,
740 Some(social_graph.clone()),
741 nostr_relay_config,
742 )
743 .context("Failed to initialize Nostr relay")?,
744 ))
745 } else {
746 None
747 };
748
749 let crawler_spambox = if config.nostr.enabled && spambox_db_max_bytes != 0 {
750 let spam_dir = opts.data_dir.join("socialgraph_spambox");
751 match socialgraph::open_embedded_social_graph_store_at_path(
752 &spam_dir,
753 Some(spambox_db_max_bytes),
754 ) {
755 Ok(store) => Some(store),
756 Err(err) => {
757 tracing::warn!("Failed to open social graph spambox for crawler: {}", err);
758 None
759 }
760 }
761 } else {
762 None
763 };
764 let crawler_spambox_backend = crawler_spambox
765 .clone()
766 .map(|store| store as Arc<dyn socialgraph::SocialGraphBackend>);
767
768 #[cfg(feature = "p2p")]
769 let (webrtc_state, peer_router_controller): (
770 Option<Arc<WebRTCState>>,
771 Option<Arc<EmbeddedPeerRouterController>>,
772 ) = if let Some(nostr_relay) = nostr_relay.clone() {
773 let router_config = crate::p2p_common::default_webrtc_config(&config);
774 let peer_classifier = crate::p2p_common::build_peer_classifier(
775 opts.data_dir.clone(),
776 Arc::clone(&social_graph_store),
777 );
778 let cashu_payment_client =
779 if config.cashu.default_mint.is_some() || !config.cashu.accepted_mints.is_empty() {
780 match crate::cashu_helper::CashuHelperClient::discover(opts.data_dir.clone()) {
781 Ok(client) => {
782 Some(Arc::new(client) as Arc<dyn crate::cashu_helper::CashuPaymentClient>)
783 }
784 Err(err) => {
785 tracing::warn!(
786 "Cashu settlement helper unavailable; paid retrieval stays disabled: {}",
787 err
788 );
789 None
790 }
791 }
792 } else {
793 None
794 };
795 let cashu_mint_metadata =
796 if config.cashu.default_mint.is_some() || !config.cashu.accepted_mints.is_empty() {
797 let metadata_path = crate::webrtc::cashu_mint_metadata_path(&opts.data_dir);
798 match crate::webrtc::CashuMintMetadataStore::load(metadata_path) {
799 Ok(store) => Some(store),
800 Err(err) => {
801 tracing::warn!(
802 "Failed to load Cashu mint metadata; falling back to in-memory state: {}",
803 err
804 );
805 Some(crate::webrtc::CashuMintMetadataStore::in_memory())
806 }
807 }
808 } else {
809 None
810 };
811
812 let state = Arc::new(WebRTCState::new_with_routing_and_cashu(
813 router_config.request_selection_strategy,
814 router_config.request_fairness_enabled,
815 router_config.request_dispatch,
816 std::time::Duration::from_millis(router_config.message_timeout_ms),
817 crate::webrtc::CashuRoutingConfig::from(&config.cashu),
818 cashu_payment_client,
819 cashu_mint_metadata,
820 ));
821 let controller = Arc::new(EmbeddedPeerRouterController::new(
822 keys.clone(),
823 opts.data_dir.clone(),
824 state.clone(),
825 Arc::clone(&store) as Arc<dyn ContentStore>,
826 peer_classifier,
827 nostr_relay.clone(),
828 ));
829 controller.apply_config(&config).await?;
830 (Some(state), Some(controller))
831 } else {
832 (None, None)
833 };
834
835 #[cfg(not(feature = "p2p"))]
836 let webrtc_state: Option<Arc<crate::webrtc::WebRTCState>> = None;
837
838 let background_services_controller = Arc::new(EmbeddedBackgroundServicesController::new(
839 keys.clone(),
840 opts.data_dir.clone(),
841 Arc::clone(&store),
842 graph_store.clone(),
843 Arc::clone(&social_graph_store),
844 crawler_spambox_backend,
845 webrtc_state.clone(),
846 ));
847
848 let upstream_blossom = config.blossom.all_read_servers();
849 let blossom_replica_queue_bytes =
850 (config.blossom.replicate_queue_mb.max(1) as usize) * 1024 * 1024;
851 let active_nostr_relays = config.nostr.active_relays();
852 let fips_handle = crate::fips_transport::start_daemon_fips_transport(
853 &config,
854 &keys,
855 Arc::clone(&store),
856 fips_peer_ids,
857 )
858 .await?
859 .map(Arc::new);
860
861 let mut server = HashtreeServer::new(Arc::clone(&store), opts.bind_address.clone())
862 .with_server_mode(config.server.mode)
863 .with_hash_get_enabled(config.server.mode.hash_get_enabled())
864 .with_fetch_from_fips_peers(config.server.fetch_from_fips_peers)
865 .with_allowed_pubkeys(allowed_pubkeys.clone())
866 .with_max_upload_bytes((config.blossom.max_upload_mb as usize) * 1024 * 1024)
867 .with_public_writes(config.server.public_writes)
868 .with_public_plaintext_reads(config.server.public_plaintext_reads)
869 .with_require_random_untrusted_ingest(config.blossom.require_random_untrusted_ingest)
870 .with_optimistic_blossom_uploads(config.blossom.optimistic_uploads)
871 .with_upstream_blossom(upstream_blossom)
872 .with_blossom_upload_replicas(
873 config.blossom.replicate_servers.clone(),
874 blossom_replica_queue_bytes,
875 keys.clone(),
876 )
877 .with_nostr_relay_urls(active_nostr_relays)
878 .with_cached_tree_roots(opts.initial_tree_roots)
879 .with_social_graph(social_graph)
880 .with_socialgraph_snapshot(
881 Arc::clone(&social_graph_store),
882 social_graph_root_bytes,
883 config.server.socialgraph_snapshot_public,
884 );
885 if let Some(nostr_relay) = nostr_relay {
886 server = server.with_nostr_relay(nostr_relay);
887 }
888
889 if crate::p2p_common::peer_router_enabled(&config) {
890 if let Some(ref state) = webrtc_state {
891 server = server.with_webrtc_peers(state.clone());
892 }
893 }
894 if let Some(ref fips_handle) = fips_handle {
895 server = server.with_fips_transport(fips_handle.transport.clone());
896 }
897
898 if let Some(extra) = opts.extra_routes {
899 server = server.with_extra_routes(extra);
900 }
901 if let Some(cors) = opts.cors {
902 server = server.with_cors(cors);
903 }
904
905 spawn_background_eviction_task(
906 Arc::clone(&store),
907 BACKGROUND_EVICTION_INTERVAL,
908 "embedded daemon",
909 );
910
911 let listener = TcpListener::bind(&opts.bind_address).await?;
912 let local_addr = listener.local_addr()?;
913 let actual_addr = format!("{}:{}", local_addr.ip(), local_addr.port());
914
915 let server_shutdown = Arc::new(Notify::new());
916 let server_shutdown_for_task = Arc::clone(&server_shutdown);
917 let server_join = tokio::spawn(async move {
918 if let Err(e) = server
919 .run_with_listener_until(listener, async move {
920 server_shutdown_for_task.notified().await;
921 })
922 .await
923 {
924 tracing::error!("Embedded daemon server error: {}", e);
925 }
926 });
927 let server_controller = Arc::new(EmbeddedServerController::new(server_shutdown, server_join));
928 background_services_controller.apply_config(&config).await?;
929 #[cfg(feature = "p2p")]
930 let daemon_controller = Arc::new(EmbeddedDaemonController::new(
931 server_controller,
932 fips_handle.clone(),
933 peer_router_controller.clone(),
934 Some(background_services_controller.clone()),
935 ));
936 #[cfg(not(feature = "p2p"))]
937 let daemon_controller = Arc::new(EmbeddedDaemonController::new(
938 server_controller,
939 fips_handle.clone(),
940 Some(background_services_controller.clone()),
941 ));
942
943 tracing::info!(
944 "Embedded daemon started on {}, identity {}",
945 actual_addr,
946 npub
947 );
948
949 Ok(EmbeddedDaemonInfo {
950 addr: actual_addr,
951 port: local_addr.port(),
952 npub,
953 store,
954 daemon_controller,
955 webrtc_state,
956 #[cfg(feature = "p2p")]
957 peer_router_controller,
958 background_services_controller: Some(background_services_controller),
959 })
960}
961
962#[cfg(test)]
963mod tests {
964 use super::EmbeddedBackgroundServicesController;
965 use crate::config::Config;
966
967 #[test]
968 fn mirror_publish_relays_orders_known_root_publish_relays_first() {
969 let relays = EmbeddedBackgroundServicesController::mirror_publish_relays(
970 &[
971 "wss://graph-relay.iris.to".to_string(),
972 "wss://relay.example".to_string(),
973 "wss://relay.primal.net".to_string(),
974 "wss://relay.damus.io".to_string(),
975 "wss://temp.iris.to".to_string(),
976 "wss://vault.iris.to".to_string(),
977 "wss://upload.iris.to/nostr".to_string(),
978 ],
979 "0.0.0.0:8080",
980 );
981 assert_eq!(
982 relays,
983 vec![
984 "wss://temp.iris.to".to_string(),
985 "wss://vault.iris.to".to_string(),
986 "wss://relay.damus.io".to_string(),
987 "wss://relay.example".to_string(),
988 "wss://relay.primal.net".to_string(),
989 ]
990 );
991 }
992
993 #[test]
994 fn mirror_publish_relays_do_not_add_non_active_publish_targets() {
995 let relays = EmbeddedBackgroundServicesController::mirror_publish_relays(
996 &[
997 "wss://graph-relay.iris.to".to_string(),
998 "wss://relay.example".to_string(),
999 ],
1000 "0.0.0.0:8080",
1001 );
1002 assert_eq!(relays, vec!["wss://relay.example".to_string()]);
1003 }
1004
1005 #[test]
1006 fn mirror_publish_relays_falls_back_to_active_relays_when_all_are_blocklisted() {
1007 let relays = EmbeddedBackgroundServicesController::mirror_publish_relays(
1008 &[
1009 "wss://graph-relay.iris.to".to_string(),
1010 "wss://upload.iris.to/nostr".to_string(),
1011 ],
1012 "0.0.0.0:8080",
1013 );
1014 assert_eq!(
1015 relays,
1016 vec![
1017 "wss://graph-relay.iris.to".to_string(),
1018 "wss://upload.iris.to/nostr".to_string(),
1019 ]
1020 );
1021 }
1022
1023 #[test]
1024 fn nostr_mirror_config_allows_disabling_full_note_paging() {
1025 let mut config = Config::default();
1026 config.nostr.full_text_note_history_max_relay_pages = 0;
1027
1028 let mirror_config = EmbeddedBackgroundServicesController::nostr_mirror_config(
1029 &config,
1030 &["wss://relay.example".to_string()],
1031 );
1032
1033 assert_eq!(mirror_config.full_text_note_history_max_relay_pages, 0);
1034
1035 config.nostr.full_text_note_history_max_relay_pages = 64;
1036 let mirror_config = EmbeddedBackgroundServicesController::nostr_mirror_config(
1037 &config,
1038 &["wss://relay.example".to_string()],
1039 );
1040
1041 assert_eq!(mirror_config.full_text_note_history_max_relay_pages, 64);
1042 }
1043
1044 #[test]
1045 fn nostr_mirror_config_can_limit_mirror_distance_independently() {
1046 let mut config = Config::default();
1047 config.nostr.social_graph_crawl_depth = 6;
1048 config.nostr.mirror_max_follow_distance = Some(2);
1049
1050 let mirror_config = EmbeddedBackgroundServicesController::nostr_mirror_config(
1051 &config,
1052 &["wss://relay.example".to_string()],
1053 );
1054
1055 assert_eq!(mirror_config.max_follow_distance, 2);
1056
1057 config.nostr.mirror_max_follow_distance = None;
1058 let mirror_config = EmbeddedBackgroundServicesController::nostr_mirror_config(
1059 &config,
1060 &["wss://relay.example".to_string()],
1061 );
1062
1063 assert_eq!(mirror_config.max_follow_distance, 6);
1064 }
1065}