1use std::collections::HashMap;
2use std::collections::{HashSet, VecDeque};
3use std::io::ErrorKind;
4use std::path::Path;
5use std::path::PathBuf;
6use std::sync::{
7 atomic::{AtomicU64, Ordering},
8 Arc,
9};
10use std::time::{Duration, Instant};
11
12use tokio::sync::{mpsc, Mutex, Semaphore};
13
14use nostr::{ClientMessage as NostrClientMessage, JsonUtil, RelayMessage as NostrRelayMessage};
15use nostr::{Event, EventId, Filter as NostrFilter, SubscriptionId};
16
17use crate::socialgraph;
18
19const BLUETOOTH_EVENT_LOG_CAPACITY: usize = 100;
20const MAX_CONCURRENT_NOSTR_STORE_BLOCKING_TASKS: usize = 4;
21
22#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
23pub struct BluetoothReceivedEventRecord {
24 pub event_id: String,
25 pub pubkey: String,
26 pub kind: u32,
27 pub created_at: u64,
28 pub received_at: u64,
29 pub peer_id: Option<String>,
30 pub cid_values: Vec<String>,
31}
32
33#[derive(Debug, Clone)]
34pub struct NostrRelayConfig {
35 pub spambox_db_max_bytes: u64,
36 pub max_query_limit: usize,
37 pub max_subs_per_client: usize,
38 pub max_filters_per_sub: usize,
39 pub spambox_max_events_per_min: u32,
40 pub spambox_max_reqs_per_min: u32,
41}
42
43impl Default for NostrRelayConfig {
44 fn default() -> Self {
45 Self {
46 spambox_db_max_bytes: 1024 * 1024 * 1024,
47 max_query_limit: 200,
48 max_subs_per_client: 64,
49 max_filters_per_sub: 32,
50 spambox_max_events_per_min: 120,
51 spambox_max_reqs_per_min: 120,
52 }
53 }
54}
55
56mod imp {
57 use super::*;
58 use anyhow::Result;
59
60 use crate::diagnostics::{
61 nostr_filter_summary, nostr_filters_summary, process_memory_snapshot,
62 trim_process_allocations,
63 };
64 use crate::socialgraph::{EventStorageClass, SocialGraphAccessControl, SocialGraphBackend};
65 use crate::storage::StorageRouter;
66 use hashtree_core::{nhash_decode, nhash_encode_full, Cid, NHashData};
67 use hashtree_nostr::{
68 is_parameterized_replaceable_kind, is_replaceable_kind, NostrEventStore, VerifiedEvent,
69 VerifiedStoredNostrEvent,
70 };
71 use tracing::{info, warn};
72
73 const NOSTR_INDEX_DIR: &str = "nostr-index";
74 const NOSTR_INDEX_LATEST_ROOT_FILE: &str = "latest-root.txt";
75 const NOSTR_INDEX_CHECKPOINT_ROOT_FILE: &str = "checkpoint-root.txt";
76
77 fn prefers_trusted_only(filter: &NostrFilter) -> bool {
78 let Some(kinds) = filter.kinds.as_ref() else {
79 return false;
80 };
81 if kinds.len() != 1 {
82 return false;
83 }
84
85 let kind = kinds.iter().next().expect("checked single kind").as_u16() as u32;
86 let has_authors = filter
87 .authors
88 .as_ref()
89 .is_some_and(|authors| !authors.is_empty());
90 if !has_authors {
91 return false;
92 }
93
94 if is_replaceable_kind(kind) {
95 return true;
96 }
97
98 if is_parameterized_replaceable_kind(kind) {
99 let d_tag = nostr::SingleLetterTag::lowercase(nostr::Alphabet::D);
100 return filter
101 .generic_tags
102 .get(&d_tag)
103 .is_some_and(|values| !values.is_empty());
104 }
105
106 false
107 }
108
109 struct NostrStore {
110 store: Arc<dyn SocialGraphBackend>,
111 blocking_permits: Arc<Semaphore>,
112 }
113
114 impl NostrStore {
115 fn new(store: Arc<dyn SocialGraphBackend>) -> Self {
116 Self {
117 store,
118 blocking_permits: Arc::new(Semaphore::new(
119 MAX_CONCURRENT_NOSTR_STORE_BLOCKING_TASKS,
120 )),
121 }
122 }
123
124 async fn ingest(&self, event: Event) -> Result<()> {
125 let store = Arc::clone(&self.store);
126 let _permit = self
127 .blocking_permits
128 .clone()
129 .acquire_owned()
130 .await
131 .map_err(|err| anyhow::anyhow!("trusted nostr store closed: {err}"))?;
132 tokio::task::spawn_blocking(move || {
133 crate::socialgraph::ingest_parsed_event(store.as_ref(), &event)
134 })
135 .await
136 .map_err(|err| anyhow::anyhow!("trusted nostr store ingest task failed: {err}"))?
137 }
138
139 async fn ingest_with_storage_class(
140 &self,
141 event: Event,
142 storage_class: EventStorageClass,
143 ) -> Result<()> {
144 let store = Arc::clone(&self.store);
145 let _permit = self
146 .blocking_permits
147 .clone()
148 .acquire_owned()
149 .await
150 .map_err(|err| anyhow::anyhow!("trusted nostr store closed: {err}"))?;
151 tokio::task::spawn_blocking(move || {
152 crate::socialgraph::ingest_parsed_event_with_storage_class(
153 store.as_ref(),
154 &event,
155 storage_class,
156 )
157 })
158 .await
159 .map_err(|err| anyhow::anyhow!("trusted nostr store ingest task failed: {err}"))?
160 }
161
162 async fn query(&self, filter: NostrFilter, limit: usize) -> Vec<Event> {
163 let store = Arc::clone(&self.store);
164 let filter_summary = nostr_filter_summary(&filter);
165 let memory_before = process_memory_snapshot();
166 let started = Instant::now();
167 let Ok(_permit) = self.blocking_permits.clone().acquire_owned().await else {
168 warn!("trusted nostr store query skipped: blocking semaphore closed");
169 return Vec::new();
170 };
171 let result = tokio::task::spawn_blocking(move || {
172 crate::socialgraph::query_events(store.as_ref(), &filter, limit)
173 })
174 .await;
175 match result {
176 Ok(events) => {
177 info!(
178 target: "hashtree_cli::nostr_relay::query",
179 limit,
180 events = events.len(),
181 elapsed_ms = started.elapsed().as_millis() as u64,
182 filter = %filter_summary,
183 memory_before = ?memory_before,
184 memory_after = ?process_memory_snapshot(),
185 "trusted nostr store query completed",
186 );
187 events
188 }
189 Err(err) => {
190 warn!("trusted nostr store query task failed: {}", err);
191 Vec::new()
192 }
193 }
194 }
195 }
196
197 struct HistoricalNostrIndex {
198 store: Arc<StorageRouter>,
199 latest_root_path: PathBuf,
200 checkpoint_root_path: PathBuf,
201 blocking_permits: Arc<Semaphore>,
202 }
203
204 impl HistoricalNostrIndex {
205 fn new(store: Arc<StorageRouter>, data_dir: PathBuf) -> Self {
206 let index_dir = data_dir.join(NOSTR_INDEX_DIR);
207 Self {
208 store,
209 latest_root_path: index_dir.join(NOSTR_INDEX_LATEST_ROOT_FILE),
210 checkpoint_root_path: index_dir.join(NOSTR_INDEX_CHECKPOINT_ROOT_FILE),
211 blocking_permits: Arc::new(Semaphore::new(
212 MAX_CONCURRENT_NOSTR_STORE_BLOCKING_TASKS,
213 )),
214 }
215 }
216
217 async fn ingest(&self, event: Event) -> Result<()> {
218 let root = self.load_existing_root().await?;
219 let store = Arc::clone(&self.store);
220 let Ok(_permit) = self.blocking_permits.clone().acquire_owned().await else {
221 anyhow::bail!("historical nostr index ingest skipped: blocking semaphore closed");
222 };
223 let next_root = tokio::task::spawn_blocking(move || {
224 let runtime = tokio::runtime::Builder::new_current_thread()
225 .enable_all()
226 .build()?;
227 runtime.block_on(async move {
228 let event_store = NostrEventStore::new(store);
229 let stored = VerifiedEvent::try_from(event)?
230 .to_stored_event()
231 .into_stored();
232 event_store
233 .build(root.as_ref(), vec![stored])
234 .await?
235 .or(root)
236 .ok_or_else(|| {
237 anyhow::anyhow!("historical nostr index ingest did not produce a root")
238 })
239 })
240 })
241 .await
242 .map_err(|err| anyhow::anyhow!("historical nostr index ingest task failed: {err}"))??;
243 self.persist_latest_root(&next_root).await
244 }
245
246 async fn query(&self, filter: &NostrFilter, limit: usize) -> Vec<Event> {
247 if limit == 0 {
248 return Vec::new();
249 }
250
251 let root = match self.load_existing_root().await {
252 Ok(Some(root)) => root,
253 Ok(None) => return Vec::new(),
254 Err(err) => {
255 warn!("historical nostr index root load failed: {}", err);
256 return Vec::new();
257 }
258 };
259
260 let filter_summary = nostr_filter_summary(filter);
261 let memory_before = process_memory_snapshot();
262 let started = Instant::now();
263 let store = Arc::clone(&self.store);
264 let filter = filter.clone();
265 let Ok(_permit) = self.blocking_permits.clone().acquire_owned().await else {
266 warn!("historical nostr index query skipped: blocking semaphore closed");
267 return Vec::new();
268 };
269 let result = tokio::task::spawn_blocking(move || {
270 let runtime = tokio::runtime::Builder::new_current_thread()
271 .enable_all()
272 .build()?;
273 let stored_events = runtime.block_on(async move {
274 let event_store = NostrEventStore::new(store);
275 event_store.query_events(Some(&root), &filter, limit).await
276 })?;
277 Ok::<_, anyhow::Error>(stored_events)
278 })
279 .await;
280 match result {
281 Ok(stored_events) => {
282 let stored_events = match stored_events {
283 Ok(stored_events) => stored_events,
284 Err(err) => {
285 warn!("historical nostr index query failed: {}", err);
286 return Vec::new();
287 }
288 };
289 let mut events = Vec::with_capacity(stored_events.len());
290 for stored in stored_events {
291 match VerifiedStoredNostrEvent::try_from(stored)
292 .and_then(|event| event.to_nostr_sdk_event())
293 .map(|event| event.into_event())
294 {
295 Ok(event) => events.push(event),
296 Err(err) => {
297 warn!("historical nostr index skipped invalid event: {}", err)
298 }
299 }
300 }
301 info!(
302 target: "hashtree_cli::nostr_relay::query",
303 limit,
304 events = events.len(),
305 elapsed_ms = started.elapsed().as_millis() as u64,
306 filter = %filter_summary,
307 memory_before = ?memory_before,
308 memory_after = ?process_memory_snapshot(),
309 "historical nostr index query completed",
310 );
311 events
312 }
313 Err(err) => {
314 warn!("historical nostr index query task failed: {}", err);
315 Vec::new()
316 }
317 }
318 }
319
320 async fn load_existing_root(&self) -> Result<Option<Cid>> {
321 if let Some(root) = load_nostr_index_root_file(&self.latest_root_path).await? {
322 return Ok(Some(root));
323 }
324 load_nostr_index_root_file(&self.checkpoint_root_path).await
325 }
326
327 async fn persist_latest_root(&self, root: &Cid) -> Result<()> {
328 if let Some(parent) = self.latest_root_path.parent() {
329 tokio::fs::create_dir_all(parent).await?;
330 }
331 tokio::fs::write(&self.latest_root_path, format!("{}\n", cid_to_nhash(root)?)).await?;
332 Ok(())
333 }
334 }
335
336 async fn load_nostr_index_root_file(path: &Path) -> Result<Option<Cid>> {
337 let root = match tokio::fs::read_to_string(path).await {
338 Ok(root) => root,
339 Err(err) if err.kind() == ErrorKind::NotFound => return Ok(None),
340 Err(err) => return Err(err.into()),
341 };
342 let trimmed = root.trim();
343 if trimmed.is_empty() {
344 return Ok(None);
345 }
346 parse_nostr_index_root(trimmed).map(Some)
347 }
348
349 fn parse_nostr_index_root(value: &str) -> Result<Cid> {
350 if value.starts_with("nhash1") {
351 let decoded = nhash_decode(value)?;
352 return Ok(Cid {
353 hash: decoded.hash,
354 key: decoded.decrypt_key,
355 });
356 }
357 Cid::parse(value).map_err(Into::into)
358 }
359
360 fn cid_to_nhash(cid: &Cid) -> Result<String> {
361 nhash_encode_full(&NHashData {
362 hash: cid.hash,
363 decrypt_key: cid.key,
364 })
365 .map_err(Into::into)
366 }
367
368 #[derive(Debug, Clone)]
369 struct ClientQuota {
370 last_reset: Instant,
371 spambox_events: u32,
372 reqs: u32,
373 }
374
375 impl ClientQuota {
376 fn new() -> Self {
377 Self {
378 last_reset: Instant::now(),
379 spambox_events: 0,
380 reqs: 0,
381 }
382 }
383
384 fn reset_if_needed(&mut self) {
385 if self.last_reset.elapsed() >= Duration::from_secs(60) {
386 self.last_reset = Instant::now();
387 self.spambox_events = 0;
388 self.reqs = 0;
389 }
390 }
391
392 fn allow_spambox_event(&mut self, limit: u32) -> bool {
393 self.reset_if_needed();
394 if self.spambox_events >= limit {
395 return false;
396 }
397 self.spambox_events += 1;
398 true
399 }
400
401 fn allow_req(&mut self, limit: u32) -> bool {
402 self.reset_if_needed();
403 if self.reqs >= limit {
404 return false;
405 }
406 self.reqs += 1;
407 true
408 }
409 }
410
411 struct ClientState {
412 sender: mpsc::UnboundedSender<String>,
413 pubkey: Option<String>,
414 quota: ClientQuota,
415 }
416
417 struct RecentEvents {
418 order: VecDeque<EventId>,
419 events: HashMap<EventId, Event>,
420 max_len: usize,
421 }
422
423 impl RecentEvents {
424 fn new(max_len: usize) -> Self {
425 Self {
426 order: VecDeque::new(),
427 events: HashMap::new(),
428 max_len: max_len.max(128),
429 }
430 }
431
432 fn insert(&mut self, event: Event) {
433 if self.events.contains_key(&event.id) {
434 return;
435 }
436 self.order.push_back(event.id);
437 self.events.insert(event.id, event);
438 while self.order.len() > self.max_len {
439 if let Some(oldest) = self.order.pop_front() {
440 self.events.remove(&oldest);
441 }
442 }
443 }
444
445 fn matching(&self, filter: &NostrFilter) -> Vec<Event> {
446 self.events
447 .values()
448 .filter(|event| filter.match_event(event, Default::default()))
449 .cloned()
450 .collect()
451 }
452 }
453
454 enum SpamboxStore {
455 Persistent(NostrStore),
456 Memory(MemorySpambox),
457 }
458
459 struct MemorySpambox {
460 events: Mutex<VecDeque<Event>>,
461 max_len: usize,
462 }
463
464 impl MemorySpambox {
465 fn new(max_len: usize) -> Self {
466 Self {
467 events: Mutex::new(VecDeque::new()),
468 max_len: max_len.max(128),
469 }
470 }
471
472 async fn ingest(&self, event: &Event) -> bool {
473 let mut events = self.events.lock().await;
474 events.push_back(event.clone());
475 while events.len() > self.max_len {
476 events.pop_front();
477 }
478 true
479 }
480 }
481
482 impl SpamboxStore {
483 async fn ingest(&self, event: &Event) -> bool {
484 match self {
485 SpamboxStore::Persistent(store) => store.ingest(event.clone()).await.is_ok(),
486 SpamboxStore::Memory(store) => store.ingest(event).await,
487 }
488 }
489 }
490
491 struct BluetoothEventLog {
492 path: PathBuf,
493 state: Mutex<BluetoothEventLogState>,
494 }
495
496 struct BluetoothEventLogState {
497 records: VecDeque<BluetoothReceivedEventRecord>,
498 event_ids: HashSet<String>,
499 }
500
501 impl BluetoothEventLog {
502 fn load(path: PathBuf) -> Self {
503 let records = std::fs::read_to_string(&path)
504 .ok()
505 .map(|serialized| {
506 serialized
507 .lines()
508 .filter_map(|line| {
509 serde_json::from_str::<BluetoothReceivedEventRecord>(line).ok()
510 })
511 .collect::<Vec<_>>()
512 })
513 .unwrap_or_default();
514 let mut trimmed = VecDeque::with_capacity(BLUETOOTH_EVENT_LOG_CAPACITY);
515 let start = records.len().saturating_sub(BLUETOOTH_EVENT_LOG_CAPACITY);
516 for record in records.into_iter().skip(start) {
517 trimmed.push_back(record);
518 }
519 let event_ids = trimmed
520 .iter()
521 .map(|record| record.event_id.clone())
522 .collect::<HashSet<_>>();
523
524 Self {
525 path,
526 state: Mutex::new(BluetoothEventLogState {
527 records: trimmed,
528 event_ids,
529 }),
530 }
531 }
532
533 async fn recent(&self, limit: usize) -> Vec<BluetoothReceivedEventRecord> {
534 let state = self.state.lock().await;
535 state
536 .records
537 .iter()
538 .rev()
539 .take(limit.max(1))
540 .cloned()
541 .collect()
542 }
543
544 async fn record(&self, event: &Event, peer_id: Option<String>) {
545 let record = BluetoothReceivedEventRecord {
546 event_id: event.id.to_hex(),
547 pubkey: event.pubkey.to_hex(),
548 kind: event.kind.as_u16() as u32,
549 created_at: event.created_at.as_secs(),
550 received_at: std::time::SystemTime::now()
551 .duration_since(std::time::UNIX_EPOCH)
552 .map(|value| value.as_secs())
553 .unwrap_or(0),
554 peer_id,
555 cid_values: cid_values_from_event(event),
556 };
557
558 let serialized = {
559 let mut state = self.state.lock().await;
560 if state.event_ids.contains(&record.event_id) {
561 return;
562 }
563
564 state.event_ids.insert(record.event_id.clone());
565 state.records.push_back(record);
566 while state.records.len() > BLUETOOTH_EVENT_LOG_CAPACITY {
567 if let Some(removed) = state.records.pop_front() {
568 state.event_ids.remove(&removed.event_id);
569 }
570 }
571
572 state
573 .records
574 .iter()
575 .filter_map(|entry| serde_json::to_string(entry).ok())
576 .collect::<Vec<_>>()
577 .join("\n")
578 };
579
580 if let Some(parent) = self.path.parent() {
581 let _ = std::fs::create_dir_all(parent);
582 }
583 let _ = std::fs::write(&self.path, serialized);
584 }
585 }
586
587 fn looks_like_cid_reference(value: &str) -> bool {
588 Cid::parse(value).is_ok() || nhash_decode(value).is_ok()
589 }
590
591 fn cid_values_from_event(event: &Event) -> Vec<String> {
592 let mut values = Vec::new();
593 let mut seen = HashSet::new();
594
595 for tag in event.tags.iter() {
596 let fields = tag.clone().to_vec();
597 if fields.first().is_some_and(|name| name == "cid") {
598 if let Some(value) = fields.get(1).filter(|value| !value.is_empty()) {
599 if seen.insert(value.clone()) {
600 values.push(value.clone());
601 }
602 }
603 continue;
604 }
605
606 for value in fields.into_iter().skip(1) {
607 if looks_like_cid_reference(&value) && seen.insert(value.clone()) {
608 values.push(value);
609 }
610 }
611 }
612
613 values
614 }
615
616 pub struct NostrRelay {
617 config: NostrRelayConfig,
618 trusted: NostrStore,
619 public_pubkeys: HashSet<String>,
620 spambox: Option<SpamboxStore>,
621 historical_index: Option<HistoricalNostrIndex>,
622 social_graph: Option<Arc<SocialGraphAccessControl>>,
623 clients: Mutex<HashMap<u64, ClientState>>,
624 subscriptions: Mutex<HashMap<u64, HashMap<SubscriptionId, Vec<NostrFilter>>>>,
625 recent_events: Mutex<RecentEvents>,
626 next_client_id: AtomicU64,
627 bluetooth_event_log: Arc<BluetoothEventLog>,
628 #[cfg(feature = "experimental-decentralized-pubsub")]
629 decentralized_pubsub_tx: std::sync::Mutex<Option<mpsc::UnboundedSender<Event>>>,
630 }
631
632 impl NostrRelay {
633 async fn collect_filter_events(
634 &self,
635 filter: &NostrFilter,
636 limit: usize,
637 seen: &mut HashSet<EventId>,
638 events: &mut Vec<Event>,
639 ) {
640 if limit == 0 {
641 return;
642 }
643
644 let mut added = 0usize;
645
646 if !prefers_trusted_only(filter) {
647 let recent = {
648 let cache = self.recent_events.lock().await;
649 cache.matching(filter)
650 };
651 for event in recent {
652 if seen.insert(event.id) {
653 events.push(event);
654 added += 1;
655 if added >= limit {
656 return;
657 }
658 }
659 }
660 }
661
662 for event in self.trusted.query(filter.clone(), limit).await {
663 if seen.insert(event.id) {
664 events.push(event);
665 added += 1;
666 if added >= limit {
667 return;
668 }
669 }
670 }
671
672 if let Some(index) = &self.historical_index {
673 let remaining = limit.saturating_sub(added);
674 for event in index.query(filter, remaining).await {
675 if seen.insert(event.id) {
676 events.push(event);
677 added += 1;
678 if added >= limit {
679 return;
680 }
681 }
682 }
683 }
684 }
685
686 async fn collect_filter_count(
687 &self,
688 filter: &NostrFilter,
689 limit: usize,
690 seen: &mut HashSet<EventId>,
691 ) {
692 if limit == 0 {
693 return;
694 }
695
696 let mut added = 0usize;
697
698 if !prefers_trusted_only(filter) {
699 let recent = {
700 let cache = self.recent_events.lock().await;
701 cache.matching(filter)
702 };
703 for event in recent {
704 if seen.insert(event.id) {
705 added += 1;
706 if added >= limit {
707 return;
708 }
709 }
710 }
711 }
712
713 for event in self.trusted.query(filter.clone(), limit).await {
714 if seen.insert(event.id) {
715 added += 1;
716 if added >= limit {
717 return;
718 }
719 }
720 }
721
722 if let Some(index) = &self.historical_index {
723 let remaining = limit.saturating_sub(added);
724 for event in index.query(filter, remaining).await {
725 if seen.insert(event.id) {
726 added += 1;
727 if added >= limit {
728 return;
729 }
730 }
731 }
732 }
733 }
734
735 pub fn new(
736 trusted_store: Arc<dyn SocialGraphBackend>,
737 data_dir: PathBuf,
738 public_pubkeys: HashSet<String>,
739 social_graph: Option<Arc<SocialGraphAccessControl>>,
740 config: NostrRelayConfig,
741 ) -> Result<Self> {
742 let spambox = if config.spambox_db_max_bytes == 0 {
743 Some(SpamboxStore::Memory(MemorySpambox::new(
744 config.max_query_limit * 2,
745 )))
746 } else {
747 let spam_dir = data_dir.join("socialgraph_spambox");
748 match socialgraph::open_social_graph_store_at_path(
749 &spam_dir,
750 Some(config.spambox_db_max_bytes),
751 ) {
752 Ok(store) => Some(SpamboxStore::Persistent(NostrStore::new(store))),
753 Err(err) => {
754 warn!(
755 "Failed to open social graph spambox (falling back to memory): {}",
756 err
757 );
758 Some(SpamboxStore::Memory(MemorySpambox::new(
759 config.max_query_limit * 2,
760 )))
761 }
762 }
763 };
764
765 let recent_size = config.max_query_limit.saturating_mul(2);
766 let bluetooth_event_log = Arc::new(BluetoothEventLog::load(
767 data_dir.join("bluetooth-events.jsonl"),
768 ));
769
770 Ok(Self {
771 config,
772 trusted: NostrStore::new(trusted_store),
773 public_pubkeys,
774 spambox,
775 historical_index: None,
776 social_graph,
777 clients: Mutex::new(HashMap::new()),
778 subscriptions: Mutex::new(HashMap::new()),
779 recent_events: Mutex::new(RecentEvents::new(recent_size)),
780 next_client_id: AtomicU64::new(1),
781 bluetooth_event_log,
782 #[cfg(feature = "experimental-decentralized-pubsub")]
783 decentralized_pubsub_tx: std::sync::Mutex::new(None),
784 })
785 }
786
787 pub fn with_historical_nostr_index(
788 mut self,
789 store: Arc<StorageRouter>,
790 data_dir: PathBuf,
791 ) -> Self {
792 self.historical_index = Some(HistoricalNostrIndex::new(store, data_dir));
793 self
794 }
795
796 pub fn next_client_id(&self) -> u64 {
797 self.next_client_id.fetch_add(1, Ordering::SeqCst)
798 }
799
800 #[cfg(feature = "experimental-decentralized-pubsub")]
801 pub fn set_decentralized_pubsub_sender(
802 &self,
803 sender: Option<mpsc::UnboundedSender<Event>>,
804 ) {
805 match self.decentralized_pubsub_tx.lock() {
806 Ok(mut slot) => {
807 *slot = sender;
808 }
809 Err(err) => {
810 warn!("nostr decentralized pubsub sender lock poisoned: {}", err);
811 }
812 }
813 }
814
815 #[cfg(feature = "experimental-decentralized-pubsub")]
816 fn enqueue_decentralized_pubsub_event(&self, event: &Event) {
817 let sender = match self.decentralized_pubsub_tx.lock() {
818 Ok(slot) => slot.clone(),
819 Err(err) => {
820 warn!("nostr decentralized pubsub sender lock poisoned: {}", err);
821 None
822 }
823 };
824
825 if let Some(sender) = sender {
826 if sender.send(event.clone()).is_err() {
827 warn!("nostr decentralized pubsub publisher is not running");
828 }
829 }
830 }
831
832 pub async fn ingest_trusted_event(&self, event: Event) -> Result<()> {
833 self.ingest_trusted_event_inner(event, true).await
834 }
835
836 pub async fn ingest_trusted_event_from_bluetooth(
837 &self,
838 event: Event,
839 peer_id: Option<String>,
840 ) -> Result<()> {
841 self.ingest_trusted_event_inner(event.clone(), true).await?;
842 self.bluetooth_event_log.record(&event, peer_id).await;
843 Ok(())
844 }
845
846 pub async fn ingest_trusted_event_silent(&self, event: Event) -> Result<()> {
847 self.ingest_trusted_event_inner(event, false).await
848 }
849
850 pub async fn ingest_peer_event_silent(&self, event: Event) -> Result<bool> {
851 event
852 .verify()
853 .map_err(|e| anyhow::anyhow!("invalid signature: {}", e))?;
854
855 if !self.is_trusted_event_for_client(None, &event).await {
856 return Ok(false);
857 }
858
859 let is_ephemeral = event.kind.is_ephemeral();
860 {
861 let mut recent = self.recent_events.lock().await;
862 recent.insert(event.clone());
863 }
864 if !is_ephemeral {
865 let storage_class = self.event_storage_class(&event);
866 self.trusted
867 .ingest_with_storage_class(event.clone(), storage_class)
868 .await?;
869 self.append_historical_index(&event).await;
870 }
871
872 Ok(true)
873 }
874
875 pub async fn bluetooth_received_events(
876 &self,
877 limit: usize,
878 ) -> Vec<BluetoothReceivedEventRecord> {
879 self.bluetooth_event_log.recent(limit).await
880 }
881
882 async fn ingest_trusted_event_inner(&self, event: Event, broadcast: bool) -> Result<()> {
883 event
884 .verify()
885 .map_err(|e| anyhow::anyhow!("invalid signature: {}", e))?;
886
887 let is_ephemeral = event.kind.is_ephemeral();
888 {
889 let mut recent = self.recent_events.lock().await;
890 recent.insert(event.clone());
891 }
892
893 if !is_ephemeral {
894 let storage_class = self.event_storage_class(&event);
895 self.trusted
896 .ingest_with_storage_class(event.clone(), storage_class)
897 .await?;
898 self.append_historical_index(&event).await;
899 }
900
901 if broadcast {
902 self.broadcast_event(&event).await;
903 }
904 Ok(())
905 }
906
907 pub async fn query_events(&self, filter: &NostrFilter, limit: usize) -> Vec<Event> {
908 let limit = limit.min(self.config.max_query_limit);
909 if limit == 0 {
910 return Vec::new();
911 }
912
913 let mut seen: HashSet<EventId> = HashSet::new();
914 let mut events = Vec::new();
915
916 if !prefers_trusted_only(filter) {
917 let recent = {
918 let cache = self.recent_events.lock().await;
919 cache.matching(filter)
920 };
921 for event in recent {
922 if seen.insert(event.id) {
923 events.push(event);
924 if events.len() >= limit {
925 return events;
926 }
927 }
928 }
929 }
930
931 for event in self.trusted.query(filter.clone(), limit).await {
932 if seen.insert(event.id) {
933 events.push(event);
934 if events.len() >= limit {
935 break;
936 }
937 }
938 }
939
940 if let Some(index) = &self.historical_index {
941 let remaining = limit.saturating_sub(events.len());
942 for event in index.query(filter, remaining).await {
943 if seen.insert(event.id) {
944 events.push(event);
945 if events.len() >= limit {
946 break;
947 }
948 }
949 }
950 }
951
952 events
953 }
954
955 pub async fn register_client(
956 &self,
957 client_id: u64,
958 sender: mpsc::UnboundedSender<String>,
959 pubkey: Option<String>,
960 ) {
961 let mut clients = self.clients.lock().await;
962 clients.insert(
963 client_id,
964 ClientState {
965 sender,
966 pubkey,
967 quota: ClientQuota::new(),
968 },
969 );
970 }
971
972 pub async fn unregister_client(&self, client_id: u64) {
973 let mut clients = self.clients.lock().await;
974 clients.remove(&client_id);
975 drop(clients);
976 let mut subs = self.subscriptions.lock().await;
977 subs.remove(&client_id);
978 }
979
980 pub async fn handle_client_message(&self, client_id: u64, msg: NostrClientMessage<'_>) {
981 match msg {
982 NostrClientMessage::Event(event) => {
983 self.handle_event(client_id, event.into_owned()).await;
984 }
985 NostrClientMessage::Req {
986 subscription_id,
987 filters,
988 } => {
989 self.handle_req(
990 client_id,
991 subscription_id.into_owned(),
992 filters
993 .into_iter()
994 .map(|filter| filter.into_owned())
995 .collect(),
996 )
997 .await;
998 }
999 NostrClientMessage::Count {
1000 subscription_id,
1001 filter,
1002 } => {
1003 self.handle_count(
1004 client_id,
1005 subscription_id.into_owned(),
1006 vec![filter.into_owned()],
1007 )
1008 .await;
1009 }
1010 NostrClientMessage::Close(subscription_id) => {
1011 self.handle_close(client_id, subscription_id.into_owned())
1012 .await;
1013 }
1014 NostrClientMessage::Auth(event) => {
1015 self.handle_auth(client_id, event.into_owned()).await;
1016 }
1017 NostrClientMessage::NegOpen { .. }
1018 | NostrClientMessage::NegMsg { .. }
1019 | NostrClientMessage::NegClose { .. } => {
1020 self.send_to_client(
1021 client_id,
1022 NostrRelayMessage::notice("negentropy not supported"),
1023 )
1024 .await;
1025 }
1026 }
1027 }
1028
1029 pub async fn register_subscription_query(
1030 &self,
1031 client_id: u64,
1032 subscription_id: SubscriptionId,
1033 mut filters: Vec<NostrFilter>,
1034 ) -> std::result::Result<Vec<Event>, &'static str> {
1035 if !self.allow_req(client_id).await {
1036 return Err("rate limited");
1037 }
1038
1039 if filters.len() > self.config.max_filters_per_sub {
1040 filters.truncate(self.config.max_filters_per_sub);
1041 }
1042
1043 {
1044 let mut subs = self.subscriptions.lock().await;
1045 let entry = subs.entry(client_id).or_default();
1046 if !entry.contains_key(&subscription_id)
1047 && entry.len() >= self.config.max_subs_per_client
1048 {
1049 return Err("too many subscriptions");
1050 }
1051 entry.insert(subscription_id.clone(), filters.clone());
1052 }
1053
1054 let mut seen: HashSet<EventId> = HashSet::new();
1055 let mut events = Vec::new();
1056 let memory_before = process_memory_snapshot();
1057 let started = Instant::now();
1058 let filter_summary = nostr_filters_summary(&filters);
1059 for filter in &filters {
1060 let remaining = self.config.max_query_limit.saturating_sub(events.len());
1061 if remaining == 0 {
1062 break;
1063 }
1064 let limit = filter
1065 .limit
1066 .unwrap_or(self.config.max_query_limit)
1067 .min(self.config.max_query_limit)
1068 .min(remaining);
1069 self.collect_filter_events(filter, limit, &mut seen, &mut events)
1070 .await;
1071 }
1072
1073 info!(
1074 target: "hashtree_cli::nostr_relay::query",
1075 client_id,
1076 subscription_id = %subscription_id,
1077 filters = filters.len(),
1078 events = events.len(),
1079 elapsed_ms = started.elapsed().as_millis() as u64,
1080 filter = %filter_summary,
1081 memory_before = ?memory_before,
1082 memory_after = ?process_memory_snapshot(),
1083 "nostr relay local subscription query completed",
1084 );
1085 Ok(events)
1086 }
1087
1088 async fn handle_auth(&self, client_id: u64, event: Event) {
1089 let ok = event.verify().is_ok();
1090 let message = if ok { "" } else { "invalid auth" };
1091 self.send_to_client(client_id, NostrRelayMessage::ok(event.id, ok, message))
1092 .await;
1093 }
1094
1095 async fn handle_close(&self, client_id: u64, subscription_id: SubscriptionId) {
1096 let mut subs = self.subscriptions.lock().await;
1097 if let Some(map) = subs.get_mut(&client_id) {
1098 map.remove(&subscription_id);
1099 }
1100 }
1101
1102 async fn handle_event(&self, client_id: u64, event: Event) {
1103 let ok = event.verify().is_ok();
1104 if !ok {
1105 self.send_to_client(
1106 client_id,
1107 NostrRelayMessage::ok(event.id, false, "invalid: signature"),
1108 )
1109 .await;
1110 return;
1111 }
1112
1113 let trusted = self.is_trusted_event(client_id, &event).await;
1114 if !trusted && !self.allow_spambox_event(client_id).await {
1115 self.send_to_client(
1116 client_id,
1117 NostrRelayMessage::ok(event.id, false, "rate limited"),
1118 )
1119 .await;
1120 return;
1121 }
1122
1123 let is_ephemeral = event.kind.is_ephemeral();
1124 if trusted {
1125 let mut recent = self.recent_events.lock().await;
1126 recent.insert(event.clone());
1127 }
1128 if !is_ephemeral {
1129 let stored = if trusted {
1130 let storage_class = self.event_storage_class(&event);
1131 let stored = self
1132 .trusted
1133 .ingest_with_storage_class(event.clone(), storage_class)
1134 .await
1135 .is_ok();
1136 if stored {
1137 self.append_historical_index(&event).await;
1138 }
1139 stored
1140 } else {
1141 match self.spambox.as_ref() {
1142 Some(spambox) => spambox.ingest(&event).await,
1143 None => false,
1144 }
1145 };
1146
1147 if !stored {
1148 let message = if trusted {
1149 "store failed"
1150 } else {
1151 "spambox full"
1152 };
1153 self.send_to_client(client_id, NostrRelayMessage::ok(event.id, false, message))
1154 .await;
1155 return;
1156 }
1157 }
1158
1159 let message = if trusted { "" } else { "spambox" };
1160 self.send_to_client(client_id, NostrRelayMessage::ok(event.id, true, message))
1161 .await;
1162
1163 if trusted {
1164 self.broadcast_event(&event).await;
1165 #[cfg(feature = "experimental-decentralized-pubsub")]
1166 self.enqueue_decentralized_pubsub_event(&event);
1167 }
1168 }
1169
1170 async fn handle_req(
1171 &self,
1172 client_id: u64,
1173 subscription_id: SubscriptionId,
1174 filters: Vec<NostrFilter>,
1175 ) {
1176 match self
1177 .register_subscription_query(client_id, subscription_id.clone(), filters)
1178 .await
1179 {
1180 Ok(events) => {
1181 for event in events {
1182 self.send_to_client(
1183 client_id,
1184 NostrRelayMessage::event(subscription_id.clone(), event),
1185 )
1186 .await;
1187 }
1188 trim_process_allocations();
1189
1190 self.send_to_client(client_id, NostrRelayMessage::eose(subscription_id))
1191 .await;
1192 }
1193 Err(message) => {
1194 self.send_to_client(
1195 client_id,
1196 NostrRelayMessage::closed(subscription_id, message),
1197 )
1198 .await;
1199 }
1200 }
1201 }
1202
1203 async fn handle_count(
1204 &self,
1205 client_id: u64,
1206 subscription_id: SubscriptionId,
1207 filters: Vec<NostrFilter>,
1208 ) {
1209 if !self.allow_req(client_id).await {
1210 self.send_to_client(
1211 client_id,
1212 NostrRelayMessage::closed(subscription_id, "rate limited"),
1213 )
1214 .await;
1215 return;
1216 }
1217
1218 let mut seen: HashSet<EventId> = HashSet::new();
1219 for filter in &filters {
1220 let limit = filter
1221 .limit
1222 .unwrap_or(self.config.max_query_limit)
1223 .min(self.config.max_query_limit);
1224 self.collect_filter_count(filter, limit, &mut seen).await;
1225 }
1226
1227 self.send_to_client(
1228 client_id,
1229 NostrRelayMessage::count(subscription_id, seen.len()),
1230 )
1231 .await;
1232 }
1233
1234 async fn is_trusted_event(&self, client_id: u64, event: &Event) -> bool {
1235 self.is_trusted_event_for_client(Some(client_id), event)
1236 .await
1237 }
1238
1239 async fn is_trusted_event_for_client(&self, client_id: Option<u64>, event: &Event) -> bool {
1240 let event_pubkey = event.pubkey.to_hex();
1241 let client_pubkey = {
1242 let clients = self.clients.lock().await;
1243 client_id.and_then(|client_id| {
1244 clients
1245 .get(&client_id)
1246 .and_then(|state| state.pubkey.clone())
1247 })
1248 };
1249 if let Some(pubkey) = client_pubkey {
1250 return pubkey == event_pubkey
1251 || self.social_graph.as_ref().is_some_and(|social_graph| {
1252 social_graph.check_write_access(&event_pubkey)
1253 });
1254 }
1255 if let Some(ref social_graph) = self.social_graph {
1256 return social_graph.check_write_access(&event_pubkey);
1257 }
1258 true
1259 }
1260
1261 async fn append_historical_index(&self, event: &Event) {
1262 if let Some(index) = &self.historical_index {
1263 if let Err(err) = index.ingest(event.clone()).await {
1264 warn!("historical nostr index ingest failed: {}", err);
1265 }
1266 }
1267 }
1268
1269 fn event_storage_class(&self, event: &Event) -> EventStorageClass {
1270 if self.public_pubkeys.contains(&event.pubkey.to_hex()) {
1271 EventStorageClass::Public
1272 } else {
1273 EventStorageClass::Ambient
1274 }
1275 }
1276
1277 async fn allow_spambox_event(&self, client_id: u64) -> bool {
1278 let mut clients = self.clients.lock().await;
1279 let Some(state) = clients.get_mut(&client_id) else {
1280 return false;
1281 };
1282 state
1283 .quota
1284 .allow_spambox_event(self.config.spambox_max_events_per_min)
1285 }
1286
1287 async fn allow_req(&self, client_id: u64) -> bool {
1288 let mut clients = self.clients.lock().await;
1289 let Some(state) = clients.get_mut(&client_id) else {
1290 return false;
1291 };
1292 state.quota.allow_req(self.config.spambox_max_reqs_per_min)
1293 }
1294
1295 async fn broadcast_event(&self, event: &Event) {
1296 let subscriptions = self.subscriptions.lock().await;
1297 let mut deliveries: Vec<(u64, SubscriptionId)> = Vec::new();
1298 for (client_id, subs) in subscriptions.iter() {
1299 for (sub_id, filters) in subs.iter() {
1300 if filters
1301 .iter()
1302 .any(|f| f.match_event(event, Default::default()))
1303 {
1304 deliveries.push((*client_id, sub_id.clone()));
1305 }
1306 }
1307 }
1308 drop(subscriptions);
1309
1310 for (client_id, sub_id) in deliveries {
1311 self.send_to_client(client_id, NostrRelayMessage::event(sub_id, event.clone()))
1312 .await;
1313 }
1314 }
1315
1316 async fn send_to_client(&self, client_id: u64, msg: NostrRelayMessage<'_>) {
1317 let sender = {
1318 let clients = self.clients.lock().await;
1319 clients.get(&client_id).map(|state| state.sender.clone())
1320 };
1321 if let Some(tx) = sender {
1322 let _ = tx.send(msg.as_json());
1323 }
1324 }
1325 }
1326}
1327
1328pub use imp::NostrRelay;
1329
1330#[cfg(test)]
1331#[path = "nostr_relay/tests.rs"]
1332mod tests;