fips_core/
nostr_relay_adapter.rs

1//! Application-owned Nostr relay carrier for embedded FIPS endpoints.
2//!
3//! The adapter connects the ordinary FIPS relay transport to relays selected
4//! by the embedding application. It does not own discovery, policy, or relay
5//! configuration.
6
7use std::sync::Arc;
8use std::time::Duration;
9
10use nostr_sdk::Event;
11use nostr_sdk::prelude::{Client, Filter, Keys, Kind, PublicKey, RelayPoolNotification};
12use tokio::sync::{Semaphore, oneshot};
13use tokio::task::JoinHandle;
14
15use crate::transport::nostr_relay::NOSTR_RELAY_DATAGRAM_KIND;
16use crate::{FipsEndpoint, FipsEndpointError, NostrRelayIo};
17
18const NOSTR_RELAY_ACTIVE_OUTBOUND_POLL_INTERVAL: Duration = Duration::from_millis(10);
19const NOSTR_RELAY_IDLE_OUTBOUND_POLL_INTERVAL: Duration = Duration::from_millis(250);
20
21const fn outbound_poll_delay(drained_events: usize) -> Duration {
22    if drained_events == 0 {
23        NOSTR_RELAY_IDLE_OUTBOUND_POLL_INTERVAL
24    } else {
25        NOSTR_RELAY_ACTIVE_OUTBOUND_POLL_INTERVAL
26    }
27}
28
29/// Pumps signed kind-21060 datagrams between FIPS and application-selected
30/// Nostr relays.
31pub struct NostrRelayAdapter {
32    client: Client,
33    shutdown: Option<oneshot::Sender<()>>,
34    task: JoinHandle<()>,
35}
36
37#[derive(Clone)]
38enum RelayEventIo {
39    Endpoint(Arc<FipsEndpoint>),
40    Node(NostrRelayIo),
41}
42
43impl RelayEventIo {
44    fn npub(&self) -> &str {
45        match self {
46            Self::Endpoint(endpoint) => endpoint.npub(),
47            Self::Node(io) => io.npub(),
48        }
49    }
50
51    async fn drain_events(&self, limit: usize) -> Result<Vec<Event>, FipsEndpointError> {
52        match self {
53            Self::Endpoint(endpoint) => endpoint.drain_nostr_relay_events(limit).await,
54            Self::Node(io) => io.drain_events(limit).await,
55        }
56    }
57
58    async fn ingest_event(&self, event: Event) -> Result<bool, FipsEndpointError> {
59        match self {
60            Self::Endpoint(endpoint) => endpoint.ingest_nostr_event(event).await,
61            Self::Node(io) => io.ingest_event(event).await,
62        }
63    }
64}
65
66impl NostrRelayAdapter {
67    /// Connect an embedded endpoint through the application-provided relays.
68    pub async fn start(
69        endpoint: Arc<FipsEndpoint>,
70        relays: &[String],
71    ) -> Result<Option<Self>, String> {
72        Self::start_with_io(RelayEventIo::Endpoint(endpoint), relays).await
73    }
74
75    /// Connect a direct [`crate::Node`] through its attached relay I/O.
76    pub async fn start_for_node(
77        io: NostrRelayIo,
78        relays: &[String],
79    ) -> Result<Option<Self>, String> {
80        Self::start_with_io(RelayEventIo::Node(io), relays).await
81    }
82
83    async fn start_with_io(io: RelayEventIo, relays: &[String]) -> Result<Option<Self>, String> {
84        if relays.is_empty() {
85            return Ok(None);
86        }
87
88        let local_pubkey = PublicKey::parse(io.npub())
89            .map_err(|error| format!("invalid FIPS endpoint identity: {error}"))?;
90        let client = Client::new(Keys::generate());
91        for relay in relays {
92            client
93                .add_relay(relay)
94                .await
95                .map_err(|error| format!("failed to add Nostr relay {relay}: {error}"))?;
96        }
97        let mut notifications = client.notifications();
98        client.connect().await;
99        client
100            .subscribe(
101                Filter::new()
102                    .kind(Kind::Custom(NOSTR_RELAY_DATAGRAM_KIND))
103                    .pubkey(local_pubkey),
104                None,
105            )
106            .await
107            .map_err(|error| format!("failed to subscribe to FIPS relay datagrams: {error}"))?;
108
109        let task_client = client.clone();
110        let publish_slots = Arc::new(Semaphore::new(32));
111        let (shutdown_tx, mut shutdown_rx) = oneshot::channel();
112        let task = tokio::spawn(async move {
113            let outbound_sleep = tokio::time::sleep(Duration::ZERO);
114            tokio::pin!(outbound_sleep);
115            loop {
116                tokio::select! {
117                    _ = &mut shutdown_rx => break,
118                    () = &mut outbound_sleep => {
119                        let next_delay = match io.drain_events(64).await {
120                            Ok(events) => {
121                                let delay = outbound_poll_delay(events.len());
122                                for event in events {
123                                    let Ok(permit) = Arc::clone(&publish_slots).try_acquire_owned() else {
124                                        tracing::debug!(event_id = %event.id, "FIPS relay publish queue is saturated");
125                                        continue;
126                                    };
127                                    let client = task_client.clone();
128                                    tokio::spawn(async move {
129                                        let _permit = permit;
130                                        if let Err(error) = client.send_event(&event).await {
131                                            tracing::debug!(%error, event_id = %event.id, "failed to publish FIPS relay datagram");
132                                        }
133                                    });
134                                }
135                                delay
136                            }
137                            Err(error) => {
138                                tracing::debug!(%error, "failed to drain FIPS relay datagrams");
139                                NOSTR_RELAY_IDLE_OUTBOUND_POLL_INTERVAL
140                            }
141                        };
142                        outbound_sleep
143                            .as_mut()
144                            .reset(tokio::time::Instant::now() + next_delay);
145                    }
146                    notification = notifications.recv() => {
147                        match notification {
148                            Ok(RelayPoolNotification::Event { event, .. })
149                                if event.kind == Kind::Custom(NOSTR_RELAY_DATAGRAM_KIND) =>
150                            {
151                                if let Err(error) = io.ingest_event((*event).clone()).await {
152                                    tracing::debug!(%error, "failed to ingest FIPS relay datagram");
153                                }
154                            }
155                            Ok(_) => {}
156                            Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
157                                tracing::debug!(skipped, "FIPS relay notification receiver lagged");
158                            }
159                            Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
160                        }
161                    }
162                }
163            }
164        });
165
166        Ok(Some(Self {
167            client,
168            shutdown: Some(shutdown_tx),
169            task,
170        }))
171    }
172
173    /// Stop the carrier and close its application-owned relay client.
174    pub async fn stop(mut self) {
175        if let Some(shutdown) = self.shutdown.take() {
176            let _ = shutdown.send(());
177        }
178        let _ = (&mut self.task).await;
179        self.client.shutdown().await;
180    }
181}
182
183impl Drop for NostrRelayAdapter {
184    fn drop(&mut self) {
185        if let Some(shutdown) = self.shutdown.take() {
186            let _ = shutdown.send(());
187        }
188        self.task.abort();
189    }
190}
191
192#[cfg(test)]
193mod tests {
194    use super::*;
195
196    #[test]
197    fn nostr_relay_outbound_poll_delay_is_adaptive() {
198        assert_eq!(
199            outbound_poll_delay(1),
200            NOSTR_RELAY_ACTIVE_OUTBOUND_POLL_INTERVAL
201        );
202        assert_eq!(
203            outbound_poll_delay(0),
204            NOSTR_RELAY_IDLE_OUTBOUND_POLL_INTERVAL
205        );
206        assert!(NOSTR_RELAY_ACTIVE_OUTBOUND_POLL_INTERVAL <= Duration::from_millis(10));
207        assert!(NOSTR_RELAY_IDLE_OUTBOUND_POLL_INTERVAL >= Duration::from_millis(250));
208    }
209}
fips_core/nostr_relay_adapter.rs

fips_core/
nostr_relay_adapter.rs
