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