fips_core/endpoint/
nostr_relay_io.rs1use tokio::sync::{mpsc, oneshot};
2
3use super::{ENDPOINT_OPERATION_TIMEOUT, FipsEndpointError};
4use crate::node::NodeEndpointControlCommand;
5
6#[derive(Clone)]
13pub struct NostrRelayIo {
14 npub: String,
15 control_tx: mpsc::Sender<NodeEndpointControlCommand>,
16}
17
18impl NostrRelayIo {
19 pub(crate) fn new(npub: String, control_tx: mpsc::Sender<NodeEndpointControlCommand>) -> Self {
20 Self { npub, control_tx }
21 }
22
23 pub fn npub(&self) -> &str {
25 &self.npub
26 }
27
28 async fn control<T>(
29 &self,
30 operation: &'static str,
31 command: NodeEndpointControlCommand,
32 response_rx: oneshot::Receiver<T>,
33 ) -> Result<T, FipsEndpointError> {
34 tokio::time::timeout(ENDPOINT_OPERATION_TIMEOUT, async {
35 self.control_tx
36 .send(command)
37 .await
38 .map_err(|_| FipsEndpointError::Closed)?;
39 response_rx.await.map_err(|_| FipsEndpointError::Closed)
40 })
41 .await
42 .map_err(|_| FipsEndpointError::Timeout { operation })?
43 }
44
45 pub async fn ingest_event(&self, event: nostr::Event) -> Result<bool, FipsEndpointError> {
47 let (response_tx, response_rx) = oneshot::channel();
48 self.control(
49 "Nostr relay event ingest",
50 NodeEndpointControlCommand::IngestNostrEvent { event, response_tx },
51 response_rx,
52 )
53 .await
54 }
55
56 pub async fn drain_events(&self, limit: usize) -> Result<Vec<nostr::Event>, FipsEndpointError> {
58 let (response_tx, response_rx) = oneshot::channel();
59 self.control(
60 "Nostr relay event drain",
61 NodeEndpointControlCommand::DrainNostrRelayEvents {
62 limit: limit.max(1),
63 response_tx,
64 },
65 response_rx,
66 )
67 .await
68 }
69}