use std::collections::{BTreeMap, BTreeSet, HashSet, VecDeque};
use std::sync::Arc;
use std::time::Duration;
use crate::{ListEventsOptions, NostrEventStore, NostrEventStoreError, StoredNostrEvent};
use futures::{stream, StreamExt};
use hashtree_core::{Cid, Store};
use nostr_sdk::{
pool::RelayLimits, Client, ClientOptions, EventId, Filter, Keys, Kind, PublicKey, SyncOptions,
Timestamp,
};
use nostr_social_graph::SocialGraphBackend;
const NEGENTROPY_FETCH_CHUNK_SIZE: usize = 256;
const NEGENTROPY_FETCH_CHUNK_CONCURRENCY: usize = 16;
const NEGENTROPY_INITIAL_TIMEOUT: Duration = Duration::from_secs(1);
const FULL_HISTORY_PAGING_CONCURRENCY_PER_RELAY: usize = 64;
const METADATA_KIND: u32 = 0;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RelayFetchMode {
AuthorBatches,
GlobalRecent,
}
#[derive(Debug, Clone)]
pub struct CrawlConfig {
pub relays: Vec<String>,
pub author_allowlist: Option<Vec<String>>,
pub max_live_bytes: Option<u64>,
pub max_events_seen: Option<usize>,
pub max_authors: Option<usize>,
pub max_follow_distance: Option<u32>,
pub author_batch_size: usize,
pub per_author_event_limit: usize,
pub per_author_live_bytes: Option<u64>,
pub fetch_timeout: Duration,
pub kinds: Option<Vec<u16>>,
pub relay_fetch_mode: RelayFetchMode,
pub require_negentropy: bool,
pub relay_event_max_size: Option<u32>,
pub relay_page_size: usize,
pub max_relay_pages: usize,
pub full_author_history: bool,
}
impl Default for CrawlConfig {
fn default() -> Self {
Self {
relays: Vec::new(),
author_allowlist: None,
max_live_bytes: None,
max_events_seen: None,
max_authors: None,
max_follow_distance: Some(1),
author_batch_size: 64,
per_author_event_limit: 256,
per_author_live_bytes: None,
fetch_timeout: Duration::from_secs(10),
kinds: None,
relay_fetch_mode: RelayFetchMode::AuthorBatches,
require_negentropy: false,
relay_event_max_size: None,
relay_page_size: 1_000,
max_relay_pages: 10,
full_author_history: false,
}
}
}
#[derive(Debug, Clone, Default, PartialEq)]
pub struct CrawlReport {
pub root: Option<Cid>,
pub authors_considered: usize,
pub authors_processed: usize,
pub events_seen: usize,
pub events_selected: usize,
pub live_bytes_selected: u64,
pub applied_events: Vec<StoredNostrEvent>,
}
pub trait EventSelectionPolicy: Send + Sync {
fn priority(&self, event: &StoredNostrEvent) -> i32;
}
#[derive(Debug, Clone)]
pub struct KindPriorityPolicy {
default_priority: i32,
priorities: BTreeMap<u32, i32>,
}
impl Default for KindPriorityPolicy {
fn default() -> Self {
let mut priorities = BTreeMap::new();
priorities.insert(1, 1_000);
priorities.insert(0, 900);
priorities.insert(3, 800);
priorities.insert(10_000, 750);
priorities.insert(6, 600);
priorities.insert(7, 500);
Self {
default_priority: 100,
priorities,
}
}
}
impl KindPriorityPolicy {
pub fn with_priority(mut self, kind: u32, priority: i32) -> Self {
self.priorities.insert(kind, priority);
self
}
}
impl EventSelectionPolicy for KindPriorityPolicy {
fn priority(&self, event: &StoredNostrEvent) -> i32 {
self.priorities
.get(&event.kind)
.copied()
.unwrap_or(self.default_priority)
}
}
#[derive(Debug, thiserror::Error)]
pub enum CrawlError {
#[error("event store error: {0}")]
EventStore(#[from] NostrEventStoreError),
#[error("crawl requires at least one relay")]
MissingRelays,
#[error("per-author event limit must be greater than zero")]
InvalidPerAuthorLimit,
#[error("per-author live byte cap must be greater than zero")]
InvalidPerAuthorLiveBytes,
#[error("author batch size must be greater than zero")]
InvalidAuthorBatchSize,
#[error("relay page size must be greater than zero")]
InvalidRelayPageSize,
#[error("max relay pages must be greater than zero")]
InvalidMaxRelayPages,
#[error("max events seen must be greater than zero")]
InvalidMaxEventsSeen,
#[error("relay event max size must be greater than zero")]
InvalidRelayEventMaxSize,
#[error("nostr error: {0}")]
Nostr(String),
#[error("social graph error: {0}")]
SocialGraph(String),
}
pub type Result<T> = std::result::Result<T, CrawlError>;
#[derive(Debug, Default)]
struct RelayFetchResult {
events_seen: usize,
events: Vec<StoredNostrEvent>,
supports_negentropy: bool,
}
#[derive(Debug, Default)]
struct BatchCrawlReport {
events_seen: usize,
events_selected: usize,
events: Vec<StoredNostrEvent>,
live_bytes_selected: u64,
}
#[derive(Debug, Default)]
struct ProfileBatchReport {
events_seen: usize,
events_by_author: BTreeMap<String, Vec<StoredNostrEvent>>,
}
#[derive(Debug, Default)]
struct GlobalRecentState {
current_root: Option<Cid>,
retained_by_author: BTreeMap<String, Vec<StoredNostrEvent>>,
events_selected: usize,
live_bytes_selected: u64,
}
pub struct NostrBridge<S: Store> {
event_store: NostrEventStore<S>,
config: CrawlConfig,
policy: Arc<dyn EventSelectionPolicy>,
}
impl<S: Store> NostrBridge<S> {
pub fn new(store: Arc<S>, config: CrawlConfig) -> Self {
Self {
event_store: NostrEventStore::new(store),
config,
policy: Arc::new(KindPriorityPolicy::default()),
}
}
pub fn with_policy(mut self, policy: Arc<dyn EventSelectionPolicy>) -> Self {
self.policy = policy;
self
}
pub async fn crawl<G: SocialGraphBackend>(
&self,
graph: &G,
existing_root: Option<&Cid>,
) -> Result<CrawlReport> {
self.crawl_with_progress(graph, existing_root, |_| {}).await
}
pub async fn crawl_with_progress<G, F>(
&self,
graph: &G,
existing_root: Option<&Cid>,
mut on_progress: F,
) -> Result<CrawlReport>
where
G: SocialGraphBackend,
F: FnMut(&CrawlReport),
{
self.validate_config()?;
let authors = self.collect_authors(graph)?;
if authors.is_empty() {
return Ok(CrawlReport::default());
}
let existing_root = self.usable_existing_root(existing_root).await?;
let client = self.connect_client().await?;
let result = if self.config.relay_fetch_mode == RelayFetchMode::AuthorBatches {
self.crawl_author_batches(&client, &authors, existing_root.as_ref(), &mut on_progress)
.await
} else {
let state = self
.load_existing_global_state(existing_root.as_ref(), &authors)
.await?;
let report = self
.crawl_global_recent_incremental(&client, &authors, state, &mut on_progress)
.await?;
on_progress(&report);
Ok(report)
};
let _ = client.disconnect().await;
result
}
async fn usable_existing_root(&self, root: Option<&Cid>) -> Result<Option<Cid>> {
let Some(root) = root else {
return Ok(None);
};
match self.event_store.validate_index_root(Some(root)).await {
Ok(()) => Ok(Some(root.clone())),
Err(err) => {
eprintln!(
"Ignoring invalid existing Nostr event index root {}: {err}",
hex::encode(root.hash)
);
Ok(None)
}
}
}
async fn crawl_author_batches(
&self,
client: &Client,
authors: &[String],
existing_root: Option<&Cid>,
on_progress: &mut impl FnMut(&CrawlReport),
) -> Result<CrawlReport> {
let mut relay_negentropy_support = BTreeMap::<String, bool>::new();
let mut failed_relays = BTreeSet::<String>::new();
let mut current_root = existing_root.cloned();
let mut events_seen = 0usize;
let mut events_selected = 0usize;
let mut live_bytes_selected = 0u64;
let mut authors_processed = 0usize;
let mut applied_events = Vec::new();
for author_batch in authors.chunks(self.config.author_batch_size) {
let batch = self
.crawl_author_batch(
client,
author_batch,
current_root.as_ref(),
&mut relay_negentropy_support,
&mut failed_relays,
live_bytes_selected,
)
.await?;
events_seen = events_seen.saturating_add(batch.events_seen);
events_selected = events_selected.saturating_add(batch.events_selected);
live_bytes_selected = batch.live_bytes_selected;
authors_processed = authors_processed.saturating_add(author_batch.len());
if !batch.events.is_empty() {
applied_events.extend(batch.events.clone());
current_root = self
.event_store
.build(current_root.as_ref(), batch.events)
.await?;
}
on_progress(&CrawlReport {
root: current_root.clone(),
authors_considered: authors.len(),
authors_processed,
events_seen,
events_selected,
live_bytes_selected,
applied_events: Vec::new(),
});
if self.reached_events_seen_limit(events_seen) {
break;
}
}
Ok(CrawlReport {
root: current_root,
authors_considered: authors.len(),
authors_processed,
events_seen,
events_selected,
live_bytes_selected,
applied_events,
})
}
async fn load_existing_global_state(
&self,
root: Option<&Cid>,
authors: &[String],
) -> Result<GlobalRecentState> {
let Some(root) = root else {
return Ok(GlobalRecentState::default());
};
match self
.event_store
.list_recent(Some(root), ListEventsOptions::default())
.await
{
Ok(events) => {
let author_set = authors.iter().map(String::as_str).collect::<BTreeSet<_>>();
let mut retained_by_author = BTreeMap::<String, Vec<StoredNostrEvent>>::new();
for event in events {
if !author_set.contains(event.pubkey.as_str()) || !self.kind_allowed(event.kind)
{
continue;
}
if !self.is_valid_stored_event(&event) {
continue;
}
retained_by_author
.entry(event.pubkey.clone())
.or_default()
.push(event);
}
let mut state = GlobalRecentState {
current_root: Some(root.clone()),
..GlobalRecentState::default()
};
for (author, events) in retained_by_author {
let selected = self.select_author_events(events)?;
state.events_selected = state.events_selected.saturating_add(selected.len());
state.live_bytes_selected = state
.live_bytes_selected
.saturating_add(self.encoded_events_size(&selected)?);
state.retained_by_author.insert(author, selected);
}
Ok(state)
}
Err(NostrEventStoreError::Validation(message))
if message == "stored nostr event blob is missing" =>
{
eprintln!(
"Falling back to per-author resume for existing root due to missing event blobs"
);
let mut state = self
.load_existing_global_state_by_author(Some(root), authors)
.await?;
state.current_root = self.rebuild_root_from_retained_state(&state).await?;
Ok(state)
}
Err(err) => Err(err.into()),
}
}
async fn load_existing_global_state_by_author(
&self,
root: Option<&Cid>,
authors: &[String],
) -> Result<GlobalRecentState> {
let mut state = GlobalRecentState {
current_root: root.cloned(),
..GlobalRecentState::default()
};
for author in authors {
let retained = self
.load_retained_events(root, author)
.await?
.into_iter()
.filter(|event| self.kind_allowed(event.kind))
.filter(|event| self.is_valid_stored_event(event))
.collect::<Vec<_>>();
let retained = self.select_author_events(retained)?;
state.events_selected = state.events_selected.saturating_add(retained.len());
state.live_bytes_selected = state
.live_bytes_selected
.saturating_add(self.encoded_events_size(&retained)?);
state.retained_by_author.insert(author.clone(), retained);
}
Ok(state)
}
async fn rebuild_root_from_retained_state(
&self,
state: &GlobalRecentState,
) -> Result<Option<Cid>> {
let events = state
.retained_by_author
.values()
.flat_map(|events| events.iter().cloned())
.collect::<Vec<_>>();
self.event_store
.build(None, events)
.await
.map_err(Into::into)
}
fn validate_config(&self) -> Result<()> {
if self.config.relays.is_empty() {
return Err(CrawlError::MissingRelays);
}
if self.config.per_author_event_limit == 0 {
return Err(CrawlError::InvalidPerAuthorLimit);
}
if self.config.per_author_live_bytes == Some(0) {
return Err(CrawlError::InvalidPerAuthorLiveBytes);
}
if self.config.author_batch_size == 0 {
return Err(CrawlError::InvalidAuthorBatchSize);
}
if self.config.relay_page_size == 0 {
return Err(CrawlError::InvalidRelayPageSize);
}
if self.config.max_relay_pages == 0 && !self.config.full_author_history {
return Err(CrawlError::InvalidMaxRelayPages);
}
if self.config.max_events_seen == Some(0) {
return Err(CrawlError::InvalidMaxEventsSeen);
}
if self.config.relay_event_max_size == Some(0) {
return Err(CrawlError::InvalidRelayEventMaxSize);
}
Ok(())
}
fn collect_authors<G: SocialGraphBackend>(&self, graph: &G) -> Result<Vec<String>> {
if let Some(author_allowlist) = &self.config.author_allowlist {
let mut seen = HashSet::new();
let mut authors = Vec::new();
for author in author_allowlist {
if !is_valid_hex_pubkey(author) {
continue;
}
if seen.insert(author.clone()) {
authors.push(author.clone());
}
}
if let Some(max_authors) = self.config.max_authors {
authors.truncate(max_authors);
}
return Ok(authors);
}
let root = graph
.get_root()
.map_err(|err| CrawlError::SocialGraph(err.to_string()))?;
let mut visited = BTreeSet::new();
let mut authors = Vec::new();
let mut queue = VecDeque::from([(root.clone(), 0u32)]);
visited.insert(root);
while let Some((author, distance)) = queue.pop_front() {
if !is_valid_hex_pubkey(&author) {
continue;
}
authors.push(author.clone());
if self
.config
.max_authors
.is_some_and(|max_authors| authors.len() >= max_authors)
{
break;
}
if self
.config
.max_follow_distance
.is_some_and(|max_distance| distance >= max_distance)
{
continue;
}
let mut follows = graph
.get_followed_by_user(&author)
.map_err(|err| CrawlError::SocialGraph(err.to_string()))?;
follows.retain(|followed| is_valid_hex_pubkey(followed));
follows.sort();
for followed in follows {
if visited.insert(followed.clone()) {
queue.push_back((followed, distance.saturating_add(1)));
}
}
}
Ok(authors)
}
async fn connect_client(&self) -> Result<Client> {
let client = if let Some(max_size) = self.config.relay_event_max_size {
let mut limits = RelayLimits::default();
limits.events.max_size = Some(max_size);
Client::builder()
.signer(Keys::generate())
.opts(ClientOptions::new().relay_limits(limits))
.build()
} else {
Client::new(Keys::generate())
};
for relay in &self.config.relays {
client
.add_relay(relay)
.await
.map_err(|err| CrawlError::Nostr(err.to_string()))?;
}
client.connect().await;
tokio::time::sleep(Duration::from_millis(250)).await;
Ok(client)
}
async fn crawl_author_batch(
&self,
client: &Client,
author_batch: &[String],
current_root: Option<&Cid>,
relay_negentropy_support: &mut BTreeMap<String, bool>,
failed_relays: &mut BTreeSet<String>,
live_bytes_selected_so_far: u64,
) -> Result<BatchCrawlReport> {
let mut existing_by_author = BTreeMap::<String, Vec<StoredNostrEvent>>::new();
let mut known = BTreeMap::<String, StoredNostrEvent>::new();
for author in author_batch {
let retained = self
.load_retained_events(current_root, author)
.await?
.into_iter()
.filter(|event| self.kind_allowed(event.kind))
.filter(|event| self.is_valid_stored_event(event))
.collect::<Vec<_>>();
for event in &retained {
known.insert(event.id.clone(), event.clone());
}
existing_by_author.insert(author.clone(), retained);
}
let pubkeys: Vec<PublicKey> = author_batch
.iter()
.filter_map(|author| author.parse::<PublicKey>().ok())
.collect();
if pubkeys.is_empty() {
return Ok(BatchCrawlReport {
events_seen: 0,
events_selected: 0,
events: Vec::new(),
live_bytes_selected: live_bytes_selected_so_far,
});
}
let initial_known_ids = known.keys().cloned().collect::<BTreeSet<_>>();
if self.config.full_author_history {
return self
.crawl_full_author_history_batch(
client,
author_batch,
existing_by_author,
known,
initial_known_ids,
relay_negentropy_support,
failed_relays,
live_bytes_selected_so_far,
)
.await;
}
let filter = self.batch_filter(pubkeys);
let mut fetched = BTreeMap::<String, StoredNostrEvent>::new();
let mut events_seen = 0usize;
for relay in &self.config.relays {
if failed_relays.contains(relay) {
continue;
}
let local_items = self.local_items_for_batch(known.values(), author_batch);
let relay_support = relay_negentropy_support.get(relay).copied();
let fetched_from_relay = match self
.fetch_events_from_relay(client, relay, filter.clone(), local_items, relay_support)
.await
{
Ok(result) => result,
Err(err) => {
eprintln!("Skipping relay {relay}: {err}");
failed_relays.insert(relay.clone());
continue;
}
};
relay_negentropy_support.insert(relay.clone(), fetched_from_relay.supports_negentropy);
events_seen = events_seen.saturating_add(fetched_from_relay.events_seen);
for event in fetched_from_relay.events {
if self.kind_allowed(event.kind)
&& known.insert(event.id.clone(), event.clone()).is_none()
{
fetched.insert(event.id.clone(), event);
}
}
}
let mut fetched_by_author = BTreeMap::<String, Vec<StoredNostrEvent>>::new();
for event in fetched.into_values() {
fetched_by_author
.entry(event.pubkey.clone())
.or_default()
.push(event);
}
let mut selected = Vec::new();
for author in author_batch {
let mut merged: BTreeMap<String, StoredNostrEvent> = BTreeMap::new();
if let Some(existing_events) = existing_by_author.remove(author) {
for event in existing_events {
merged.insert(event.id.clone(), event);
}
}
if let Some(events) = fetched_by_author.remove(author) {
for event in events {
merged.insert(event.id.clone(), event);
}
}
selected.extend(self.select_author_events(merged.into_values().collect())?);
}
let selected = selected
.into_iter()
.filter(|event| self.is_valid_stored_event(event))
.collect::<Vec<_>>();
let (selected, live_bytes_selected) =
self.apply_live_byte_cap_from(selected, live_bytes_selected_so_far)?;
let events_selected = selected.len();
let events_to_apply = selected
.into_iter()
.filter(|event| !initial_known_ids.contains(&event.id))
.collect::<Vec<_>>();
Ok(BatchCrawlReport {
events_seen,
events_selected,
events: events_to_apply,
live_bytes_selected,
})
}
#[allow(clippy::too_many_arguments)]
async fn crawl_full_author_history_batch(
&self,
client: &Client,
author_batch: &[String],
mut existing_by_author: BTreeMap<String, Vec<StoredNostrEvent>>,
mut known: BTreeMap<String, StoredNostrEvent>,
initial_known_ids: BTreeSet<String>,
relay_negentropy_support: &mut BTreeMap<String, bool>,
failed_relays: &mut BTreeSet<String>,
live_bytes_selected_so_far: u64,
) -> Result<BatchCrawlReport> {
let mut events_seen = 0usize;
let mut selected = Vec::new();
let author_set = author_batch
.iter()
.map(String::as_str)
.collect::<BTreeSet<_>>();
let pubkeys = author_batch
.iter()
.filter_map(|author| author.parse::<PublicKey>().ok())
.collect::<Vec<_>>();
if pubkeys.is_empty() {
return Ok(BatchCrawlReport {
events_seen: 0,
events_selected: 0,
events: Vec::new(),
live_bytes_selected: live_bytes_selected_so_far,
});
}
let local_items = self.local_items_for_batch(known.values(), author_batch);
let mut fetched_by_author = BTreeMap::<String, Vec<StoredNostrEvent>>::new();
let relays_to_fetch = self
.config
.relays
.iter()
.filter(|relay| !failed_relays.contains(*relay))
.map(|relay| (relay.clone(), relay_negentropy_support.get(relay).copied()))
.collect::<Vec<_>>();
let relay_fetches = relays_to_fetch.into_iter().map(|(relay, relay_support)| {
let local_items = local_items.clone();
let pubkeys = &pubkeys;
async move {
let result = self
.fetch_full_history_from_relay(
client,
&relay,
pubkeys,
local_items,
relay_support,
)
.await;
(relay, result)
}
});
let mut relay_fetches =
stream::iter(relay_fetches).buffer_unordered(self.config.relays.len().max(1));
while let Some((relay, result)) = relay_fetches.next().await {
match result {
Ok(fetched_from_relay) => {
relay_negentropy_support
.insert(relay.clone(), fetched_from_relay.supports_negentropy);
events_seen = events_seen.saturating_add(fetched_from_relay.events_seen);
for event in fetched_from_relay.events {
if self.kind_allowed(event.kind)
&& author_set.contains(event.pubkey.as_str())
&& known.insert(event.id.clone(), event.clone()).is_none()
{
fetched_by_author
.entry(event.pubkey.clone())
.or_default()
.push(event);
}
}
}
Err(err) => {
eprintln!("Skipping relay {relay}: {err}");
failed_relays.insert(relay.clone());
}
}
}
for author in author_batch {
let mut merged = BTreeMap::<String, StoredNostrEvent>::new();
if let Some(existing_events) = existing_by_author.remove(author) {
for event in existing_events {
merged.insert(event.id.clone(), event);
}
}
if let Some(events) = fetched_by_author.remove(author) {
for event in events {
merged.insert(event.id.clone(), event);
}
}
let author_selected = self
.select_author_events_with_limits(
merged.into_values().collect(),
self.config.per_author_event_limit,
self.config.per_author_live_bytes,
)?
.into_iter()
.filter(|event| self.is_valid_stored_event(event))
.collect::<Vec<_>>();
selected.extend(author_selected);
}
let (events, live_bytes_selected) =
self.apply_live_byte_cap_from(selected, live_bytes_selected_so_far)?;
let events_selected = events.len();
let events_to_apply = events
.into_iter()
.filter(|event| !initial_known_ids.contains(&event.id))
.collect::<Vec<_>>();
Ok(BatchCrawlReport {
events_seen,
events_selected,
events: events_to_apply,
live_bytes_selected,
})
}
async fn crawl_global_recent_incremental(
&self,
client: &Client,
authors: &[String],
mut state: GlobalRecentState,
on_progress: &mut impl FnMut(&CrawlReport),
) -> Result<CrawlReport> {
let author_set = authors.iter().map(String::as_str).collect::<BTreeSet<_>>();
let mut known_ids = state
.retained_by_author
.values()
.flat_map(|events| events.iter().map(|event| event.id.clone()))
.collect::<BTreeSet<_>>();
let mut authors_processed = state
.retained_by_author
.values()
.filter(|events| !events.is_empty())
.count();
let mut failed_relays = BTreeSet::<String>::new();
let mut relay_negentropy_support = BTreeMap::<String, bool>::new();
let mut events_seen = 0usize;
let mut applied_events = Vec::new();
self.hydrate_global_recent_profiles(
client,
authors,
&mut state,
&mut known_ids,
&mut authors_processed,
&mut applied_events,
&mut relay_negentropy_support,
&mut failed_relays,
&mut events_seen,
on_progress,
)
.await?;
if self.reached_events_seen_limit(events_seen) {
return Ok(CrawlReport {
root: state.current_root,
authors_considered: authors.len(),
authors_processed,
events_seen,
events_selected: state.events_selected,
live_bytes_selected: state.live_bytes_selected,
applied_events,
});
}
for relay in &self.config.relays {
if failed_relays.contains(relay) {
continue;
}
let mut until = None;
for _ in 0..self.config.max_relay_pages {
let filter = self.global_recent_filter(until);
let events = match client
.fetch_events_from([relay], filter, self.config.fetch_timeout)
.await
.map(|events| events.to_vec())
{
Ok(events) => events,
Err(err) => {
eprintln!("Skipping relay {relay}: {}", err);
failed_relays.insert(relay.clone());
break;
}
};
let fetched_count = events.len();
events_seen = events_seen.saturating_add(fetched_count);
if fetched_count == 0 {
break;
}
let mut min_created_at = u64::MAX;
let mut incoming_by_author = BTreeMap::<String, Vec<StoredNostrEvent>>::new();
for event in events {
min_created_at = min_created_at.min(event.created_at.as_secs());
if event.kind.is_ephemeral() {
continue;
}
let stored = stored_event_from_nostr(&event);
if !author_set.contains(stored.pubkey.as_str())
|| !self.kind_allowed(stored.kind)
{
continue;
}
incoming_by_author
.entry(stored.pubkey.clone())
.or_default()
.push(stored);
}
let pending_apply = self.merge_author_events_into_state(
&mut state,
incoming_by_author,
&mut known_ids,
&mut authors_processed,
)?;
if !pending_apply.is_empty() {
applied_events.extend(pending_apply.clone());
state.current_root = self
.event_store
.build(state.current_root.as_ref(), pending_apply)
.await?;
}
on_progress(&CrawlReport {
root: state.current_root.clone(),
authors_considered: authors.len(),
authors_processed,
events_seen,
events_selected: state.events_selected,
live_bytes_selected: state.live_bytes_selected,
applied_events: Vec::new(),
});
if min_created_at == u64::MAX || min_created_at == 0 {
break;
}
let next_until = min_created_at.saturating_sub(1);
if until == Some(next_until) {
break;
}
until = Some(next_until);
if self.reached_events_seen_limit(events_seen) {
break;
}
}
if self.reached_events_seen_limit(events_seen) {
break;
}
}
Ok(CrawlReport {
root: state.current_root,
authors_considered: authors.len(),
authors_processed,
events_seen,
events_selected: state.events_selected,
live_bytes_selected: state.live_bytes_selected,
applied_events,
})
}
fn batch_filter(&self, pubkeys: Vec<PublicKey>) -> Filter {
let mut filter = Filter::new().authors(pubkeys);
if let Some(kinds) = &self.config.kinds {
filter = filter.kinds(kinds.iter().copied().map(Kind::from));
}
let mut relay_limit = self
.config
.author_batch_size
.saturating_mul(self.config.per_author_event_limit);
let relay_page_budget = self
.config
.relay_page_size
.saturating_mul(self.config.max_relay_pages.max(1));
if relay_page_budget > 0 {
relay_limit = relay_limit.min(relay_page_budget);
}
if relay_limit > 0 {
filter = filter.limit(relay_limit);
}
filter
}
fn full_history_negentropy_filter(&self, pubkeys: Vec<PublicKey>) -> Filter {
let mut filter = Filter::new().authors(pubkeys);
if let Some(kinds) = &self.config.kinds {
filter = filter.kinds(kinds.iter().copied().map(Kind::from));
}
filter
}
fn global_recent_filter(&self, until: Option<u64>) -> Filter {
let mut filter = Filter::new().limit(self.config.relay_page_size);
if let Some(kinds) = &self.config.kinds {
filter = filter.kinds(kinds.iter().copied().map(Kind::from));
}
if let Some(until) = until {
filter = filter.until(Timestamp::from_secs(until));
}
filter
}
fn reached_events_seen_limit(&self, events_seen: usize) -> bool {
self.config
.max_events_seen
.is_some_and(|limit| events_seen >= limit)
}
fn is_valid_stored_event(&self, event: &StoredNostrEvent) -> bool {
self.event_store.encode_event(event).is_ok()
}
fn encoded_events_size(&self, events: &[StoredNostrEvent]) -> Result<u64> {
let mut total = 0u64;
for event in events {
total = total.saturating_add(self.event_store.encode_event(event)?.len() as u64);
}
Ok(total)
}
fn local_items_for_batch<'a, I>(
&self,
known_events: I,
author_batch: &[String],
) -> Vec<(EventId, Timestamp)>
where
I: Iterator<Item = &'a StoredNostrEvent>,
{
let authors = author_batch
.iter()
.map(String::as_str)
.collect::<BTreeSet<_>>();
known_events
.filter(|event| {
authors.contains(event.pubkey.as_str()) && self.kind_allowed(event.kind)
})
.filter_map(|event| {
let event_id = EventId::parse(&event.id).ok()?;
Some((event_id, Timestamp::from_secs(event.created_at)))
})
.collect()
}
fn local_items_for_batch_by_kind<'a, I>(
&self,
known_events: I,
author_batch: &[String],
kind: u32,
) -> Vec<(EventId, Timestamp)>
where
I: Iterator<Item = &'a StoredNostrEvent>,
{
let authors = author_batch
.iter()
.map(String::as_str)
.collect::<BTreeSet<_>>();
known_events
.filter(|event| {
authors.contains(event.pubkey.as_str())
&& event.kind == kind
&& self.kind_allowed(event.kind)
})
.filter_map(|event| {
let event_id = EventId::parse(&event.id).ok()?;
Some((event_id, Timestamp::from_secs(event.created_at)))
})
.collect()
}
async fn load_retained_events(
&self,
root: Option<&Cid>,
author: &str,
) -> Result<Vec<StoredNostrEvent>> {
self.event_store
.list_by_author_lossy(root, author, ListEventsOptions::default())
.await
.map_err(Into::into)
}
fn merge_author_events_into_state(
&self,
state: &mut GlobalRecentState,
incoming_by_author: BTreeMap<String, Vec<StoredNostrEvent>>,
known_ids: &mut BTreeSet<String>,
authors_processed: &mut usize,
) -> Result<Vec<StoredNostrEvent>> {
let mut pending_apply = Vec::new();
for (author, incoming) in incoming_by_author {
let retained = state.retained_by_author.entry(author).or_default();
let was_empty = retained.is_empty();
let old_len = retained.len();
let old_live_bytes = self.encoded_events_size(retained)?;
let mut merged = BTreeMap::<String, StoredNostrEvent>::new();
for existing in retained.drain(..) {
merged.insert(existing.id.clone(), existing);
}
for event in incoming {
merged.insert(event.id.clone(), event);
}
let selected = self.select_author_events(merged.into_values().collect())?;
let selected_live_bytes = self.encoded_events_size(&selected)?;
for selected_event in &selected {
if known_ids.insert(selected_event.id.clone()) {
pending_apply.push(selected_event.clone());
}
}
state.events_selected = state
.events_selected
.saturating_sub(old_len)
.saturating_add(selected.len());
state.live_bytes_selected = state
.live_bytes_selected
.saturating_sub(old_live_bytes)
.saturating_add(selected_live_bytes);
if was_empty && !selected.is_empty() {
*authors_processed = authors_processed.saturating_add(1);
}
*retained = selected;
}
Ok(pending_apply)
}
#[allow(clippy::too_many_arguments)]
async fn hydrate_global_recent_profiles(
&self,
client: &Client,
authors: &[String],
state: &mut GlobalRecentState,
known_ids: &mut BTreeSet<String>,
authors_processed: &mut usize,
applied_events: &mut Vec<StoredNostrEvent>,
relay_negentropy_support: &mut BTreeMap<String, bool>,
failed_relays: &mut BTreeSet<String>,
events_seen: &mut usize,
on_progress: &mut impl FnMut(&CrawlReport),
) -> Result<()> {
if !self.kind_allowed(METADATA_KIND) {
return Ok(());
}
let authors_missing_profiles = authors
.iter()
.filter(|author| {
!state
.retained_by_author
.get(*author)
.is_some_and(|events| events.iter().any(|event| event.kind == METADATA_KIND))
})
.cloned()
.collect::<Vec<_>>();
if authors_missing_profiles.is_empty() {
return Ok(());
}
for author_batch in authors_missing_profiles.chunks(self.config.author_batch_size.max(1)) {
let batch = self
.crawl_profile_batch(
client,
author_batch,
state
.retained_by_author
.values()
.flat_map(|events| events.iter()),
relay_negentropy_support,
failed_relays,
)
.await?;
*events_seen = events_seen.saturating_add(batch.events_seen);
let pending_apply = self.merge_author_events_into_state(
state,
batch.events_by_author,
known_ids,
authors_processed,
)?;
if !pending_apply.is_empty() {
applied_events.extend(pending_apply.clone());
state.current_root = self
.event_store
.build(state.current_root.as_ref(), pending_apply)
.await?;
}
on_progress(&CrawlReport {
root: state.current_root.clone(),
authors_considered: authors.len(),
authors_processed: *authors_processed,
events_seen: *events_seen,
events_selected: state.events_selected,
live_bytes_selected: state.live_bytes_selected,
applied_events: Vec::new(),
});
if self.reached_events_seen_limit(*events_seen) {
break;
}
}
Ok(())
}
async fn crawl_profile_batch<'a, I>(
&self,
client: &Client,
author_batch: &[String],
known_events: I,
relay_negentropy_support: &mut BTreeMap<String, bool>,
failed_relays: &mut BTreeSet<String>,
) -> Result<ProfileBatchReport>
where
I: Iterator<Item = &'a StoredNostrEvent>,
{
let pubkeys: Vec<PublicKey> = author_batch
.iter()
.filter_map(|author| author.parse::<PublicKey>().ok())
.collect();
if pubkeys.is_empty() {
return Ok(ProfileBatchReport::default());
}
let filter = Filter::new()
.authors(pubkeys)
.kind(Kind::Metadata)
.limit(author_batch.len().saturating_mul(2).max(1));
let local_items =
self.local_items_for_batch_by_kind(known_events, author_batch, METADATA_KIND);
let mut fetched_by_author = BTreeMap::<String, Vec<StoredNostrEvent>>::new();
let mut events_seen = 0usize;
for relay in &self.config.relays {
if failed_relays.contains(relay) {
continue;
}
let relay_support = relay_negentropy_support.get(relay).copied();
let fetched_from_relay = match self
.fetch_events_from_relay(
client,
relay,
filter.clone(),
local_items.clone(),
relay_support,
)
.await
{
Ok(result) => result,
Err(err) => {
eprintln!("Skipping relay {relay}: {err}");
failed_relays.insert(relay.clone());
continue;
}
};
relay_negentropy_support.insert(relay.clone(), fetched_from_relay.supports_negentropy);
events_seen = events_seen.saturating_add(fetched_from_relay.events_seen);
for event in fetched_from_relay.events {
if event.kind == METADATA_KIND && self.kind_allowed(event.kind) {
fetched_by_author
.entry(event.pubkey.clone())
.or_default()
.push(event);
}
}
}
Ok(ProfileBatchReport {
events_seen,
events_by_author: fetched_by_author,
})
}
async fn fetch_events_from_relay(
&self,
client: &Client,
relay: &str,
filter: Filter,
local_items: Vec<(EventId, Timestamp)>,
supports_negentropy: Option<bool>,
) -> Result<RelayFetchResult> {
if supports_negentropy == Some(false) {
if self.config.require_negentropy {
return Ok(RelayFetchResult {
events_seen: 0,
events: Vec::new(),
supports_negentropy: false,
});
}
return self
.fetch_full_filter(client, relay, filter)
.await
.map(|events| RelayFetchResult {
events_seen: events.len(),
events,
supports_negentropy: false,
});
}
match self
.reconcile_missing_ids(client, relay, filter.clone(), local_items)
.await
{
Ok(Some(missing)) => self.fetch_missing_ids(client, relay, missing).await.map(
|RelayFetchResult {
events_seen,
events,
..
}| RelayFetchResult {
events_seen,
events,
supports_negentropy: true,
},
),
Ok(None) | Err(_) => {
if self.config.require_negentropy {
Ok(RelayFetchResult {
events_seen: 0,
events: Vec::new(),
supports_negentropy: false,
})
} else {
self.fetch_full_filter(client, relay, filter)
.await
.map(|events| RelayFetchResult {
events_seen: events.len(),
events,
supports_negentropy: false,
})
}
}
}
}
async fn fetch_missing_ids(
&self,
client: &Client,
relay: &str,
missing_ids: Vec<EventId>,
) -> Result<RelayFetchResult> {
if missing_ids.is_empty() {
return Ok(RelayFetchResult {
events_seen: 0,
events: Vec::new(),
supports_negentropy: true,
});
}
let mut out = BTreeMap::<String, StoredNostrEvent>::new();
let mut events_seen = 0usize;
let filters = missing_ids
.chunks(NEGENTROPY_FETCH_CHUNK_SIZE)
.map(|chunk| Filter::new().ids(chunk.iter().cloned()))
.collect::<Vec<_>>();
let fetches = filters.into_iter().map(|filter| async move {
client
.fetch_events_from([relay], filter, self.config.fetch_timeout)
.await
.map(|events| events.to_vec())
.map_err(|err| CrawlError::Nostr(err.to_string()))
});
let mut fetches =
stream::iter(fetches).buffer_unordered(NEGENTROPY_FETCH_CHUNK_CONCURRENCY);
while let Some(result) = fetches.next().await {
let events = result?;
events_seen = events_seen.saturating_add(events.len());
for event in events {
if event.kind.is_ephemeral() {
continue;
}
let stored = stored_event_from_nostr(&event);
out.insert(stored.id.clone(), stored);
}
}
Ok(RelayFetchResult {
events_seen,
events: out.into_values().collect(),
supports_negentropy: true,
})
}
async fn fetch_full_filter(
&self,
client: &Client,
relay: &str,
filter: Filter,
) -> Result<Vec<StoredNostrEvent>> {
let mut out = Vec::new();
let events = client
.fetch_events_from([relay], filter, self.config.fetch_timeout)
.await
.map(|events| events.to_vec())
.map_err(|err| CrawlError::Nostr(err.to_string()))?;
for event in events {
if event.kind.is_ephemeral() {
continue;
}
out.push(stored_event_from_nostr(&event));
}
Ok(out)
}
async fn fetch_full_history_from_relay(
&self,
client: &Client,
relay: &str,
pubkeys: &[PublicKey],
local_items: Vec<(EventId, Timestamp)>,
supports_negentropy: Option<bool>,
) -> Result<RelayFetchResult> {
if supports_negentropy != Some(false) {
let filter = self.full_history_negentropy_filter(pubkeys.to_vec());
match self
.reconcile_missing_ids(client, relay, filter, local_items)
.await
{
Ok(Some(missing)) => {
return match self.fetch_missing_ids(client, relay, missing).await {
Ok(RelayFetchResult {
events_seen,
events,
..
}) => Ok(RelayFetchResult {
events_seen,
events,
supports_negentropy: true,
}),
Err(err)
if !self.config.require_negentropy
&& self.config.max_relay_pages > 0 =>
{
self.fetch_full_history_by_paging_from_relay(client, relay, pubkeys)
.await
.map_err(|_| err)
}
Err(err) => Err(err),
};
}
Ok(None) | Err(_) if self.config.require_negentropy => {
return Ok(RelayFetchResult {
events_seen: 0,
events: Vec::new(),
supports_negentropy: false,
});
}
Ok(None) | Err(_) => {}
}
}
if self.config.require_negentropy || self.config.max_relay_pages == 0 {
return Ok(RelayFetchResult {
events_seen: 0,
events: Vec::new(),
supports_negentropy: false,
});
}
self.fetch_full_history_by_paging_from_relay(client, relay, pubkeys)
.await
}
async fn reconcile_missing_ids(
&self,
client: &Client,
relay: &str,
filter: Filter,
local_items: Vec<(EventId, Timestamp)>,
) -> Result<Option<Vec<EventId>>> {
let initial_timeout = self.config.fetch_timeout.min(NEGENTROPY_INITIAL_TIMEOUT);
let opts = SyncOptions::default()
.initial_timeout(initial_timeout)
.dry_run();
let targets = [(relay.to_owned(), (filter, local_items))];
let sync = client.pool().sync_targeted(targets, &opts);
let output = match tokio::time::timeout(self.config.fetch_timeout, sync).await {
Ok(Ok(output)) => output,
Ok(Err(_)) | Err(_) => return Ok(None),
};
if output.success.is_empty() {
return Ok(None);
}
Ok(Some(output.remote.iter().cloned().collect()))
}
async fn fetch_full_history_by_paging_from_relay(
&self,
client: &Client,
relay: &str,
pubkeys: &[PublicKey],
) -> Result<RelayFetchResult> {
let mut out = BTreeMap::<String, StoredNostrEvent>::new();
let mut events_seen = 0usize;
let concurrency = FULL_HISTORY_PAGING_CONCURRENCY_PER_RELAY
.min(pubkeys.len().max(1))
.max(1);
let fetches = pubkeys.iter().copied().map(|pubkey| async move {
self.fetch_full_author_history_by_paging_from_relay(client, relay, pubkey)
.await
});
let mut fetches = stream::iter(fetches).buffer_unordered(concurrency);
while let Some(result) = fetches.next().await {
let fetched = result?;
events_seen = events_seen.saturating_add(fetched.events_seen);
for event in fetched.events {
out.insert(event.id.clone(), event);
}
if self.reached_events_seen_limit(events_seen) {
break;
}
}
Ok(RelayFetchResult {
events_seen,
events: out.into_values().collect(),
supports_negentropy: false,
})
}
async fn fetch_full_author_history_by_paging_from_relay(
&self,
client: &Client,
relay: &str,
pubkey: PublicKey,
) -> Result<RelayFetchResult> {
let mut out = BTreeMap::<String, StoredNostrEvent>::new();
let mut events_seen = 0usize;
let mut until = None;
for _ in 0..self.config.max_relay_pages {
let remaining = self.config.per_author_event_limit.saturating_sub(out.len());
if remaining == 0 {
break;
}
let mut filter = Filter::new()
.author(pubkey)
.limit(self.config.relay_page_size.min(remaining));
if let Some(kinds) = &self.config.kinds {
filter = filter.kinds(kinds.iter().copied().map(Kind::from));
}
if let Some(until) = until {
filter = filter.until(Timestamp::from_secs(until));
}
let events = client
.fetch_events_from([relay], filter, self.config.fetch_timeout)
.await
.map(|events| events.to_vec())
.map_err(|err| CrawlError::Nostr(err.to_string()))?;
let fetched_count = events.len();
events_seen = events_seen.saturating_add(fetched_count);
if fetched_count == 0 {
break;
}
let mut min_created_at = u64::MAX;
for event in events {
min_created_at = min_created_at.min(event.created_at.as_secs());
if event.kind.is_ephemeral() {
continue;
}
let stored = stored_event_from_nostr(&event);
out.insert(stored.id.clone(), stored);
if out.len() >= self.config.per_author_event_limit {
break;
}
}
if out.len() >= self.config.per_author_event_limit {
break;
}
if min_created_at == u64::MAX || min_created_at == 0 {
break;
}
let next_until = min_created_at.saturating_sub(1);
if until == Some(next_until) {
break;
}
until = Some(next_until);
if self.reached_events_seen_limit(events_seen) {
break;
}
}
Ok(RelayFetchResult {
events_seen,
events: out.into_values().collect(),
supports_negentropy: false,
})
}
fn select_author_events(&self, events: Vec<StoredNostrEvent>) -> Result<Vec<StoredNostrEvent>> {
self.select_author_events_with_limits(
events,
self.config.per_author_event_limit,
self.config.per_author_live_bytes,
)
}
fn select_author_events_with_limits(
&self,
mut events: Vec<StoredNostrEvent>,
event_limit: usize,
live_byte_limit: Option<u64>,
) -> Result<Vec<StoredNostrEvent>> {
let sticky_events = self.select_sticky_author_events(&events);
let sticky_ids = sticky_events
.iter()
.map(|event| event.id.clone())
.collect::<HashSet<_>>();
events.retain(|event| !sticky_ids.contains(&event.id));
events.sort_by(|left, right| {
self.policy
.priority(right)
.cmp(&self.policy.priority(left))
.then_with(|| right.created_at.cmp(&left.created_at))
.then_with(|| left.id.cmp(&right.id))
});
if let Some(max_live_bytes) = live_byte_limit {
let mut selected = sticky_events.clone();
let mut live_bytes_selected = self.encoded_events_size(&selected)?;
if live_bytes_selected > max_live_bytes {
return Ok(selected);
}
for event in events {
let encoded_len = self.event_store.encode_event(&event)?.len() as u64;
if live_bytes_selected.saturating_add(encoded_len) > max_live_bytes {
continue;
}
live_bytes_selected = live_bytes_selected.saturating_add(encoded_len);
selected.push(event);
if selected.len().saturating_sub(sticky_events.len()) >= event_limit {
break;
}
}
return Ok(selected);
}
let mut selected = sticky_events;
selected.extend(events.into_iter().take(event_limit));
Ok(selected)
}
fn select_sticky_author_events(&self, events: &[StoredNostrEvent]) -> Vec<StoredNostrEvent> {
let latest_metadata = events
.iter()
.filter(|event| event.kind == METADATA_KIND)
.cloned()
.max_by(|left, right| {
left.created_at
.cmp(&right.created_at)
.then_with(|| right.id.cmp(&left.id))
});
latest_metadata.into_iter().collect()
}
fn apply_live_byte_cap_from(
&self,
mut events: Vec<StoredNostrEvent>,
live_bytes_selected_so_far: u64,
) -> Result<(Vec<StoredNostrEvent>, u64)> {
events.sort_by(|left, right| {
self.policy
.priority(right)
.cmp(&self.policy.priority(left))
.then_with(|| right.created_at.cmp(&left.created_at))
.then_with(|| left.id.cmp(&right.id))
});
let Some(max_live_bytes) = self.config.max_live_bytes else {
let live_bytes_selected =
events
.iter()
.try_fold(live_bytes_selected_so_far, |total, event| {
let encoded = self.event_store.encode_event(event)?;
Ok::<u64, NostrEventStoreError>(total.saturating_add(encoded.len() as u64))
})?;
return Ok((events, live_bytes_selected));
};
let mut selected = Vec::new();
let mut live_bytes_selected = live_bytes_selected_so_far;
for event in events {
let encoded_len = self.event_store.encode_event(&event)?.len() as u64;
if live_bytes_selected.saturating_add(encoded_len) > max_live_bytes {
continue;
}
live_bytes_selected = live_bytes_selected.saturating_add(encoded_len);
selected.push(event);
}
Ok((selected, live_bytes_selected))
}
fn kind_allowed(&self, kind: u32) -> bool {
self.config.kinds.as_ref().is_none_or(|allowed| {
allowed
.iter()
.any(|candidate| u32::from(*candidate) == kind)
})
}
}
fn stored_event_from_nostr(event: &nostr_sdk::Event) -> StoredNostrEvent {
StoredNostrEvent {
id: event.id.to_hex(),
pubkey: event.pubkey.to_hex(),
created_at: event.created_at.as_secs(),
kind: event.kind.as_u16() as u32,
tags: event
.tags
.iter()
.map(|tag| tag.as_slice().to_vec())
.collect(),
content: event.content.clone(),
sig: event.sig.to_string(),
}
}
fn is_valid_hex_pubkey(value: &str) -> bool {
value.len() == 64
&& value
.bytes()
.all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use hashtree_core::MemoryStore;
use nostr_social_graph::{NostrEvent, SocialGraphBackend as NostrSocialGraphBackend};
use super::{CrawlConfig, NostrBridge, StoredNostrEvent};
#[derive(Default)]
struct FakeGraphBackend;
impl NostrSocialGraphBackend for FakeGraphBackend {
type Error = std::io::Error;
fn get_root(&self) -> std::result::Result<String, Self::Error> {
Ok("0".repeat(64))
}
fn set_root(&mut self, _root: &str) -> std::result::Result<(), Self::Error> {
Ok(())
}
fn handle_event(
&mut self,
_event: &NostrEvent,
_allow_unknown_authors: bool,
_overmute_threshold: f64,
) -> std::result::Result<(), Self::Error> {
Ok(())
}
fn get_follow_distance(&self, _user: &str) -> std::result::Result<u32, Self::Error> {
Ok(0)
}
fn is_following(
&self,
_follower: &str,
_followed_user: &str,
) -> std::result::Result<bool, Self::Error> {
Ok(false)
}
fn get_followed_by_user(
&self,
user: &str,
) -> std::result::Result<Vec<String>, Self::Error> {
if user == "0".repeat(64) {
return Ok(vec![
"1".repeat(64),
"NOT-HEX".to_string(),
"a".repeat(63),
"A".repeat(64),
]);
}
Ok(Vec::new())
}
fn get_followers_by_user(
&self,
_user: &str,
) -> std::result::Result<Vec<String>, Self::Error> {
Ok(Vec::new())
}
fn get_muted_by_user(&self, _user: &str) -> std::result::Result<Vec<String>, Self::Error> {
Ok(Vec::new())
}
fn get_user_muted_by(&self, _user: &str) -> std::result::Result<Vec<String>, Self::Error> {
Ok(Vec::new())
}
fn get_follow_list_created_at(
&self,
_user: &str,
) -> std::result::Result<Option<u64>, Self::Error> {
Ok(None)
}
fn get_mute_list_created_at(
&self,
_user: &str,
) -> std::result::Result<Option<u64>, Self::Error> {
Ok(None)
}
fn is_overmuted(
&self,
_user: &str,
_threshold: f64,
) -> std::result::Result<bool, Self::Error> {
Ok(false)
}
}
#[test]
fn rejects_invalid_stored_event_shape() {
let bridge = NostrBridge::new(Arc::new(MemoryStore::new()), CrawlConfig::default());
let invalid = StoredNostrEvent {
id: "f".repeat(64),
pubkey: "not-hex".to_string(),
created_at: 1,
kind: 1,
tags: Vec::new(),
content: String::new(),
sig: "f".repeat(128),
};
assert!(!bridge.is_valid_stored_event(&invalid));
}
#[test]
fn collect_authors_skips_invalid_graph_pubkeys() {
let bridge = NostrBridge::new(Arc::new(MemoryStore::new()), CrawlConfig::default());
let authors = bridge
.collect_authors(&FakeGraphBackend)
.expect("collect authors");
assert_eq!(authors, vec!["0".repeat(64), "1".repeat(64)]);
}
#[test]
fn collect_authors_prefers_allowlist_and_applies_limits() {
let bridge = NostrBridge::new(
Arc::new(MemoryStore::new()),
CrawlConfig {
author_allowlist: Some(vec![
"1".repeat(64),
"NOT-HEX".to_string(),
"0".repeat(64),
"1".repeat(64),
]),
max_authors: Some(1),
..CrawlConfig::default()
},
);
let authors = bridge
.collect_authors(&FakeGraphBackend)
.expect("collect authors");
assert_eq!(authors, vec!["1".repeat(64)]);
}
}