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 if !is_ephemeral {
898 let storage_class = self.event_storage_class(&event);
899 if let Some(trusted) = self.trusted.as_ref() {
900 trusted
901 .ingest_with_storage_class(event.clone(), storage_class)
902 .await?;
903 self.append_historical_index(&event).await;
904 }
905 }
906 self.recent_events.lock().await.insert(event);
907
908 Ok(true)
909 }
910
911 pub async fn bluetooth_received_events(
912 &self,
913 limit: usize,
914 ) -> Vec<BluetoothReceivedEventRecord> {
915 self.bluetooth_event_log.recent(limit).await
916 }
917
918 async fn ingest_trusted_event_inner(&self, event: Event, broadcast: bool) -> Result<()> {
919 event
920 .verify()
921 .map_err(|e| anyhow::anyhow!("invalid signature: {}", e))?;
922
923 let is_ephemeral = event.kind.is_ephemeral();
924 if !is_ephemeral {
925 let storage_class = self.event_storage_class(&event);
926 if let Some(trusted) = self.trusted.as_ref() {
927 trusted
928 .ingest_with_storage_class(event.clone(), storage_class)
929 .await?;
930 self.append_historical_index(&event).await;
931 }
932 }
933 self.recent_events.lock().await.insert(event.clone());
934
935 if broadcast {
936 self.broadcast_event(&event).await;
937 }
938 Ok(())
939 }
940
941 pub async fn query_events(&self, filter: &NostrFilter, limit: usize) -> Vec<Event> {
942 let limit = limit.min(self.config.max_query_limit);
943 if limit == 0 {
944 return Vec::new();
945 }
946
947 let mut seen: HashSet<EventId> = HashSet::new();
948 let mut events = Vec::new();
949
950 if self.trusted.is_none() || !prefers_trusted_only(filter) {
951 let recent = {
952 let cache = self.recent_events.lock().await;
953 cache.matching(filter)
954 };
955 for event in recent {
956 if seen.insert(event.id) {
957 events.push(event);
958 if events.len() >= limit {
959 return events;
960 }
961 }
962 }
963 }
964
965 if let Some(trusted) = self.trusted.as_ref() {
966 for event in trusted.query(filter.clone(), limit).await {
967 if seen.insert(event.id) {
968 events.push(event);
969 if events.len() >= limit {
970 break;
971 }
972 }
973 }
974 }
975
976 if let Some(index) = &self.historical_index {
977 let remaining = limit.saturating_sub(events.len());
978 for event in index.query(filter, remaining).await {
979 if seen.insert(event.id) {
980 events.push(event);
981 if events.len() >= limit {
982 break;
983 }
984 }
985 }
986 }
987
988 events
989 }
990
991 pub async fn register_client(
992 &self,
993 client_id: u64,
994 sender: mpsc::UnboundedSender<String>,
995 pubkey: Option<String>,
996 ) {
997 let mut clients = self.clients.lock().await;
998 clients.insert(
999 client_id,
1000 ClientState {
1001 sender,
1002 pubkey,
1003 quota: ClientQuota::new(),
1004 },
1005 );
1006 }
1007
1008 pub async fn unregister_client(&self, client_id: u64) {
1009 let mut clients = self.clients.lock().await;
1010 clients.remove(&client_id);
1011 drop(clients);
1012 let mut subs = self.subscriptions.lock().await;
1013 subs.remove(&client_id);
1014 }
1015
1016 pub async fn handle_client_message(&self, client_id: u64, msg: NostrClientMessage<'_>) {
1017 match msg {
1018 NostrClientMessage::Event(event) => {
1019 self.handle_event(client_id, event.into_owned()).await;
1020 }
1021 NostrClientMessage::Req {
1022 subscription_id,
1023 filters,
1024 } => {
1025 self.handle_req(
1026 client_id,
1027 subscription_id.into_owned(),
1028 filters
1029 .into_iter()
1030 .map(|filter| filter.into_owned())
1031 .collect(),
1032 )
1033 .await;
1034 }
1035 NostrClientMessage::Count {
1036 subscription_id,
1037 filter,
1038 } => {
1039 self.handle_count(
1040 client_id,
1041 subscription_id.into_owned(),
1042 vec![filter.into_owned()],
1043 )
1044 .await;
1045 }
1046 NostrClientMessage::Close(subscription_id) => {
1047 self.handle_close(client_id, subscription_id.into_owned())
1048 .await;
1049 }
1050 NostrClientMessage::Auth(event) => {
1051 self.handle_auth(client_id, event.into_owned()).await;
1052 }
1053 NostrClientMessage::NegOpen { .. }
1054 | NostrClientMessage::NegMsg { .. }
1055 | NostrClientMessage::NegClose { .. } => {
1056 self.send_to_client(
1057 client_id,
1058 NostrRelayMessage::notice("negentropy not supported"),
1059 )
1060 .await;
1061 }
1062 }
1063 }
1064
1065 pub async fn register_subscription_query(
1066 &self,
1067 client_id: u64,
1068 subscription_id: SubscriptionId,
1069 mut filters: Vec<NostrFilter>,
1070 ) -> std::result::Result<Vec<Event>, &'static str> {
1071 if !self.allow_req(client_id).await {
1072 return Err("rate limited");
1073 }
1074
1075 if filters.len() > self.config.max_filters_per_sub {
1076 filters.truncate(self.config.max_filters_per_sub);
1077 }
1078
1079 {
1080 let mut subs = self.subscriptions.lock().await;
1081 let entry = subs.entry(client_id).or_default();
1082 if !entry.contains_key(&subscription_id)
1083 && entry.len() >= self.config.max_subs_per_client
1084 {
1085 return Err("too many subscriptions");
1086 }
1087 entry.insert(subscription_id.clone(), filters.clone());
1088 }
1089
1090 let mut seen: HashSet<EventId> = HashSet::new();
1091 let mut events = Vec::new();
1092 let memory_before = process_memory_snapshot();
1093 let started = Instant::now();
1094 let filter_summary = nostr_filters_summary(&filters);
1095 for filter in &filters {
1096 let remaining = self.config.max_query_limit.saturating_sub(events.len());
1097 if remaining == 0 {
1098 break;
1099 }
1100 let limit = filter
1101 .limit
1102 .unwrap_or(self.config.max_query_limit)
1103 .min(self.config.max_query_limit)
1104 .min(remaining);
1105 self.collect_filter_events(filter, limit, &mut seen, &mut events)
1106 .await;
1107 }
1108
1109 info!(
1110 target: "hashtree_cli::nostr_relay::query",
1111 client_id,
1112 subscription_id = %subscription_id,
1113 filters = filters.len(),
1114 events = events.len(),
1115 elapsed_ms = started.elapsed().as_millis() as u64,
1116 filter = %filter_summary,
1117 memory_before = ?memory_before,
1118 memory_after = ?process_memory_snapshot(),
1119 "nostr relay local subscription query completed",
1120 );
1121 Ok(events)
1122 }
1123
1124 async fn handle_auth(&self, client_id: u64, event: Event) {
1125 let ok = event.verify().is_ok();
1126 let message = if ok { "" } else { "invalid auth" };
1127 self.send_to_client(client_id, NostrRelayMessage::ok(event.id, ok, message))
1128 .await;
1129 }
1130
1131 async fn handle_close(&self, client_id: u64, subscription_id: SubscriptionId) {
1132 let mut subs = self.subscriptions.lock().await;
1133 if let Some(map) = subs.get_mut(&client_id) {
1134 map.remove(&subscription_id);
1135 }
1136 }
1137
1138 async fn handle_event(&self, client_id: u64, event: Event) {
1139 let ok = event.verify().is_ok();
1140 if !ok {
1141 self.send_to_client(
1142 client_id,
1143 NostrRelayMessage::ok(event.id, false, "invalid: signature"),
1144 )
1145 .await;
1146 return;
1147 }
1148
1149 if self.trusted.is_none() {
1150 self.send_to_client(
1151 client_id,
1152 NostrRelayMessage::ok(event.id, false, "read-only"),
1153 )
1154 .await;
1155 return;
1156 }
1157
1158 let trusted = self.is_trusted_event(client_id, &event).await;
1159 if !trusted && !self.allow_spambox_event(client_id).await {
1160 self.send_to_client(
1161 client_id,
1162 NostrRelayMessage::ok(event.id, false, "rate limited"),
1163 )
1164 .await;
1165 return;
1166 }
1167
1168 let is_ephemeral = event.kind.is_ephemeral();
1169 if !is_ephemeral {
1170 let stored = if trusted {
1171 let storage_class = self.event_storage_class(&event);
1172 let stored = self
1173 .trusted
1174 .as_ref()
1175 .expect("read-only relay rejected client publication")
1176 .ingest_with_storage_class(event.clone(), storage_class)
1177 .await
1178 .is_ok();
1179 if stored {
1180 self.append_historical_index(&event).await;
1181 }
1182 stored
1183 } else {
1184 match self.spambox.as_ref() {
1185 Some(spambox) => spambox.ingest(&event).await,
1186 None => false,
1187 }
1188 };
1189
1190 if !stored {
1191 let message = if trusted {
1192 "store failed"
1193 } else {
1194 "spambox full"
1195 };
1196 self.send_to_client(client_id, NostrRelayMessage::ok(event.id, false, message))
1197 .await;
1198 return;
1199 }
1200 }
1201 if trusted {
1202 self.recent_events.lock().await.insert(event.clone());
1203 }
1204
1205 let message = if trusted { "" } else { "spambox" };
1206 self.send_to_client(client_id, NostrRelayMessage::ok(event.id, true, message))
1207 .await;
1208
1209 if trusted {
1210 self.broadcast_event(&event).await;
1211 #[cfg(feature = "experimental-decentralized-pubsub")]
1212 self.enqueue_decentralized_pubsub_event(&event);
1213 }
1214 }
1215
1216 async fn handle_req(
1217 &self,
1218 client_id: u64,
1219 subscription_id: SubscriptionId,
1220 filters: Vec<NostrFilter>,
1221 ) {
1222 match self
1223 .register_subscription_query(client_id, subscription_id.clone(), filters)
1224 .await
1225 {
1226 Ok(events) => {
1227 for event in events {
1228 self.send_to_client(
1229 client_id,
1230 NostrRelayMessage::event(subscription_id.clone(), event),
1231 )
1232 .await;
1233 }
1234 trim_process_allocations();
1235
1236 self.send_to_client(client_id, NostrRelayMessage::eose(subscription_id))
1237 .await;
1238 }
1239 Err(message) => {
1240 self.send_to_client(
1241 client_id,
1242 NostrRelayMessage::closed(subscription_id, message),
1243 )
1244 .await;
1245 }
1246 }
1247 }
1248
1249 async fn handle_count(
1250 &self,
1251 client_id: u64,
1252 subscription_id: SubscriptionId,
1253 filters: Vec<NostrFilter>,
1254 ) {
1255 if !self.allow_req(client_id).await {
1256 self.send_to_client(
1257 client_id,
1258 NostrRelayMessage::closed(subscription_id, "rate limited"),
1259 )
1260 .await;
1261 return;
1262 }
1263
1264 let mut seen: HashSet<EventId> = HashSet::new();
1265 for filter in &filters {
1266 let limit = filter
1267 .limit
1268 .unwrap_or(self.config.max_query_limit)
1269 .min(self.config.max_query_limit);
1270 self.collect_filter_count(filter, limit, &mut seen).await;
1271 }
1272
1273 self.send_to_client(
1274 client_id,
1275 NostrRelayMessage::count(subscription_id, seen.len()),
1276 )
1277 .await;
1278 }
1279
1280 async fn is_trusted_event(&self, client_id: u64, event: &Event) -> bool {
1281 self.is_trusted_event_for_client(Some(client_id), event)
1282 .await
1283 }
1284
1285 async fn is_trusted_event_for_client(&self, client_id: Option<u64>, event: &Event) -> bool {
1286 let event_pubkey = event.pubkey.to_hex();
1287 let client_pubkey = {
1288 let clients = self.clients.lock().await;
1289 client_id.and_then(|client_id| {
1290 clients
1291 .get(&client_id)
1292 .and_then(|state| state.pubkey.clone())
1293 })
1294 };
1295 if let Some(pubkey) = client_pubkey {
1296 return pubkey == event_pubkey
1297 || self.social_graph.as_ref().is_some_and(|social_graph| {
1298 social_graph.check_write_access(&event_pubkey)
1299 });
1300 }
1301 if let Some(ref social_graph) = self.social_graph {
1302 return social_graph.check_write_access(&event_pubkey);
1303 }
1304 true
1305 }
1306
1307 async fn append_historical_index(&self, event: &Event) {
1308 if let Some(index) = &self.historical_index {
1309 if let Err(err) = index.ingest(event.clone()).await {
1310 warn!("historical nostr index ingest failed: {}", err);
1311 }
1312 }
1313 }
1314
1315 fn event_storage_class(&self, event: &Event) -> EventStorageClass {
1316 if self.public_pubkeys.contains(&event.pubkey.to_hex()) {
1317 EventStorageClass::Public
1318 } else {
1319 EventStorageClass::Ambient
1320 }
1321 }
1322
1323 async fn allow_spambox_event(&self, client_id: u64) -> bool {
1324 let mut clients = self.clients.lock().await;
1325 let Some(state) = clients.get_mut(&client_id) else {
1326 return false;
1327 };
1328 state
1329 .quota
1330 .allow_spambox_event(self.config.spambox_max_events_per_min)
1331 }
1332
1333 async fn allow_req(&self, client_id: u64) -> bool {
1334 let mut clients = self.clients.lock().await;
1335 let Some(state) = clients.get_mut(&client_id) else {
1336 return false;
1337 };
1338 state.quota.allow_req(self.config.spambox_max_reqs_per_min)
1339 }
1340
1341 async fn broadcast_event(&self, event: &Event) {
1342 let subscriptions = self.subscriptions.lock().await;
1343 let mut deliveries: Vec<(u64, SubscriptionId)> = Vec::new();
1344 for (client_id, subs) in subscriptions.iter() {
1345 for (sub_id, filters) in subs.iter() {
1346 if filters
1347 .iter()
1348 .any(|f| f.match_event(event, Default::default()))
1349 {
1350 deliveries.push((*client_id, sub_id.clone()));
1351 }
1352 }
1353 }
1354 drop(subscriptions);
1355
1356 for (client_id, sub_id) in deliveries {
1357 self.send_to_client(client_id, NostrRelayMessage::event(sub_id, event.clone()))
1358 .await;
1359 }
1360 }
1361
1362 async fn send_to_client(&self, client_id: u64, msg: NostrRelayMessage<'_>) {
1363 let sender = {
1364 let clients = self.clients.lock().await;
1365 clients.get(&client_id).map(|state| state.sender.clone())
1366 };
1367 if let Some(tx) = sender {
1368 let _ = tx.send(msg.as_json());
1369 }
1370 }
1371 }
1372}
1373
1374pub use imp::NostrRelay;
1375
1376#[cfg(test)]
1377#[path = "nostr_relay/tests.rs"]
1378mod tests;