vector_core/
negentropy.rs

1//! Shared NIP-77 negentropy set reconciliation.
2//!
3//! One acquisition primitive for the whole app. DM and community sync differ only
4//! in their fingerprint source and processing; the reconcile-against-relays
5//! step is identical, so it lives here.
6
7use std::collections::HashSet;
8use std::time::Duration;
9
10use nostr_sdk::prelude::*;
11
12/// Race every trusted relay exchanging negentropy fingerprints for `filter`,
13/// and return the union of event IDs that relays hold but we don't.
14///
15/// `local_items` is our fingerprint set: `(event_id, created_at)` for
16/// everything we already possess. Each relay reports only the IDs absent from
17/// that set, so the union across relays is the complete missing set reachable
18/// from our trusted relays.
19///
20/// Every relay is drained (not just the first to respond): completeness beats
21/// latency here, and one relay may lack events another holds. Each relay is
22/// bounded by `timeout`.
23pub async fn reconcile_missing(
24    filter: Filter,
25    local_items: Vec<(EventId, Timestamp)>,
26    timeout: Duration,
27) -> Result<HashSet<EventId>, String> {
28    use futures_util::stream::{FuturesUnordered, StreamExt};
29
30    let client = crate::state::nostr_client().ok_or("Nostr client not initialized")?;
31
32    let opts = SyncOptions::new()
33        .direction(SyncDirection::Down)
34        .initial_timeout(timeout)
35        .dry_run();
36
37    // Resolve trusted relay URLs to live Relay handles.
38    let relay_map = client.relays().await;
39    let trusted = crate::state::active_trusted_relays().await;
40    let relays: Vec<(String, Relay)> = trusted.iter().filter_map(|url| {
41        let normalized = url.trim_end_matches('/');
42        relay_map.iter()
43            .find(|(u, _)| u.as_str().trim_end_matches('/') == normalized)
44            .map(|(_, r)| (url.to_string(), r.clone()))
45    }).collect();
46    drop(relay_map);
47
48    if relays.is_empty() {
49        crate::log_warn!("[Negentropy] No trusted relays available for reconciliation");
50        return Ok(HashSet::new());
51    }
52
53    let mut futs = FuturesUnordered::new();
54    for (url, relay) in &relays {
55        let url = url.clone();
56        let relay = relay.clone();
57        let f = filter.clone();
58        let items = local_items.clone();
59        let o = opts.clone();
60        futs.push(async move {
61            let r = tokio::time::timeout(timeout, relay.sync_with_items(f, items, &o)).await;
62            (url, r)
63        });
64    }
65
66    let mut missing: HashSet<EventId> = HashSet::new();
67    while let Some((url, result)) = futs.next().await {
68        match result {
69            Ok(Ok(recon)) => {
70                let n = recon.remote.len();
71                missing.extend(recon.remote);
72                crate::log_debug!("[Negentropy] {} reconciled: {} missing", url, n);
73            }
74            Ok(Err(e)) => crate::log_warn!("[Negentropy] {} failed: {}", url, e),
75            Err(_) => crate::log_warn!("[Negentropy] {} timed out", url),
76        }
77    }
78
79    Ok(missing)
80}
vector_core/negentropy.rs

vector_core/
negentropy.rs
