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: Option<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, durable_writes_enabled: bool) -> 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: durable_writes_enabled.then_some(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 let Some(path) = self.path.as_ref() else {
581 return;
582 };
583 if let Some(parent) = path.parent() {
584 let _ = std::fs::create_dir_all(parent);
585 }
586 let _ = std::fs::write(path, serialized);
587 }
588 }
589
590 fn looks_like_cid_reference(value: &str) -> bool {
591 Cid::parse(value).is_ok() || nhash_decode(value).is_ok()
592 }
593
594 fn cid_values_from_event(event: &Event) -> Vec<String> {
595 let mut values = Vec::new();
596 let mut seen = HashSet::new();
597
598 for tag in event.tags.iter() {
599 let fields = tag.clone().to_vec();
600 if fields.first().is_some_and(|name| name == "cid") {
601 if let Some(value) = fields.get(1).filter(|value| !value.is_empty()) {
602 if seen.insert(value.clone()) {
603 values.push(value.clone());
604 }
605 }
606 continue;
607 }
608
609 for value in fields.into_iter().skip(1) {
610 if looks_like_cid_reference(&value) && seen.insert(value.clone()) {
611 values.push(value);
612 }
613 }
614 }
615
616 values
617 }
618
619 pub struct NostrRelay {
620 config: NostrRelayConfig,
621 trusted: Option<NostrStore>,
622 public_pubkeys: HashSet<String>,
623 spambox: Option<SpamboxStore>,
624 historical_index: Option<HistoricalNostrIndex>,
625 social_graph: Option<Arc<SocialGraphAccessControl>>,
626 clients: Mutex<HashMap<u64, ClientState>>,
627 subscriptions: Mutex<HashMap<u64, HashMap<SubscriptionId, Vec<NostrFilter>>>>,
628 recent_events: Mutex<RecentEvents>,
629 next_client_id: AtomicU64,
630 bluetooth_event_log: Arc<BluetoothEventLog>,
631 #[cfg(feature = "experimental-decentralized-pubsub")]
632 decentralized_pubsub_tx: std::sync::Mutex<Option<mpsc::UnboundedSender<Event>>>,
633 }
634
635 impl NostrRelay {
636 async fn collect_filter_events(
637 &self,
638 filter: &NostrFilter,
639 limit: usize,
640 seen: &mut HashSet<EventId>,
641 events: &mut Vec<Event>,
642 ) {
643 if limit == 0 {
644 return;
645 }
646
647 let mut added = 0usize;
648
649 if self.trusted.is_none() || !prefers_trusted_only(filter) {
650 let recent = {
651 let cache = self.recent_events.lock().await;
652 cache.matching(filter)
653 };
654 for event in recent {
655 if seen.insert(event.id) {
656 events.push(event);
657 added += 1;
658 if added >= limit {
659 return;
660 }
661 }
662 }
663 }
664
665 if let Some(trusted) = self.trusted.as_ref() {
666 for event in trusted.query(filter.clone(), limit).await {
667 if seen.insert(event.id) {
668 events.push(event);
669 added += 1;
670 if added >= limit {
671 return;
672 }
673 }
674 }
675 }
676
677 if let Some(index) = &self.historical_index {
678 let remaining = limit.saturating_sub(added);
679 for event in index.query(filter, remaining).await {
680 if seen.insert(event.id) {
681 events.push(event);
682 added += 1;
683 if added >= limit {
684 return;
685 }
686 }
687 }
688 }
689 }
690
691 async fn collect_filter_count(
692 &self,
693 filter: &NostrFilter,
694 limit: usize,
695 seen: &mut HashSet<EventId>,
696 ) {
697 if limit == 0 {
698 return;
699 }
700
701 let mut added = 0usize;
702
703 if self.trusted.is_none() || !prefers_trusted_only(filter) {
704 let recent = {
705 let cache = self.recent_events.lock().await;
706 cache.matching(filter)
707 };
708 for event in recent {
709 if seen.insert(event.id) {
710 added += 1;
711 if added >= limit {
712 return;
713 }
714 }
715 }
716 }
717
718 if let Some(trusted) = self.trusted.as_ref() {
719 for event in trusted.query(filter.clone(), limit).await {
720 if seen.insert(event.id) {
721 added += 1;
722 if added >= limit {
723 return;
724 }
725 }
726 }
727 }
728
729 if let Some(index) = &self.historical_index {
730 let remaining = limit.saturating_sub(added);
731 for event in index.query(filter, remaining).await {
732 if seen.insert(event.id) {
733 added += 1;
734 if added >= limit {
735 return;
736 }
737 }
738 }
739 }
740 }
741
742 pub fn new(
743 trusted_store: Arc<dyn SocialGraphBackend>,
744 data_dir: PathBuf,
745 public_pubkeys: HashSet<String>,
746 social_graph: Option<Arc<SocialGraphAccessControl>>,
747 config: NostrRelayConfig,
748 ) -> Result<Self> {
749 let spambox = if config.spambox_db_max_bytes == 0 {
750 Some(SpamboxStore::Memory(MemorySpambox::new(
751 config.max_query_limit * 2,
752 )))
753 } else {
754 let spam_dir = data_dir.join("socialgraph_spambox");
755 match socialgraph::open_social_graph_store_at_path(
756 &spam_dir,
757 Some(config.spambox_db_max_bytes),
758 ) {
759 Ok(store) => Some(SpamboxStore::Persistent(NostrStore::new(store))),
760 Err(err) => {
761 warn!(
762 "Failed to open social graph spambox (falling back to memory): {}",
763 err
764 );
765 Some(SpamboxStore::Memory(MemorySpambox::new(
766 config.max_query_limit * 2,
767 )))
768 }
769 }
770 };
771
772 let recent_size = config.max_query_limit.saturating_mul(2);
773 let bluetooth_event_log = Arc::new(BluetoothEventLog::load(
774 data_dir.join("bluetooth-events.jsonl"),
775 true,
776 ));
777
778 Ok(Self {
779 config,
780 trusted: Some(NostrStore::new(trusted_store)),
781 public_pubkeys,
782 spambox,
783 historical_index: None,
784 social_graph,
785 clients: Mutex::new(HashMap::new()),
786 subscriptions: Mutex::new(HashMap::new()),
787 recent_events: Mutex::new(RecentEvents::new(recent_size)),
788 next_client_id: AtomicU64::new(1),
789 bluetooth_event_log,
790 #[cfg(feature = "experimental-decentralized-pubsub")]
791 decentralized_pubsub_tx: std::sync::Mutex::new(None),
792 })
793 }
794
795 pub fn new_read_only(data_dir: PathBuf, config: NostrRelayConfig) -> Self {
800 let recent_size = config.max_query_limit.saturating_mul(2);
801 let bluetooth_event_log = Arc::new(BluetoothEventLog::load(
802 data_dir.join("bluetooth-events.jsonl"),
803 false,
804 ));
805 Self {
806 config,
807 trusted: None,
808 public_pubkeys: HashSet::new(),
809 spambox: None,
810 historical_index: None,
811 social_graph: None,
812 clients: Mutex::new(HashMap::new()),
813 subscriptions: Mutex::new(HashMap::new()),
814 recent_events: Mutex::new(RecentEvents::new(recent_size)),
815 next_client_id: AtomicU64::new(1),
816 bluetooth_event_log,
817 #[cfg(feature = "experimental-decentralized-pubsub")]
818 decentralized_pubsub_tx: std::sync::Mutex::new(None),
819 }
820 }
821
822 pub fn with_historical_nostr_index(
823 mut self,
824 store: Arc<StorageRouter>,
825 data_dir: PathBuf,
826 ) -> Self {
827 if self.trusted.is_some() {
828 self.historical_index = Some(HistoricalNostrIndex::new(store, data_dir));
829 }
830 self
831 }
832
833 pub fn next_client_id(&self) -> u64 {
834 self.next_client_id.fetch_add(1, Ordering::SeqCst)
835 }
836
837 #[cfg(feature = "experimental-decentralized-pubsub")]
838 pub fn set_decentralized_pubsub_sender(
839 &self,
840 sender: Option<mpsc::UnboundedSender<Event>>,
841 ) {
842 match self.decentralized_pubsub_tx.lock() {
843 Ok(mut slot) => {
844 *slot = sender;
845 }
846 Err(err) => {
847 warn!("nostr decentralized pubsub sender lock poisoned: {}", err);
848 }
849 }
850 }
851
852 #[cfg(feature = "experimental-decentralized-pubsub")]
853 fn enqueue_decentralized_pubsub_event(&self, event: &Event) {
854 let sender = match self.decentralized_pubsub_tx.lock() {
855 Ok(slot) => slot.clone(),
856 Err(err) => {
857 warn!("nostr decentralized pubsub sender lock poisoned: {}", err);
858 None
859 }
860 };
861
862 if let Some(sender) = sender {
863 if sender.send(event.clone()).is_err() {
864 warn!("nostr decentralized pubsub publisher is not running");
865 }
866 }
867 }
868
869 pub async fn ingest_trusted_event(&self, event: Event) -> Result<()> {
870 self.ingest_trusted_event_inner(event, true).await
871 }
872
873 pub async fn ingest_trusted_event_from_bluetooth(
874 &self,
875 event: Event,
876 peer_id: Option<String>,
877 ) -> Result<()> {
878 self.ingest_trusted_event_inner(event.clone(), true).await?;
879 self.bluetooth_event_log.record(&event, peer_id).await;
880 Ok(())
881 }
882
883 pub async fn ingest_trusted_event_silent(&self, event: Event) -> Result<()> {
884 self.ingest_trusted_event_inner(event, false).await
885 }
886
887 pub async fn ingest_peer_event_silent(&self, event: Event) -> Result<bool> {
888 event
889 .verify()
890 .map_err(|e| anyhow::anyhow!("invalid signature: {}", e))?;
891
892 if !self.is_trusted_event_for_client(None, &event).await {
893 return Ok(false);
894 }
895
896 let is_ephemeral = event.kind.is_ephemeral();
897 {
898 let mut recent = self.recent_events.lock().await;
899 recent.insert(event.clone());
900 }
901 if !is_ephemeral {
902 let storage_class = self.event_storage_class(&event);
903 if let Some(trusted) = self.trusted.as_ref() {
904 trusted
905 .ingest_with_storage_class(event.clone(), storage_class)
906 .await?;
907 self.append_historical_index(&event).await;
908 }
909 }
910
911 Ok(true)
912 }
913
914 pub async fn bluetooth_received_events(
915 &self,
916 limit: usize,
917 ) -> Vec<BluetoothReceivedEventRecord> {
918 self.bluetooth_event_log.recent(limit).await
919 }
920
921 async fn ingest_trusted_event_inner(&self, event: Event, broadcast: bool) -> Result<()> {
922 event
923 .verify()
924 .map_err(|e| anyhow::anyhow!("invalid signature: {}", e))?;
925
926 let is_ephemeral = event.kind.is_ephemeral();
927 {
928 let mut recent = self.recent_events.lock().await;
929 recent.insert(event.clone());
930 }
931
932 if !is_ephemeral {
933 let storage_class = self.event_storage_class(&event);
934 if let Some(trusted) = self.trusted.as_ref() {
935 trusted
936 .ingest_with_storage_class(event.clone(), storage_class)
937 .await?;
938 self.append_historical_index(&event).await;
939 }
940 }
941
942 if broadcast {
943 self.broadcast_event(&event).await;
944 }
945 Ok(())
946 }
947
948 pub async fn query_events(&self, filter: &NostrFilter, limit: usize) -> Vec<Event> {
949 let limit = limit.min(self.config.max_query_limit);
950 if limit == 0 {
951 return Vec::new();
952 }
953
954 let mut seen: HashSet<EventId> = HashSet::new();
955 let mut events = Vec::new();
956
957 if self.trusted.is_none() || !prefers_trusted_only(filter) {
958 let recent = {
959 let cache = self.recent_events.lock().await;
960 cache.matching(filter)
961 };
962 for event in recent {
963 if seen.insert(event.id) {
964 events.push(event);
965 if events.len() >= limit {
966 return events;
967 }
968 }
969 }
970 }
971
972 if let Some(trusted) = self.trusted.as_ref() {
973 for event in trusted.query(filter.clone(), limit).await {
974 if seen.insert(event.id) {
975 events.push(event);
976 if events.len() >= limit {
977 break;
978 }
979 }
980 }
981 }
982
983 if let Some(index) = &self.historical_index {
984 let remaining = limit.saturating_sub(events.len());
985 for event in index.query(filter, remaining).await {
986 if seen.insert(event.id) {
987 events.push(event);
988 if events.len() >= limit {
989 break;
990 }
991 }
992 }
993 }
994
995 events
996 }
997
998 pub async fn register_client(
999 &self,
1000 client_id: u64,
1001 sender: mpsc::UnboundedSender<String>,
1002 pubkey: Option<String>,
1003 ) {
1004 let mut clients = self.clients.lock().await;
1005 clients.insert(
1006 client_id,
1007 ClientState {
1008 sender,
1009 pubkey,
1010 quota: ClientQuota::new(),
1011 },
1012 );
1013 }
1014
1015 pub async fn unregister_client(&self, client_id: u64) {
1016 let mut clients = self.clients.lock().await;
1017 clients.remove(&client_id);
1018 drop(clients);
1019 let mut subs = self.subscriptions.lock().await;
1020 subs.remove(&client_id);
1021 }
1022
1023 pub async fn handle_client_message(&self, client_id: u64, msg: NostrClientMessage<'_>) {
1024 match msg {
1025 NostrClientMessage::Event(event) => {
1026 self.handle_event(client_id, event.into_owned()).await;
1027 }
1028 NostrClientMessage::Req {
1029 subscription_id,
1030 filters,
1031 } => {
1032 self.handle_req(
1033 client_id,
1034 subscription_id.into_owned(),
1035 filters
1036 .into_iter()
1037 .map(|filter| filter.into_owned())
1038 .collect(),
1039 )
1040 .await;
1041 }
1042 NostrClientMessage::Count {
1043 subscription_id,
1044 filter,
1045 } => {
1046 self.handle_count(
1047 client_id,
1048 subscription_id.into_owned(),
1049 vec![filter.into_owned()],
1050 )
1051 .await;
1052 }
1053 NostrClientMessage::Close(subscription_id) => {
1054 self.handle_close(client_id, subscription_id.into_owned())
1055 .await;
1056 }
1057 NostrClientMessage::Auth(event) => {
1058 self.handle_auth(client_id, event.into_owned()).await;
1059 }
1060 NostrClientMessage::NegOpen { .. }
1061 | NostrClientMessage::NegMsg { .. }
1062 | NostrClientMessage::NegClose { .. } => {
1063 self.send_to_client(
1064 client_id,
1065 NostrRelayMessage::notice("negentropy not supported"),
1066 )
1067 .await;
1068 }
1069 }
1070 }
1071
1072 pub async fn register_subscription_query(
1073 &self,
1074 client_id: u64,
1075 subscription_id: SubscriptionId,
1076 mut filters: Vec<NostrFilter>,
1077 ) -> std::result::Result<Vec<Event>, &'static str> {
1078 if !self.allow_req(client_id).await {
1079 return Err("rate limited");
1080 }
1081
1082 if filters.len() > self.config.max_filters_per_sub {
1083 filters.truncate(self.config.max_filters_per_sub);
1084 }
1085
1086 {
1087 let mut subs = self.subscriptions.lock().await;
1088 let entry = subs.entry(client_id).or_default();
1089 if !entry.contains_key(&subscription_id)
1090 && entry.len() >= self.config.max_subs_per_client
1091 {
1092 return Err("too many subscriptions");
1093 }
1094 entry.insert(subscription_id.clone(), filters.clone());
1095 }
1096
1097 let mut seen: HashSet<EventId> = HashSet::new();
1098 let mut events = Vec::new();
1099 let memory_before = process_memory_snapshot();
1100 let started = Instant::now();
1101 let filter_summary = nostr_filters_summary(&filters);
1102 for filter in &filters {
1103 let remaining = self.config.max_query_limit.saturating_sub(events.len());
1104 if remaining == 0 {
1105 break;
1106 }
1107 let limit = filter
1108 .limit
1109 .unwrap_or(self.config.max_query_limit)
1110 .min(self.config.max_query_limit)
1111 .min(remaining);
1112 self.collect_filter_events(filter, limit, &mut seen, &mut events)
1113 .await;
1114 }
1115
1116 info!(
1117 target: "hashtree_cli::nostr_relay::query",
1118 client_id,
1119 subscription_id = %subscription_id,
1120 filters = filters.len(),
1121 events = events.len(),
1122 elapsed_ms = started.elapsed().as_millis() as u64,
1123 filter = %filter_summary,
1124 memory_before = ?memory_before,
1125 memory_after = ?process_memory_snapshot(),
1126 "nostr relay local subscription query completed",
1127 );
1128 Ok(events)
1129 }
1130
1131 async fn handle_auth(&self, client_id: u64, event: Event) {
1132 let ok = event.verify().is_ok();
1133 let message = if ok { "" } else { "invalid auth" };
1134 self.send_to_client(client_id, NostrRelayMessage::ok(event.id, ok, message))
1135 .await;
1136 }
1137
1138 async fn handle_close(&self, client_id: u64, subscription_id: SubscriptionId) {
1139 let mut subs = self.subscriptions.lock().await;
1140 if let Some(map) = subs.get_mut(&client_id) {
1141 map.remove(&subscription_id);
1142 }
1143 }
1144
1145 async fn handle_event(&self, client_id: u64, event: Event) {
1146 let ok = event.verify().is_ok();
1147 if !ok {
1148 self.send_to_client(
1149 client_id,
1150 NostrRelayMessage::ok(event.id, false, "invalid: signature"),
1151 )
1152 .await;
1153 return;
1154 }
1155
1156 if self.trusted.is_none() {
1157 self.send_to_client(
1158 client_id,
1159 NostrRelayMessage::ok(event.id, false, "read-only"),
1160 )
1161 .await;
1162 return;
1163 }
1164
1165 let trusted = self.is_trusted_event(client_id, &event).await;
1166 if !trusted && !self.allow_spambox_event(client_id).await {
1167 self.send_to_client(
1168 client_id,
1169 NostrRelayMessage::ok(event.id, false, "rate limited"),
1170 )
1171 .await;
1172 return;
1173 }
1174
1175 let is_ephemeral = event.kind.is_ephemeral();
1176 if trusted {
1177 let mut recent = self.recent_events.lock().await;
1178 recent.insert(event.clone());
1179 }
1180 if !is_ephemeral {
1181 let stored = if trusted {
1182 let storage_class = self.event_storage_class(&event);
1183 let stored = self
1184 .trusted
1185 .as_ref()
1186 .expect("read-only relay rejected client publication")
1187 .ingest_with_storage_class(event.clone(), storage_class)
1188 .await
1189 .is_ok();
1190 if stored {
1191 self.append_historical_index(&event).await;
1192 }
1193 stored
1194 } else {
1195 match self.spambox.as_ref() {
1196 Some(spambox) => spambox.ingest(&event).await,
1197 None => false,
1198 }
1199 };
1200
1201 if !stored {
1202 let message = if trusted {
1203 "store failed"
1204 } else {
1205 "spambox full"
1206 };
1207 self.send_to_client(client_id, NostrRelayMessage::ok(event.id, false, message))
1208 .await;
1209 return;
1210 }
1211 }
1212
1213 let message = if trusted { "" } else { "spambox" };
1214 self.send_to_client(client_id, NostrRelayMessage::ok(event.id, true, message))
1215 .await;
1216
1217 if trusted {
1218 self.broadcast_event(&event).await;
1219 #[cfg(feature = "experimental-decentralized-pubsub")]
1220 self.enqueue_decentralized_pubsub_event(&event);
1221 }
1222 }
1223
1224 async fn handle_req(
1225 &self,
1226 client_id: u64,
1227 subscription_id: SubscriptionId,
1228 filters: Vec<NostrFilter>,
1229 ) {
1230 match self
1231 .register_subscription_query(client_id, subscription_id.clone(), filters)
1232 .await
1233 {
1234 Ok(events) => {
1235 for event in events {
1236 self.send_to_client(
1237 client_id,
1238 NostrRelayMessage::event(subscription_id.clone(), event),
1239 )
1240 .await;
1241 }
1242 trim_process_allocations();
1243
1244 self.send_to_client(client_id, NostrRelayMessage::eose(subscription_id))
1245 .await;
1246 }
1247 Err(message) => {
1248 self.send_to_client(
1249 client_id,
1250 NostrRelayMessage::closed(subscription_id, message),
1251 )
1252 .await;
1253 }
1254 }
1255 }
1256
1257 async fn handle_count(
1258 &self,
1259 client_id: u64,
1260 subscription_id: SubscriptionId,
1261 filters: Vec<NostrFilter>,
1262 ) {
1263 if !self.allow_req(client_id).await {
1264 self.send_to_client(
1265 client_id,
1266 NostrRelayMessage::closed(subscription_id, "rate limited"),
1267 )
1268 .await;
1269 return;
1270 }
1271
1272 let mut seen: HashSet<EventId> = HashSet::new();
1273 for filter in &filters {
1274 let limit = filter
1275 .limit
1276 .unwrap_or(self.config.max_query_limit)
1277 .min(self.config.max_query_limit);
1278 self.collect_filter_count(filter, limit, &mut seen).await;
1279 }
1280
1281 self.send_to_client(
1282 client_id,
1283 NostrRelayMessage::count(subscription_id, seen.len()),
1284 )
1285 .await;
1286 }
1287
1288 async fn is_trusted_event(&self, client_id: u64, event: &Event) -> bool {
1289 self.is_trusted_event_for_client(Some(client_id), event)
1290 .await
1291 }
1292
1293 async fn is_trusted_event_for_client(&self, client_id: Option<u64>, event: &Event) -> bool {
1294 let event_pubkey = event.pubkey.to_hex();
1295 let client_pubkey = {
1296 let clients = self.clients.lock().await;
1297 client_id.and_then(|client_id| {
1298 clients
1299 .get(&client_id)
1300 .and_then(|state| state.pubkey.clone())
1301 })
1302 };
1303 if let Some(pubkey) = client_pubkey {
1304 return pubkey == event_pubkey
1305 || self.social_graph.as_ref().is_some_and(|social_graph| {
1306 social_graph.check_write_access(&event_pubkey)
1307 });
1308 }
1309 if let Some(ref social_graph) = self.social_graph {
1310 return social_graph.check_write_access(&event_pubkey);
1311 }
1312 true
1313 }
1314
1315 async fn append_historical_index(&self, event: &Event) {
1316 if let Some(index) = &self.historical_index {
1317 if let Err(err) = index.ingest(event.clone()).await {
1318 warn!("historical nostr index ingest failed: {}", err);
1319 }
1320 }
1321 }
1322
1323 fn event_storage_class(&self, event: &Event) -> EventStorageClass {
1324 if self.public_pubkeys.contains(&event.pubkey.to_hex()) {
1325 EventStorageClass::Public
1326 } else {
1327 EventStorageClass::Ambient
1328 }
1329 }
1330
1331 async fn allow_spambox_event(&self, client_id: u64) -> bool {
1332 let mut clients = self.clients.lock().await;
1333 let Some(state) = clients.get_mut(&client_id) else {
1334 return false;
1335 };
1336 state
1337 .quota
1338 .allow_spambox_event(self.config.spambox_max_events_per_min)
1339 }
1340
1341 async fn allow_req(&self, client_id: u64) -> bool {
1342 let mut clients = self.clients.lock().await;
1343 let Some(state) = clients.get_mut(&client_id) else {
1344 return false;
1345 };
1346 state.quota.allow_req(self.config.spambox_max_reqs_per_min)
1347 }
1348
1349 async fn broadcast_event(&self, event: &Event) {
1350 let subscriptions = self.subscriptions.lock().await;
1351 let mut deliveries: Vec<(u64, SubscriptionId)> = Vec::new();
1352 for (client_id, subs) in subscriptions.iter() {
1353 for (sub_id, filters) in subs.iter() {
1354 if filters
1355 .iter()
1356 .any(|f| f.match_event(event, Default::default()))
1357 {
1358 deliveries.push((*client_id, sub_id.clone()));
1359 }
1360 }
1361 }
1362 drop(subscriptions);
1363
1364 for (client_id, sub_id) in deliveries {
1365 self.send_to_client(client_id, NostrRelayMessage::event(sub_id, event.clone()))
1366 .await;
1367 }
1368 }
1369
1370 async fn send_to_client(&self, client_id: u64, msg: NostrRelayMessage<'_>) {
1371 let sender = {
1372 let clients = self.clients.lock().await;
1373 clients.get(&client_id).map(|state| state.sender.clone())
1374 };
1375 if let Some(tx) = sender {
1376 let _ = tx.send(msg.as_json());
1377 }
1378 }
1379 }
1380}
1381
1382pub use imp::NostrRelay;
1383
1384#[cfg(test)]
1385#[path = "nostr_relay/tests.rs"]
1386mod tests;