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 archive_history_follow_distance: config.nostr.archive_history_follow_distance,
330 archive_history_max_relay_pages: config.nostr.archive_history_max_relay_pages,
331 ..crate::nostr_mirror::NostrMirrorConfig::default()
332 }
333 }
334
335 pub async fn apply_config(&self, config: &Config) -> Result<EmbeddedBackgroundServicesStatus> {
336 let mut runtime = self.runtime.lock().await;
337
338 Self::shutdown_crawler(&mut runtime.crawler).await;
339 Self::shutdown_mirror(&mut runtime.mirror).await;
340 Self::shutdown_sync(&mut runtime.sync).await;
341
342 if !config.server.mode.background_services_enabled() {
343 return Ok(runtime.status());
344 }
345
346 let active_relays = config.nostr.active_relays();
347
348 if config.nostr.enabled
349 && config.nostr.social_graph_crawl_depth > 0
350 && !active_relays.is_empty()
351 {
352 runtime.crawler = Some(socialgraph::crawler::spawn_social_graph_tasks(
353 self.graph_store.clone(),
354 self.keys.clone(),
355 active_relays.clone(),
356 config.nostr.social_graph_crawl_depth,
357 self.spambox.clone(),
358 self.data_dir.clone(),
359 ));
360
361 let service = Arc::new(
362 crate::nostr_mirror::BackgroundNostrMirror::new(
363 Self::nostr_mirror_config(config, &active_relays),
364 self.store.clone(),
365 self.graph_store_concrete.clone(),
366 Some(
367 nostr_sdk::Keys::parse(&self.keys.secret_key().to_bech32()?)
368 .context("Failed to parse keys for background nostr mirror")?,
369 ),
370 )
371 .await
372 .context("Failed to create background nostr mirror")?,
373 );
374 let service_for_task = service.clone();
375 let join = tokio::task::spawn_blocking(move || {
376 let runtime = tokio::runtime::Builder::new_current_thread()
377 .enable_all()
378 .build()
379 .expect("build background nostr mirror runtime");
380 runtime.block_on(async {
381 if let Err(err) = service_for_task.run().await {
382 tracing::error!("Background nostr mirror error: {:#}", err);
383 }
384 });
385 });
386 runtime.mirror = Some(BackgroundMirrorRuntime {
387 service,
388 join: Some(join),
389 });
390 }
391
392 if config.sync.enabled && !active_relays.is_empty() {
393 let has_pinned_refs = self
394 .store
395 .list_pinned_refs()
396 .map(|refs| !refs.is_empty())
397 .unwrap_or(false);
398 let has_tracked_authors = self
399 .store
400 .list_tracked_authors()
401 .map(|authors| !authors.is_empty())
402 .unwrap_or(false);
403 let should_sync = config.sync.sync_own
404 || config.sync.sync_followed
405 || has_pinned_refs
406 || has_tracked_authors;
407 if !should_sync {
408 return Ok(runtime.status());
409 }
410
411 let sync_config = crate::sync::SyncConfig {
412 sync_own: config.sync.sync_own,
413 sync_followed: config.sync.sync_followed,
414 relays: active_relays,
415 max_concurrent: config.sync.max_concurrent,
416 webrtc_timeout_ms: config.sync.webrtc_timeout_ms,
417 blossom_timeout_ms: config.sync.blossom_timeout_ms,
418 };
419
420 let sync_keys = nostr_sdk::Keys::parse(&self.keys.secret_key().to_bech32()?)
421 .context("Failed to parse keys for sync")?;
422 let service = Arc::new(
423 crate::sync::BackgroundSync::new(
424 sync_config,
425 self.store.clone(),
426 sync_keys,
427 self.webrtc_state.clone(),
428 )
429 .await
430 .context("Failed to create background sync service")?,
431 );
432 let contacts_file = self.data_dir.join("contacts.json");
433 let service_for_task = service.clone();
434 let join = tokio::spawn(async move {
435 if let Err(err) = service_for_task.run(contacts_file).await {
436 tracing::error!("Background sync error: {}", err);
437 }
438 });
439 runtime.sync = Some(BackgroundSyncRuntime {
440 service,
441 join: Some(join),
442 });
443 }
444
445 Ok(runtime.status())
446 }
447}
448
449#[cfg(feature = "p2p")]
450pub struct EmbeddedPeerRouterController {
451 keys: Keys,
452 data_dir: PathBuf,
453 state: Arc<WebRTCState>,
454 store: Arc<dyn ContentStore>,
455 peer_classifier: PeerClassifier,
456 nostr_relay: Arc<NostrRelay>,
457 runtime: Mutex<Option<PeerRouterRuntime>>,
458}
459
460#[cfg(feature = "p2p")]
461impl EmbeddedPeerRouterController {
462 pub fn new(
463 keys: Keys,
464 data_dir: PathBuf,
465 state: Arc<WebRTCState>,
466 store: Arc<dyn ContentStore>,
467 peer_classifier: PeerClassifier,
468 nostr_relay: Arc<NostrRelay>,
469 ) -> Self {
470 Self {
471 keys,
472 data_dir,
473 state,
474 store,
475 peer_classifier,
476 nostr_relay,
477 runtime: Mutex::new(None),
478 }
479 }
480
481 pub fn state(&self) -> Arc<WebRTCState> {
482 self.state.clone()
483 }
484
485 pub async fn apply_config(&self, config: &Config) -> Result<bool> {
486 let mut runtime = self.runtime.lock().await;
487 if let Some(runtime_handle) = runtime.take() {
488 if let Err(err) =
489 crate::p2p_common::persist_peer_state(&self.data_dir, &self.state).await
490 {
491 tracing::warn!("Failed to persist mesh peer state before router restart: {err:#}");
492 }
493 let _ = runtime_handle.shutdown.send(true);
494 runtime_handle.peer_state_persist.abort();
495 let mut join = runtime_handle.join;
496 match tokio::time::timeout(std::time::Duration::from_secs(3), &mut join).await {
497 Ok(Ok(())) => {}
498 Ok(Err(err)) => {
499 tracing::warn!("Peer router task ended with join error: {}", err);
500 }
501 Err(_) => {
502 tracing::warn!("Timed out waiting for peer router shutdown");
503 join.abort();
504 }
505 }
506 }
507
508 self.state.reset_runtime_state().await;
509 if let Err(err) = crate::p2p_common::load_peer_state(&self.data_dir, &self.state).await {
510 tracing::warn!("Failed to load persisted mesh peer state: {err:#}");
511 }
512
513 if !crate::p2p_common::peer_router_enabled(config) {
514 return Ok(false);
515 }
516
517 let webrtc_config = crate::p2p_common::default_webrtc_config(config);
518 let mut manager = if config.server.mode.hash_get_enabled() {
519 WebRTCManager::new_with_state_and_store_and_classifier(
520 self.keys.clone(),
521 webrtc_config,
522 self.state.clone(),
523 self.store.clone(),
524 self.peer_classifier.clone(),
525 )
526 } else {
527 let mut manager =
528 WebRTCManager::new_with_state(self.keys.clone(), webrtc_config, self.state.clone());
529 manager.set_peer_classifier(self.peer_classifier.clone());
530 manager
531 };
532 manager
533 .set_nostr_relay(self.nostr_relay.clone() as hashtree_network::SharedMeshRelayClient);
534 let shutdown = manager.shutdown_signal();
535 let join = tokio::spawn(async move {
536 if let Err(err) = manager.run().await {
537 tracing::error!("Peer router error: {}", err);
538 }
539 });
540 let peer_state_persist = crate::p2p_common::spawn_peer_state_persist_task(
541 self.data_dir.clone(),
542 self.state.clone(),
543 );
544 *runtime = Some(PeerRouterRuntime {
545 shutdown,
546 join,
547 peer_state_persist,
548 });
549 Ok(true)
550 }
551
552 pub async fn shutdown(&self) {
553 let mut runtime = self.runtime.lock().await;
554 let Some(runtime_handle) = runtime.take() else {
555 return;
556 };
557
558 if let Err(err) = crate::p2p_common::persist_peer_state(&self.data_dir, &self.state).await {
559 tracing::warn!("Failed to persist mesh peer state during router shutdown: {err:#}");
560 }
561 let _ = runtime_handle.shutdown.send(true);
562 runtime_handle.peer_state_persist.abort();
563 let mut join = runtime_handle.join;
564 match tokio::time::timeout(std::time::Duration::from_secs(3), &mut join).await {
565 Ok(Ok(())) => {}
566 Ok(Err(err)) => tracing::warn!("Peer router task ended with join error: {}", err),
567 Err(_) => {
568 tracing::warn!("Timed out waiting for peer router shutdown");
569 join.abort();
570 }
571 }
572
573 self.state.reset_runtime_state().await;
574 }
575}
576
577pub struct EmbeddedDaemonController {
578 server_controller: Arc<EmbeddedServerController>,
579 fips_handle: Option<Arc<crate::fips_transport::DaemonFipsHandle>>,
580 #[cfg(feature = "experimental-decentralized-pubsub")]
581 nostr_pubsub_handle: Option<Arc<crate::fips_transport::DaemonNostrPubsubHandle>>,
582 #[cfg(feature = "p2p")]
583 peer_router_controller: Option<Arc<EmbeddedPeerRouterController>>,
584 background_services_controller: Option<Arc<EmbeddedBackgroundServicesController>>,
585}
586
587impl EmbeddedDaemonController {
588 #[cfg(feature = "p2p")]
589 pub fn new(
590 server_controller: Arc<EmbeddedServerController>,
591 fips_handle: Option<Arc<crate::fips_transport::DaemonFipsHandle>>,
592 #[cfg(feature = "experimental-decentralized-pubsub")] nostr_pubsub_handle: Option<
593 Arc<crate::fips_transport::DaemonNostrPubsubHandle>,
594 >,
595 peer_router_controller: Option<Arc<EmbeddedPeerRouterController>>,
596 background_services_controller: Option<Arc<EmbeddedBackgroundServicesController>>,
597 ) -> Self {
598 Self {
599 server_controller,
600 fips_handle,
601 #[cfg(feature = "experimental-decentralized-pubsub")]
602 nostr_pubsub_handle,
603 #[cfg(feature = "p2p")]
604 peer_router_controller,
605 background_services_controller,
606 }
607 }
608
609 #[cfg(not(feature = "p2p"))]
610 pub fn new(
611 server_controller: Arc<EmbeddedServerController>,
612 fips_handle: Option<Arc<crate::fips_transport::DaemonFipsHandle>>,
613 #[cfg(feature = "experimental-decentralized-pubsub")] nostr_pubsub_handle: Option<
614 Arc<crate::fips_transport::DaemonNostrPubsubHandle>,
615 >,
616 background_services_controller: Option<Arc<EmbeddedBackgroundServicesController>>,
617 ) -> Self {
618 Self {
619 server_controller,
620 fips_handle,
621 #[cfg(feature = "experimental-decentralized-pubsub")]
622 nostr_pubsub_handle,
623 background_services_controller,
624 }
625 }
626
627 pub async fn shutdown(&self) {
628 self.server_controller.shutdown().await;
629 #[cfg(feature = "experimental-decentralized-pubsub")]
630 if let Some(handle) = self.nostr_pubsub_handle.as_ref() {
631 handle.shutdown();
632 }
633 if let Some(handle) = self.fips_handle.as_ref() {
634 handle.shutdown().await;
635 }
636 if let Some(controller) = self.background_services_controller.as_ref() {
637 controller.shutdown().await;
638 }
639 #[cfg(feature = "p2p")]
640 if let Some(controller) = self.peer_router_controller.as_ref() {
641 controller.shutdown().await;
642 }
643 }
644}
645
646pub struct EmbeddedDaemonOptions {
647 pub config: Config,
648 pub data_dir: PathBuf,
649 pub config_dir: Option<PathBuf>,
650 pub bind_address: String,
651 pub relays: Option<Vec<String>>,
652 pub initial_tree_roots: Vec<(String, Cid)>,
653 pub extra_routes: Option<Router<AppState>>,
654 pub cors: Option<CorsLayer>,
655}
656
657pub struct EmbeddedDaemonInfo {
658 pub addr: String,
659 pub port: u16,
660 pub npub: String,
661 pub store: Arc<HashtreeStore>,
662 pub daemon_controller: Arc<EmbeddedDaemonController>,
663 #[allow(dead_code)]
664 pub webrtc_state: Option<Arc<WebRTCState>>,
665 #[cfg(feature = "p2p")]
666 #[allow(dead_code)]
667 pub peer_router_controller: Option<Arc<EmbeddedPeerRouterController>>,
668 #[allow(dead_code)]
669 pub background_services_controller: Option<Arc<EmbeddedBackgroundServicesController>>,
670}
671
672pub async fn start_embedded(opts: EmbeddedDaemonOptions) -> Result<EmbeddedDaemonInfo> {
673 let _ = rustls::crypto::ring::default_provider().install_default();
674
675 let mut config = opts.config;
676 config.server.bind_address = opts.bind_address.clone();
677 if let Some(relays) = opts.relays {
678 config.nostr.relays = relays;
679 config.nostr.enabled = embedded_nostr_enabled_after_relay_override(&config);
680 }
681
682 let max_size_bytes = config.storage.max_size_gb * 1024 * 1024 * 1024;
683 let nostr_db_max_bytes = config
684 .nostr
685 .db_max_size_gb
686 .saturating_mul(1024 * 1024 * 1024);
687 let spambox_db_max_bytes = config
688 .nostr
689 .spambox_max_size_gb
690 .saturating_mul(1024 * 1024 * 1024);
691
692 let store = Arc::new(HashtreeStore::with_embedded_options(
693 &opts.data_dir,
694 config.storage.s3.as_ref(),
695 max_size_bytes,
696 )?);
697
698 let (keys, _was_generated) = if let Some(config_dir) = opts.config_dir.as_ref() {
699 ensure_keys_in(config_dir, Some(&opts.data_dir), Some(&config))?
700 } else {
701 ensure_keys()?
702 };
703 let pk_bytes = pubkey_bytes(&keys);
704 let npub = keys
705 .public_key()
706 .to_bech32()
707 .context("Failed to encode npub")?;
708
709 let mut allowed_pubkeys: HashSet<String> = HashSet::new();
710 allowed_pubkeys.insert(hex::encode(pk_bytes));
711 for npub_str in &config.nostr.allowed_npubs {
712 if let Ok(pk) = parse_npub(npub_str) {
713 allowed_pubkeys.insert(hex::encode(pk));
714 } else {
715 tracing::warn!("Invalid npub in allowed_npubs: {}", npub_str);
716 }
717 }
718
719 let graph_store = socialgraph::open_embedded_social_graph_store_with_storage(
720 &opts.data_dir,
721 store.store_arc(),
722 Some(nostr_db_max_bytes),
723 )
724 .context("Failed to initialize social graph store")?;
725 graph_store.set_profile_index_overmute_threshold(config.nostr.overmute_threshold);
726
727 let social_graph_root_bytes = if let Some(ref root_npub) = config.nostr.socialgraph_root {
728 parse_npub(root_npub).unwrap_or(pk_bytes)
729 } else {
730 pk_bytes
731 };
732 socialgraph::set_social_graph_root(&graph_store, &social_graph_root_bytes);
733 socialgraph::sync_local_list_files_force(graph_store.as_ref(), &opts.data_dir, &keys)
734 .context("Failed to sync local social graph lists")?;
735 let fips_peer_ids = crate::fips_transport::fips_peer_ids_from_pubkeys(
736 socialgraph::get_follows(graph_store.as_ref(), &pk_bytes),
737 );
738 let social_graph_store: Arc<dyn socialgraph::SocialGraphBackend> = graph_store.clone();
739
740 let social_graph = Arc::new(socialgraph::SocialGraphAccessControl::new(
741 Arc::clone(&social_graph_store),
742 config.nostr.max_write_distance,
743 allowed_pubkeys.clone(),
744 ));
745
746 let nostr_relay_config = NostrRelayConfig {
747 spambox_db_max_bytes,
748 ..Default::default()
749 };
750 let nostr_relay = if config.nostr.enabled {
751 let mut public_event_pubkeys = HashSet::new();
752 public_event_pubkeys.insert(hex::encode(pk_bytes));
753 Some(Arc::new(
754 NostrRelay::new(
755 Arc::clone(&social_graph_store),
756 opts.data_dir.clone(),
757 public_event_pubkeys,
758 Some(social_graph.clone()),
759 nostr_relay_config,
760 )
761 .map(|relay| {
762 relay.with_historical_nostr_index(store.store_arc(), opts.data_dir.clone())
763 })
764 .context("Failed to initialize Nostr relay")?,
765 ))
766 } else {
767 None
768 };
769
770 let crawler_spambox = if config.nostr.enabled && spambox_db_max_bytes != 0 {
771 let spam_dir = opts.data_dir.join("socialgraph_spambox");
772 match socialgraph::open_embedded_social_graph_store_at_path(
773 &spam_dir,
774 Some(spambox_db_max_bytes),
775 ) {
776 Ok(store) => Some(store),
777 Err(err) => {
778 tracing::warn!("Failed to open social graph spambox for crawler: {}", err);
779 None
780 }
781 }
782 } else {
783 None
784 };
785 let crawler_spambox_backend = crawler_spambox
786 .clone()
787 .map(|store| store as Arc<dyn socialgraph::SocialGraphBackend>);
788
789 #[cfg(feature = "p2p")]
790 let (webrtc_state, peer_router_controller): (
791 Option<Arc<WebRTCState>>,
792 Option<Arc<EmbeddedPeerRouterController>>,
793 ) = if let Some(nostr_relay) = nostr_relay.clone() {
794 let router_config = crate::p2p_common::default_webrtc_config(&config);
795 let peer_classifier = crate::p2p_common::build_peer_classifier(
796 opts.data_dir.clone(),
797 Arc::clone(&social_graph_store),
798 );
799 let cashu_payment_client =
800 if config.cashu.default_mint.is_some() || !config.cashu.accepted_mints.is_empty() {
801 match crate::cashu_helper::CashuHelperClient::discover(opts.data_dir.clone()) {
802 Ok(client) => {
803 Some(Arc::new(client) as Arc<dyn crate::cashu_helper::CashuPaymentClient>)
804 }
805 Err(err) => {
806 tracing::warn!(
807 "Cashu settlement helper unavailable; paid retrieval stays disabled: {}",
808 err
809 );
810 None
811 }
812 }
813 } else {
814 None
815 };
816 let cashu_mint_metadata =
817 if config.cashu.default_mint.is_some() || !config.cashu.accepted_mints.is_empty() {
818 let metadata_path = crate::webrtc::cashu_mint_metadata_path(&opts.data_dir);
819 match crate::webrtc::CashuMintMetadataStore::load(metadata_path) {
820 Ok(store) => Some(store),
821 Err(err) => {
822 tracing::warn!(
823 "Failed to load Cashu mint metadata; falling back to in-memory state: {}",
824 err
825 );
826 Some(crate::webrtc::CashuMintMetadataStore::in_memory())
827 }
828 }
829 } else {
830 None
831 };
832
833 let state = Arc::new(WebRTCState::new_with_routing_and_cashu(
834 router_config.request_selection_strategy,
835 router_config.request_fairness_enabled,
836 router_config.request_dispatch,
837 std::time::Duration::from_millis(router_config.message_timeout_ms),
838 crate::webrtc::CashuRoutingConfig::from(&config.cashu),
839 cashu_payment_client,
840 cashu_mint_metadata,
841 ));
842 let controller = Arc::new(EmbeddedPeerRouterController::new(
843 keys.clone(),
844 opts.data_dir.clone(),
845 state.clone(),
846 Arc::clone(&store) as Arc<dyn ContentStore>,
847 peer_classifier,
848 nostr_relay.clone(),
849 ));
850 controller.apply_config(&config).await?;
851 (Some(state), Some(controller))
852 } else {
853 (None, None)
854 };
855
856 #[cfg(not(feature = "p2p"))]
857 let webrtc_state: Option<Arc<crate::webrtc::WebRTCState>> = None;
858
859 let background_services_controller = Arc::new(EmbeddedBackgroundServicesController::new(
860 keys.clone(),
861 opts.data_dir.clone(),
862 Arc::clone(&store),
863 graph_store.clone(),
864 Arc::clone(&social_graph_store),
865 crawler_spambox_backend,
866 webrtc_state.clone(),
867 ));
868
869 let upstream_blossom = config.blossom.upstream_read_servers(&opts.bind_address);
870 let blossom_replica_queue_bytes = crate::server::bounded_upload_queue_bytes(
871 config
872 .blossom
873 .replicate_queue_mb
874 .saturating_mul(1024 * 1024),
875 );
876 let active_nostr_relays = config.nostr.active_relays();
877 let fips_handle = crate::fips_transport::start_daemon_fips_transport(
878 &config,
879 &keys,
880 Arc::clone(&store),
881 fips_peer_ids,
882 )
883 .await?
884 .map(Arc::new);
885 let nostr_provider =
886 crate::fips_transport::start_daemon_nostr_provider(&config, fips_handle.as_deref()).await?;
887 #[cfg(feature = "experimental-decentralized-pubsub")]
888 let nostr_pubsub_handle = crate::fips_transport::start_daemon_nostr_pubsub(
889 &config,
890 fips_handle.as_deref(),
891 nostr_relay.clone(),
892 )
893 .await?;
894
895 let mut server = HashtreeServer::new(Arc::clone(&store), opts.bind_address.clone())
896 .with_server_mode(config.server.mode)
897 .with_hash_get_enabled(config.server.mode.hash_get_enabled())
898 .with_fetch_from_fips_peers(config.server.fetch_from_fips_peers)
899 .with_allowed_pubkeys(allowed_pubkeys.clone())
900 .with_max_upload_bytes((config.blossom.max_upload_mb as usize) * 1024 * 1024)
901 .with_public_writes(config.server.public_writes)
902 .with_public_plaintext_reads(config.server.public_plaintext_reads)
903 .with_require_random_untrusted_ingest(config.blossom.require_random_untrusted_ingest)
904 .with_optimistic_blossom_uploads(config.blossom.optimistic_uploads)
905 .with_upstream_blossom(upstream_blossom)
906 .with_blossom_upload_replicas(
907 config.blossom.replicate_servers.clone(),
908 blossom_replica_queue_bytes,
909 keys.clone(),
910 )
911 .with_nostr_relay_urls(active_nostr_relays)
912 .with_cached_tree_roots(opts.initial_tree_roots)
913 .with_social_graph(social_graph)
914 .with_socialgraph_snapshot(
915 Arc::clone(&social_graph_store),
916 social_graph_root_bytes,
917 config.server.socialgraph_snapshot_public,
918 );
919 if let Some(nostr_relay) = nostr_relay {
920 server = server.with_nostr_relay(nostr_relay);
921 }
922 if let Some(provider) = nostr_provider {
923 server = server.with_nostr_provider(provider);
924 }
925
926 if crate::p2p_common::peer_router_enabled(&config) {
927 if let Some(ref state) = webrtc_state {
928 server = server.with_webrtc_peers(state.clone());
929 }
930 }
931 if let Some(ref fips_handle) = fips_handle {
932 server = server
933 .with_fips_endpoint(fips_handle.endpoint.clone())
934 .with_fips_blob_resolver(fips_handle.blob_resolver.clone());
935 }
936
937 if let Some(extra) = opts.extra_routes {
938 server = server.with_extra_routes(extra);
939 }
940 if let Some(cors) = opts.cors {
941 server = server.with_cors(cors);
942 }
943
944 spawn_background_eviction_task(
945 Arc::clone(&store),
946 BACKGROUND_EVICTION_INTERVAL,
947 "embedded daemon",
948 );
949
950 let listener = TcpListener::bind(&opts.bind_address).await?;
951 let local_addr = listener.local_addr()?;
952 let actual_addr = format!("{}:{}", local_addr.ip(), local_addr.port());
953
954 let server_shutdown = Arc::new(Notify::new());
955 let server_shutdown_for_task = Arc::clone(&server_shutdown);
956 let server_join = tokio::spawn(async move {
957 if let Err(e) = server
958 .run_with_listener_until(listener, async move {
959 server_shutdown_for_task.notified().await;
960 })
961 .await
962 {
963 tracing::error!("Embedded daemon server error: {}", e);
964 }
965 });
966 let server_controller = Arc::new(EmbeddedServerController::new(server_shutdown, server_join));
967 background_services_controller.apply_config(&config).await?;
968 #[cfg(feature = "p2p")]
969 let daemon_controller = Arc::new(EmbeddedDaemonController::new(
970 server_controller,
971 fips_handle.clone(),
972 #[cfg(feature = "experimental-decentralized-pubsub")]
973 nostr_pubsub_handle.clone(),
974 peer_router_controller.clone(),
975 Some(background_services_controller.clone()),
976 ));
977 #[cfg(not(feature = "p2p"))]
978 let daemon_controller = Arc::new(EmbeddedDaemonController::new(
979 server_controller,
980 fips_handle.clone(),
981 #[cfg(feature = "experimental-decentralized-pubsub")]
982 nostr_pubsub_handle.clone(),
983 Some(background_services_controller.clone()),
984 ));
985
986 tracing::info!(
987 "Embedded daemon started on {}, identity {}",
988 actual_addr,
989 npub
990 );
991
992 Ok(EmbeddedDaemonInfo {
993 addr: actual_addr,
994 port: local_addr.port(),
995 npub,
996 store,
997 daemon_controller,
998 webrtc_state,
999 #[cfg(feature = "p2p")]
1000 peer_router_controller,
1001 background_services_controller: Some(background_services_controller),
1002 })
1003}
1004
1005fn embedded_nostr_enabled_after_relay_override(config: &Config) -> bool {
1006 config.nostr.decentralized_pubsub || !config.nostr.relays.is_empty()
1007}
1008
1009#[cfg(test)]
1010mod tests {
1011 use super::{
1012 embedded_nostr_enabled_after_relay_override, EmbeddedBackgroundServicesController,
1013 };
1014 use crate::config::Config;
1015
1016 #[test]
1017 fn mirror_publish_relays_orders_known_root_publish_relays_first() {
1018 let relays = EmbeddedBackgroundServicesController::mirror_publish_relays(
1019 &[
1020 "wss://graph-relay.iris.to".to_string(),
1021 "wss://relay.example".to_string(),
1022 "wss://relay.primal.net".to_string(),
1023 "wss://relay.damus.io".to_string(),
1024 "wss://temp.iris.to".to_string(),
1025 "wss://vault.iris.to".to_string(),
1026 "wss://upload.iris.to/nostr".to_string(),
1027 ],
1028 "0.0.0.0:8080",
1029 );
1030 assert_eq!(
1031 relays,
1032 vec![
1033 "wss://temp.iris.to".to_string(),
1034 "wss://vault.iris.to".to_string(),
1035 "wss://relay.damus.io".to_string(),
1036 "wss://relay.example".to_string(),
1037 "wss://relay.primal.net".to_string(),
1038 ]
1039 );
1040 }
1041
1042 #[test]
1043 fn mirror_publish_relays_do_not_add_non_active_publish_targets() {
1044 let relays = EmbeddedBackgroundServicesController::mirror_publish_relays(
1045 &[
1046 "wss://graph-relay.iris.to".to_string(),
1047 "wss://relay.example".to_string(),
1048 ],
1049 "0.0.0.0:8080",
1050 );
1051 assert_eq!(relays, vec!["wss://relay.example".to_string()]);
1052 }
1053
1054 #[test]
1055 fn mirror_publish_relays_falls_back_to_active_relays_when_all_are_blocklisted() {
1056 let relays = EmbeddedBackgroundServicesController::mirror_publish_relays(
1057 &[
1058 "wss://graph-relay.iris.to".to_string(),
1059 "wss://upload.iris.to/nostr".to_string(),
1060 ],
1061 "0.0.0.0:8080",
1062 );
1063 assert_eq!(
1064 relays,
1065 vec![
1066 "wss://graph-relay.iris.to".to_string(),
1067 "wss://upload.iris.to/nostr".to_string(),
1068 ]
1069 );
1070 }
1071
1072 #[test]
1073 fn nostr_mirror_config_maps_legacy_and_complete_archive_settings() {
1074 let mut config = Config::default();
1075 config.nostr.full_text_note_history_max_relay_pages = 0;
1076 config.nostr.archive_history_follow_distance = None;
1077 config.nostr.archive_history_max_relay_pages = 0;
1078
1079 let mirror_config = EmbeddedBackgroundServicesController::nostr_mirror_config(
1080 &config,
1081 &["wss://relay.example".to_string()],
1082 );
1083
1084 assert_eq!(mirror_config.full_text_note_history_max_relay_pages, 0);
1085 assert_eq!(mirror_config.archive_history_follow_distance, None);
1086 assert_eq!(mirror_config.archive_history_max_relay_pages, 0);
1087
1088 config.nostr.full_text_note_history_max_relay_pages = 64;
1089 config.nostr.archive_history_follow_distance = Some(1);
1090 config.nostr.archive_history_max_relay_pages = 32;
1091 let mirror_config = EmbeddedBackgroundServicesController::nostr_mirror_config(
1092 &config,
1093 &["wss://relay.example".to_string()],
1094 );
1095
1096 assert_eq!(mirror_config.full_text_note_history_max_relay_pages, 64);
1097 assert_eq!(mirror_config.archive_history_follow_distance, Some(1));
1098 assert_eq!(mirror_config.archive_history_max_relay_pages, 32);
1099 }
1100
1101 #[test]
1102 fn nostr_mirror_config_can_limit_mirror_distance_independently() {
1103 let mut config = Config::default();
1104 config.nostr.social_graph_crawl_depth = 6;
1105 config.nostr.mirror_max_follow_distance = Some(2);
1106
1107 let mirror_config = EmbeddedBackgroundServicesController::nostr_mirror_config(
1108 &config,
1109 &["wss://relay.example".to_string()],
1110 );
1111
1112 assert_eq!(mirror_config.max_follow_distance, 2);
1113
1114 config.nostr.mirror_max_follow_distance = None;
1115 let mirror_config = EmbeddedBackgroundServicesController::nostr_mirror_config(
1116 &config,
1117 &["wss://relay.example".to_string()],
1118 );
1119
1120 assert_eq!(mirror_config.max_follow_distance, 6);
1121 }
1122
1123 #[test]
1124 fn embedded_empty_relays_keep_nostr_enabled_for_decentralized_pubsub() {
1125 let mut config = Config::default();
1126 config.nostr.relays = Vec::new();
1127 config.nostr.decentralized_pubsub = false;
1128 assert!(!embedded_nostr_enabled_after_relay_override(&config));
1129
1130 config.nostr.decentralized_pubsub = true;
1131 assert!(embedded_nostr_enabled_after_relay_override(&config));
1132
1133 config.nostr.decentralized_pubsub = false;
1134 config.nostr.relays = vec!["wss://relay.example".to_string()];
1135 assert!(embedded_nostr_enabled_after_relay_override(&config));
1136 }
1137}