1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
use super::*;
impl NostrRootResolver {
/// Observe a mutable root for a bounded window, keeping the subscription open
/// after EOSE. Returns the newest usable signed root observed in that window;
/// a quiet window is a timeout, not proof that the root does not exist.
/// Unlike quick resolution, this does not use cached observations or a quorum
/// shortcut. It closes only its own subscription before returning.
pub async fn resolve_open(&self, key: &str, window: Duration) -> Result<Cid, ResolverError> {
self.observe_roots(key, window, false).await
}
/// Return the first usable authenticated root without waiting for EOSE or
/// a relay quorum. A quiet deadline is a timeout, never proof of absence.
/// Closes this lookup's subscription before returning.
pub async fn resolve_ready(&self, key: &str, timeout: Duration) -> Result<Cid, ResolverError> {
self.observe_roots(key, timeout, true).await
}
async fn observe_roots(
&self,
key: &str,
window: Duration,
ready: bool,
) -> Result<Cid, ResolverError> {
let (pubkey, tree_name) = Self::parse_key(key)?;
if self.config.relays.is_empty() {
return Err(ResolverError::Network(
"No active Nostr relays configured".into(),
));
}
let filter = Self::build_tree_filter(pubkey, &tree_name);
let id = SubscriptionId::generate();
// Register before REQ so even an immediate relay response is observed.
let mut notifications = self.client.notifications();
let mut latest: Option<(VerifiedEvent, Cid)> = None;
let observed: Result<Result<(), ResolverError>, _> = tokio::time::timeout(window, async {
self.client
.subscribe_with_id(id.clone(), filter, None)
.await
.map_err(|error| ResolverError::Network(error.to_string()))?;
loop {
let notification = notifications
.recv()
.await
.map_err(|error| ResolverError::Network(error.to_string()))?;
// Event notifications are globally deduplicated by the SDK. Raw
// messages also cover a root already seen on another subscription.
let RelayPoolNotification::Message {
message:
RelayMessage::Event {
subscription_id,
event,
},
..
} = notification
else {
continue;
};
if subscription_id.as_ref() != &id {
continue;
}
let Ok(event) = VerifiedEvent::try_from(event.into_owned()) else {
continue;
};
if event.as_event().pubkey != pubkey || !is_matching_tree_event(&event, &tree_name)
{
continue;
}
let Some(cid) = self.cid_from_event(event.as_event()) else {
continue;
};
if latest.as_ref().is_none_or(|(current, _)| {
is_newer_event(event.as_event(), current.created_at(), Some(current.id()))
}) {
latest = Some((event, cid));
if ready {
return Ok(());
}
}
}
})
.await;
self.client.unsubscribe(&id).await;
if let Ok(result) = observed {
result?;
}
latest.map(|(_, cid)| cid).ok_or_else(|| {
ResolverError::Network(format!("Timed out waiting for a signed root for {key}"))
})
}
}