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