1use std::collections::HashSet;
4use std::sync::Arc;
5
6use async_trait::async_trait;
7use hashtree_core::{Cid, Store};
8use hashtree_nostr::{
9 stored_event_from_nostr_sdk_event, NostrEventStore, StoredNostrEvent, VerifiedStoredNostrEvent,
10};
11use nostr_pubsub::{
12 EventBus, EventRetentionPolicy, EventSource, PublishReport, PubsubError, QueryEvent,
13 QueryOptions, QueryReport, Result, SourceRoute, VerifiedEvent, CAP_HASHTREE_FETCH,
14};
15use tokio::sync::Mutex;
16
17#[derive(Clone)]
18pub struct HashtreeNostrIndexEventBus<S> {
19 store: Arc<S>,
20 root: Option<Cid>,
21 source: EventSource,
22 priority: i32,
23}
24
25impl<S> HashtreeNostrIndexEventBus<S> {
26 pub fn new(store: Arc<S>, root: Option<Cid>, source: EventSource) -> Self {
27 Self {
28 store,
29 root,
30 source,
31 priority: 0,
32 }
33 }
34
35 pub fn with_priority(mut self, priority: i32) -> Self {
36 self.priority = priority;
37 self
38 }
39
40 pub fn with_root(mut self, root: Option<Cid>) -> Self {
41 self.root = root;
42 self
43 }
44
45 pub fn root(&self) -> Option<&Cid> {
46 self.root.as_ref()
47 }
48
49 pub fn source(&self) -> &EventSource {
50 &self.source
51 }
52
53 pub fn source_route(&self, dataset_id: impl Into<String>) -> Result<SourceRoute> {
54 SourceRoute::from_source(self.source.clone())
55 .with_priority(self.priority)
56 .with_capability(CAP_HASHTREE_FETCH)
57 .with_dataset(dataset_id)
58 }
59}
60
61#[async_trait]
62impl<S> EventBus for HashtreeNostrIndexEventBus<S>
63where
64 S: Store + 'static,
65{
66 async fn publish(&self, _event: VerifiedEvent, _source: EventSource) -> Result<PublishReport> {
67 Err(PubsubError::Validation(
68 "hashtree nostr index event bus is read-only".to_string(),
69 ))
70 }
71
72 async fn query(
73 &self,
74 filters: Vec<nostr_pubsub::Filter>,
75 options: QueryOptions,
76 ) -> Result<QueryReport> {
77 query_index(
78 Arc::clone(&self.store),
79 self.root.clone(),
80 self.source.clone(),
81 self.priority,
82 filters,
83 options,
84 )
85 .await
86 }
87}
88
89#[derive(Clone)]
90pub struct HashtreeNostrBoundedEventCache<S> {
91 store: Arc<S>,
92 root: Arc<Mutex<Option<Cid>>>,
93 source: EventSource,
94 priority: i32,
95 retention: EventRetentionPolicy,
96}
97
98impl<S> HashtreeNostrBoundedEventCache<S> {
99 pub fn new(
100 store: Arc<S>,
101 root: Option<Cid>,
102 source: EventSource,
103 retention: EventRetentionPolicy,
104 ) -> Self {
105 Self {
106 store,
107 root: Arc::new(Mutex::new(root)),
108 source,
109 priority: 0,
110 retention,
111 }
112 }
113
114 pub fn with_priority(mut self, priority: i32) -> Self {
115 self.priority = priority;
116 self
117 }
118
119 pub async fn root_cid(&self) -> Option<Cid> {
120 self.root.lock().await.clone()
121 }
122
123 pub fn source(&self) -> &EventSource {
124 &self.source
125 }
126
127 pub fn retention(&self) -> &EventRetentionPolicy {
128 &self.retention
129 }
130
131 pub fn source_route(&self, dataset_id: impl Into<String>) -> Result<SourceRoute> {
132 SourceRoute::from_source(self.source.clone())
133 .with_priority(self.priority)
134 .with_capability(CAP_HASHTREE_FETCH)
135 .with_dataset(dataset_id)
136 }
137}
138
139#[async_trait]
140impl<S> EventBus for HashtreeNostrBoundedEventCache<S>
141where
142 S: Store + 'static,
143{
144 async fn publish(&self, event: VerifiedEvent, _source: EventSource) -> Result<PublishReport> {
145 if !self.retention.accepts(&event) {
146 return Ok(PublishReport {
147 accepted: false,
148 priority: 0,
149 reason: Some("event outside retention policy".to_string()),
150 });
151 }
152
153 let stored_event = stored_event_from_nostr_sdk_event(event.as_event());
154 let mut root = self.root.lock().await;
155 let next_root = append_bounded_index_event(
156 Arc::clone(&self.store),
157 root.clone(),
158 self.retention.clone(),
159 stored_event,
160 )
161 .await?;
162 *root = next_root;
163
164 Ok(PublishReport {
165 accepted: true,
166 priority: self.priority,
167 reason: None,
168 })
169 }
170
171 async fn query(
172 &self,
173 filters: Vec<nostr_pubsub::Filter>,
174 options: QueryOptions,
175 ) -> Result<QueryReport> {
176 let root = self.root.lock().await.clone();
177 query_index(
178 Arc::clone(&self.store),
179 root,
180 self.source.clone(),
181 self.priority,
182 filters,
183 options,
184 )
185 .await
186 }
187}
188
189async fn append_bounded_index_event<S>(
190 store: Arc<S>,
191 root: Option<Cid>,
192 retention: EventRetentionPolicy,
193 event: StoredNostrEvent,
194) -> Result<Option<Cid>>
195where
196 S: Store + 'static,
197{
198 tokio::task::spawn_blocking(move || {
199 let runtime = tokio::runtime::Builder::new_current_thread()
200 .enable_all()
201 .build()
202 .map_err(|error| {
203 PubsubError::Storage(format!("build hashtree cache runtime: {error}"))
204 })?;
205 runtime
206 .block_on(async move {
207 let event_store = NostrEventStore::new(store);
208 let mut seen = HashSet::new();
209 let mut retained = Vec::new();
210 let filters = if retention.filters.is_empty() {
211 vec![nostr_pubsub::Filter::new()]
212 } else {
213 retention.filters
214 };
215 for filter in filters {
216 for stored in event_store
217 .query_events(root.as_ref(), &filter, retention.max_events)
218 .await?
219 {
220 if seen.insert(stored.id.clone()) {
221 retained.push(stored);
222 }
223 }
224 }
225
226 retained.retain(|stored| stored.id != event.id);
227 retained.push(event);
228 retained.sort_by(|left, right| {
229 right
230 .created_at
231 .cmp(&left.created_at)
232 .then_with(|| left.id.cmp(&right.id))
233 });
234 retained.truncate(retention.max_events);
235 event_store.build(None, retained).await
236 })
237 .map_err(|error| PubsubError::Storage(format!("write hashtree nostr cache: {error}")))
238 })
239 .await
240 .map_err(|error| PubsubError::Storage(format!("join hashtree nostr cache write: {error}")))?
241}
242
243async fn query_index<S>(
244 store: Arc<S>,
245 root: Option<Cid>,
246 source: EventSource,
247 priority: i32,
248 filters: Vec<nostr_pubsub::Filter>,
249 options: QueryOptions,
250) -> Result<QueryReport>
251where
252 S: Store + 'static,
253{
254 let limit = query_limit(&filters, options);
255 if limit == 0 {
256 return Ok(QueryReport::default());
257 }
258
259 let filters = if filters.is_empty() {
260 vec![nostr_pubsub::Filter::new()]
261 } else {
262 filters
263 };
264
265 let mut seen = HashSet::new();
266 let mut events = Vec::new();
267 for filter in filters {
268 let remaining = limit.saturating_sub(events.len());
269 if remaining == 0 {
270 break;
271 }
272 let stored_events =
273 query_index_filter(Arc::clone(&store), root.clone(), filter, remaining).await?;
274 for stored in stored_events {
275 if !seen.insert(stored.id.clone()) {
276 continue;
277 }
278 let verified = VerifiedStoredNostrEvent::try_from(stored).map_err(|error| {
279 PubsubError::Validation(format!("verify stored nostr event: {error}"))
280 })?;
281 let event = verified
282 .to_nostr_sdk_event()
283 .map_err(|error| {
284 PubsubError::Validation(format!("decode stored nostr event: {error}"))
285 })?
286 .into_event();
287 let event = VerifiedEvent::try_from(event)
288 .map_err(|error| PubsubError::Validation(format!("verify nostr event: {error}")))?;
289 events.push(QueryEvent {
290 event,
291 source: source.clone(),
292 priority,
293 });
294 if events.len() >= limit {
295 break;
296 }
297 }
298 }
299
300 Ok(QueryReport { events })
301}
302
303async fn query_index_filter<S>(
304 store: Arc<S>,
305 root: Option<Cid>,
306 filter: nostr_pubsub::Filter,
307 limit: usize,
308) -> Result<Vec<StoredNostrEvent>>
309where
310 S: Store + 'static,
311{
312 tokio::task::spawn_blocking(move || {
313 let runtime = tokio::runtime::Builder::new_current_thread()
314 .enable_all()
315 .build()
316 .map_err(|error| {
317 PubsubError::Storage(format!("build hashtree query runtime: {error}"))
318 })?;
319 runtime
320 .block_on(async move {
321 NostrEventStore::new(store)
322 .query_events(root.as_ref(), &filter, limit)
323 .await
324 })
325 .map_err(|error| PubsubError::Storage(format!("query hashtree nostr index: {error}")))
326 })
327 .await
328 .map_err(|error| PubsubError::Storage(format!("join hashtree nostr index query: {error}")))?
329}
330
331fn query_limit(filters: &[nostr_pubsub::Filter], options: QueryOptions) -> usize {
332 options
333 .limit
334 .or_else(|| filters.iter().filter_map(|filter| filter.limit).min())
335 .unwrap_or(usize::MAX)
336}
337
338#[cfg(test)]
339mod tests {
340 use std::sync::Arc;
341
342 use hashtree_core::MemoryStore;
343 use hashtree_nostr::{stored_event_from_nostr_sdk_event, NostrEventStore};
344 use nostr_pubsub::{
345 EventBus, EventRetentionPolicy, EventSource, QueryOptions, VerifiedEvent,
346 CAP_HASHTREE_FETCH,
347 };
348 use nostr_sdk::{EventBuilder, Filter, Keys, Kind, Timestamp};
349
350 use super::{HashtreeNostrBoundedEventCache, HashtreeNostrIndexEventBus};
351
352 #[test]
353 fn source_routes_identify_hashtree_datasets_and_capability() {
354 let source = EventSource::local_index("hashtree-cache");
355 let bus = HashtreeNostrBoundedEventCache::new(
356 Arc::new(MemoryStore::new()),
357 None,
358 source.clone(),
359 EventRetentionPolicy::new(4, Vec::new()),
360 )
361 .with_priority(15);
362
363 let route = bus.source_route("account-events").expect("valid route");
364
365 assert_eq!(route.source, source);
366 assert_eq!(route.dataset_id, "account-events");
367 assert_eq!(route.priority, 15);
368 assert_eq!(route.capabilities, vec![CAP_HASHTREE_FETCH]);
369 assert!(bus.source_route("").is_err());
370 }
371
372 #[tokio::test]
373 async fn event_bus_queries_historical_index_with_normal_nostr_filter() {
374 let backing = Arc::new(MemoryStore::new());
375 let event_store = NostrEventStore::new(Arc::clone(&backing));
376 let author = Keys::generate();
377 let other = Keys::generate();
378 let wanted = EventBuilder::text_note("historical hello")
379 .custom_created_at(Timestamp::from(20))
380 .sign_with_keys(&author)
381 .expect("sign wanted event");
382 let ignored = EventBuilder::text_note("wrong author")
383 .custom_created_at(Timestamp::from(30))
384 .sign_with_keys(&other)
385 .expect("sign ignored event");
386 let root = event_store
387 .build(
388 None,
389 [wanted.clone(), ignored]
390 .iter()
391 .map(stored_event_from_nostr_sdk_event),
392 )
393 .await
394 .expect("build nostr index")
395 .expect("index root");
396
397 let bus = HashtreeNostrIndexEventBus::new(
398 Arc::clone(&backing),
399 Some(root),
400 EventSource::peer("hashtree-index"),
401 )
402 .with_priority(25);
403 let report = bus
404 .query(
405 vec![Filter::new()
406 .author(author.public_key())
407 .kind(Kind::TextNote)],
408 QueryOptions { limit: Some(10) },
409 )
410 .await
411 .expect("query bus");
412
413 assert_eq!(report.events.len(), 1);
414 assert_eq!(report.events[0].event.as_event().id, wanted.id);
415 assert_eq!(report.events[0].source, EventSource::peer("hashtree-index"));
416 assert_eq!(report.events[0].priority, 25);
417 }
418
419 #[tokio::test]
420 async fn publish_is_explicitly_rejected_for_read_only_index_bus() {
421 let bus = HashtreeNostrIndexEventBus::new(
422 Arc::new(MemoryStore::new()),
423 None,
424 EventSource::peer("hashtree-index"),
425 );
426 let event = EventBuilder::text_note("not written here")
427 .sign_with_keys(&Keys::generate())
428 .expect("sign event");
429 let event = nostr_pubsub::VerifiedEvent::try_from(event).expect("verify event");
430
431 let error = bus
432 .publish(event, EventSource::peer("writer"))
433 .await
434 .expect_err("index bus should be read-only");
435 assert!(error.to_string().contains("read-only"));
436 }
437
438 #[tokio::test]
439 async fn bounded_cache_publishes_matching_events_into_hashtree_index() {
440 let backing = Arc::new(MemoryStore::new());
441 let author = Keys::generate();
442 let event = EventBuilder::text_note("cached hello")
443 .custom_created_at(Timestamp::from(20))
444 .sign_with_keys(&author)
445 .expect("sign event");
446 let bus = HashtreeNostrBoundedEventCache::new(
447 Arc::clone(&backing),
448 None,
449 EventSource::local_index("hashtree-cache"),
450 EventRetentionPolicy::new(4, vec![Filter::new().kind(Kind::TextNote)]),
451 )
452 .with_priority(15);
453
454 let report = bus
455 .publish(
456 VerifiedEvent::try_from(event.clone()).expect("verify event"),
457 EventSource::peer("writer"),
458 )
459 .await
460 .expect("publish event");
461
462 assert!(report.accepted);
463 assert_eq!(report.priority, 15);
464 assert!(bus.root_cid().await.is_some());
465
466 let query = bus
467 .query(
468 vec![Filter::new()
469 .author(author.public_key())
470 .kind(Kind::TextNote)],
471 QueryOptions { limit: Some(10) },
472 )
473 .await
474 .expect("query event");
475
476 assert_eq!(query.events.len(), 1);
477 assert_eq!(query.events[0].event.as_event().id, event.id);
478 assert_eq!(
479 query.events[0].source,
480 EventSource::local_index("hashtree-cache")
481 );
482 assert_eq!(query.events[0].priority, 15);
483 }
484
485 #[tokio::test]
486 async fn bounded_cache_rejects_events_outside_retention_policy() {
487 let bus = HashtreeNostrBoundedEventCache::new(
488 Arc::new(MemoryStore::new()),
489 None,
490 EventSource::local_index("hashtree-cache"),
491 EventRetentionPolicy::new(4, vec![Filter::new().kind(Kind::TextNote)]),
492 );
493 let event = EventBuilder::new(Kind::Metadata, "{}")
494 .sign_with_keys(&Keys::generate())
495 .expect("sign event");
496
497 let report = bus
498 .publish(
499 VerifiedEvent::try_from(event).expect("verify event"),
500 EventSource::peer("writer"),
501 )
502 .await
503 .expect("publish event");
504 let query = bus
505 .query(vec![Filter::new()], QueryOptions { limit: Some(10) })
506 .await
507 .expect("query cache");
508
509 assert!(!report.accepted);
510 assert_eq!(
511 report.reason.as_deref(),
512 Some("event outside retention policy")
513 );
514 assert!(bus.root_cid().await.is_none());
515 assert!(query.events.is_empty());
516 }
517
518 #[tokio::test]
519 async fn bounded_cache_keeps_newest_events() {
520 let bus = HashtreeNostrBoundedEventCache::new(
521 Arc::new(MemoryStore::new()),
522 None,
523 EventSource::local_index("hashtree-cache"),
524 EventRetentionPolicy::new(2, vec![Filter::new().kind(Kind::TextNote)]),
525 );
526 let author = Keys::generate();
527 for created_at in [10, 20, 30] {
528 let event = EventBuilder::text_note(format!("event-{created_at}"))
529 .custom_created_at(Timestamp::from(created_at))
530 .sign_with_keys(&author)
531 .expect("sign event");
532 bus.publish(
533 VerifiedEvent::try_from(event).expect("verify event"),
534 EventSource::peer("writer"),
535 )
536 .await
537 .expect("publish event");
538 }
539
540 let query = bus
541 .query(
542 vec![Filter::new().kind(Kind::TextNote)],
543 QueryOptions { limit: Some(10) },
544 )
545 .await
546 .expect("query cache");
547 let contents = query
548 .events
549 .iter()
550 .map(|event| event.event.as_event().content.as_str())
551 .collect::<Vec<_>>();
552
553 assert_eq!(contents, vec!["event-30", "event-20"]);
554 }
555
556 #[tokio::test]
557 async fn bounded_cache_keeps_lowest_id_same_second_replaceable_event() {
558 let bus = HashtreeNostrBoundedEventCache::new(
559 Arc::new(MemoryStore::new()),
560 None,
561 EventSource::local_index("hashtree-cache"),
562 EventRetentionPolicy::new(1, vec![Filter::new().kind(Kind::Metadata)]),
563 );
564 let author = Keys::generate();
565 let first = EventBuilder::new(Kind::Metadata, r#"{"name":"first"}"#)
566 .custom_created_at(Timestamp::from(20))
567 .sign_with_keys(&author)
568 .expect("sign first event");
569 let second = EventBuilder::new(Kind::Metadata, r#"{"name":"second"}"#)
570 .custom_created_at(Timestamp::from(20))
571 .sign_with_keys(&author)
572 .expect("sign second event");
573 let (low, high) = if first.id < second.id {
574 (first, second)
575 } else {
576 (second, first)
577 };
578
579 for event in [high, low.clone()] {
580 bus.publish(
581 VerifiedEvent::try_from(event).expect("verify event"),
582 EventSource::peer("writer"),
583 )
584 .await
585 .expect("publish event");
586 }
587
588 let query = bus
589 .query(
590 vec![Filter::new().kind(Kind::Metadata)],
591 QueryOptions { limit: Some(10) },
592 )
593 .await
594 .expect("query cache");
595
596 assert_eq!(query.events.len(), 1);
597 assert_eq!(query.events[0].event.as_event().id, low.id);
598 }
599}