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
18/// Pumps signed kind-21060 datagrams between FIPS and application-selected
19/// Nostr relays.
20pub struct NostrRelayAdapter {
21    client: Client,
22    shutdown: Option<oneshot::Sender<()>>,
23    task: JoinHandle<()>,
24}
25
26#[derive(Clone)]
27enum RelayEventIo {
28    Endpoint(Arc<FipsEndpoint>),
29    Node(NostrRelayIo),
30}
31
32impl RelayEventIo {
33    fn npub(&self) -> &str {
34        match self {
35            Self::Endpoint(endpoint) => endpoint.npub(),
36            Self::Node(io) => io.npub(),
37        }
38    }
39
40    async fn drain_events(&self, limit: usize) -> Result<Vec<Event>, FipsEndpointError> {
41        match self {
42            Self::Endpoint(endpoint) => endpoint.drain_nostr_relay_events(limit).await,
43            Self::Node(io) => io.drain_events(limit).await,
44        }
45    }
46
47    async fn ingest_event(&self, event: Event) -> Result<bool, FipsEndpointError> {
48        match self {
49            Self::Endpoint(endpoint) => endpoint.ingest_nostr_event(event).await,
50            Self::Node(io) => io.ingest_event(event).await,
51        }
52    }
53}
54
55impl NostrRelayAdapter {
56    /// Connect an embedded endpoint through the application-provided relays.
57    pub async fn start(
58        endpoint: Arc<FipsEndpoint>,
59        relays: &[String],
60    ) -> Result<Option<Self>, String> {
61        Self::start_with_io(RelayEventIo::Endpoint(endpoint), relays).await
62    }
63
64    /// Connect a direct [`crate::Node`] through its attached relay I/O.
65    pub async fn start_for_node(
66        io: NostrRelayIo,
67        relays: &[String],
68    ) -> Result<Option<Self>, String> {
69        Self::start_with_io(RelayEventIo::Node(io), relays).await
70    }
71
72    async fn start_with_io(io: RelayEventIo, relays: &[String]) -> Result<Option<Self>, String> {
73        if relays.is_empty() {
74            return Ok(None);
75        }
76
77        let local_pubkey = PublicKey::parse(io.npub())
78            .map_err(|error| format!("invalid FIPS endpoint identity: {error}"))?;
79        let client = Client::new(Keys::generate());
80        for relay in relays {
81            client
82                .add_relay(relay)
83                .await
84                .map_err(|error| format!("failed to add Nostr relay {relay}: {error}"))?;
85        }
86        let mut notifications = client.notifications();
87        client.connect().await;
88        client
89            .subscribe(
90                Filter::new()
91                    .kind(Kind::Custom(NOSTR_RELAY_DATAGRAM_KIND))
92                    .pubkey(local_pubkey),
93                None,
94            )
95            .await
96            .map_err(|error| format!("failed to subscribe to FIPS relay datagrams: {error}"))?;
97
98        let task_client = client.clone();
99        let publish_slots = Arc::new(Semaphore::new(32));
100        let (shutdown_tx, mut shutdown_rx) = oneshot::channel();
101        let task = tokio::spawn(async move {
102            let mut outbound_tick = tokio::time::interval(Duration::from_millis(10));
103            outbound_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
104            loop {
105                tokio::select! {
106                    _ = &mut shutdown_rx => break,
107                    _ = outbound_tick.tick() => {
108                        match io.drain_events(64).await {
109                            Ok(events) => {
110                                for event in events {
111                                    let Ok(permit) = Arc::clone(&publish_slots).try_acquire_owned() else {
112                                        tracing::debug!(event_id = %event.id, "FIPS relay publish queue is saturated");
113                                        continue;
114                                    };
115                                    let client = task_client.clone();
116                                    tokio::spawn(async move {
117                                        let _permit = permit;
118                                        if let Err(error) = client.send_event(&event).await {
119                                            tracing::debug!(%error, event_id = %event.id, "failed to publish FIPS relay datagram");
120                                        }
121                                    });
122                                }
123                            }
124                            Err(error) => tracing::debug!(%error, "failed to drain FIPS relay datagrams"),
125                        }
126                    }
127                    notification = notifications.recv() => {
128                        match notification {
129                            Ok(RelayPoolNotification::Event { event, .. })
130                                if event.kind == Kind::Custom(NOSTR_RELAY_DATAGRAM_KIND) =>
131                            {
132                                if let Err(error) = io.ingest_event((*event).clone()).await {
133                                    tracing::debug!(%error, "failed to ingest FIPS relay datagram");
134                                }
135                            }
136                            Ok(_) => {}
137                            Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
138                                tracing::debug!(skipped, "FIPS relay notification receiver lagged");
139                            }
140                            Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
141                        }
142                    }
143                }
144            }
145        });
146
147        Ok(Some(Self {
148            client,
149            shutdown: Some(shutdown_tx),
150            task,
151        }))
152    }
153
154    /// Stop the carrier and close its application-owned relay client.
155    pub async fn stop(mut self) {
156        if let Some(shutdown) = self.shutdown.take() {
157            let _ = shutdown.send(());
158        }
159        let _ = (&mut self.task).await;
160        self.client.shutdown().await;
161    }
162}
163
164impl Drop for NostrRelayAdapter {
165    fn drop(&mut self) {
166        if let Some(shutdown) = self.shutdown.take() {
167            let _ = shutdown.send(());
168        }
169        self.task.abort();
170    }
171}
fips_core/nostr_relay_adapter.rs

fips_core/
nostr_relay_adapter.rs
