hashtree_resolver/nostr/
open.rs

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

hashtree_resolver/nostr/
open.rs
