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