rings_node/
rpc_impl.rs

1#![warn(missing_docs)]
2
3//! RPC handler for both feature=browser and feature=node.
4//! We support handling the RPC request in either native or browser environment by `InternalRpcHandler` and `ExternalRpcHandler` from rings_rpc crate.
5//! For the native environment, we use jsonrpc_core to handle requests.
6//! For the browser environment, we use `InternalRpcHandler` to process the requests.
7
8use std::collections::HashSet;
9use std::str::FromStr;
10
11use async_trait::async_trait;
12use futures::future::join_all;
13use jsonrpc_core::types::error::Error;
14use jsonrpc_core::types::error::ErrorCode;
15use jsonrpc_core::Result;
16use rings_core::dht::Did;
17use rings_core::ecc::PublicKey;
18use rings_core::message::e2e;
19use rings_core::message::Decoder;
20use rings_core::message::Encoded;
21use rings_core::message::Encoder;
22use rings_core::message::MessagePayload;
23use rings_core::message::MessageVerificationExt;
24use rings_core::prelude::entry::Entry;
25use rings_rpc::protos::rings_node::*;
26use rings_rpc::protos::rings_node_handler::HandleRpc;
27
28use crate::error::Error as ServerError;
29use crate::processor::Processor;
30use crate::seed::Seed;
31
32#[cfg_attr(feature = "browser", async_trait(?Send))]
33#[cfg_attr(not(feature = "browser"), async_trait)]
34impl HandleRpc<ConnectPeerViaHttpRequest, ConnectPeerViaHttpResponse> for Processor {
35    async fn handle_rpc(
36        &self,
37        req: ConnectPeerViaHttpRequest,
38    ) -> Result<ConnectPeerViaHttpResponse> {
39        let client = rings_rpc::jsonrpc::Client::new(&req.url);
40
41        let did = client
42            .node_did(&NodeDidRequest {})
43            .await
44            .map_err(|e| ServerError::RemoteRpcError(e.to_string()))?
45            .did;
46
47        let offer = self
48            .handle_rpc(CreateOfferRequest { did: did.clone() })
49            .await?
50            .offer;
51
52        let answer = client
53            .answer_offer(&AnswerOfferRequest { offer })
54            .await
55            .map_err(|e| ServerError::RemoteRpcError(e.to_string()))?
56            .answer;
57
58        self.handle_rpc(AcceptAnswerRequest { answer }).await?;
59
60        Ok(ConnectPeerViaHttpResponse { did })
61    }
62}
63
64#[cfg_attr(feature = "browser", async_trait(?Send))]
65#[cfg_attr(not(feature = "browser"), async_trait)]
66impl HandleRpc<ConnectWithDidRequest, ConnectWithDidResponse> for Processor {
67    async fn handle_rpc(&self, req: ConnectWithDidRequest) -> Result<ConnectWithDidResponse> {
68        let did = s2d(&req.did)?;
69        self.connect_with_did(did).await.map_err(Error::from)?;
70        Ok(ConnectWithDidResponse {})
71    }
72}
73
74#[cfg_attr(feature = "browser", async_trait(?Send))]
75#[cfg_attr(not(feature = "browser"), async_trait)]
76impl HandleRpc<ConnectWithSeedRequest, ConnectWithSeedResponse> for Processor {
77    async fn handle_rpc(&self, req: ConnectWithSeedRequest) -> Result<ConnectWithSeedResponse> {
78        let seed: Seed = Seed::try_from(req)?;
79
80        let mut connected: HashSet<String> =
81            HashSet::from_iter(self.swarm.peers().into_iter().map(|peer| peer.did));
82        connected.insert(self.swarm.did().to_string());
83
84        let tasks = seed
85            .peers
86            .iter()
87            .filter(|&x| !connected.contains(&x.did))
88            .map(|x| {
89                self.handle_rpc(ConnectPeerViaHttpRequest {
90                    url: x.url.to_string(),
91                })
92            });
93
94        let results = join_all(tasks).await;
95
96        let first_err = results.into_iter().find(|x| x.is_err());
97        if let Some(err) = first_err {
98            err?;
99        }
100
101        Ok(ConnectWithSeedResponse {})
102    }
103}
104
105#[cfg_attr(feature = "browser", async_trait(?Send))]
106#[cfg_attr(not(feature = "browser"), async_trait)]
107impl HandleRpc<ListPeersRequest, ListPeersResponse> for Processor {
108    async fn handle_rpc(&self, _req: ListPeersRequest) -> Result<ListPeersResponse> {
109        let peers = self
110            .swarm
111            .peers()
112            .into_iter()
113            .map(|peer| peer.into())
114            .collect();
115        Ok(ListPeersResponse { peers })
116    }
117}
118
119#[cfg_attr(feature = "browser", async_trait(?Send))]
120#[cfg_attr(not(feature = "browser"), async_trait)]
121impl HandleRpc<CreateOfferRequest, CreateOfferResponse> for Processor {
122    async fn handle_rpc(&self, req: CreateOfferRequest) -> Result<CreateOfferResponse> {
123        let did = s2d(&req.did)?;
124        let offer_payload = self
125            .swarm
126            .create_offer(did)
127            .await
128            .map_err(ServerError::CreateOffer)
129            .map_err(Error::from)?;
130
131        let encoded = offer_payload
132            .encode()
133            .map_err(|_| ServerError::EncodeError)?;
134
135        Ok(CreateOfferResponse {
136            offer: encoded.to_string(),
137        })
138    }
139}
140
141#[cfg_attr(feature = "browser", async_trait(?Send))]
142#[cfg_attr(not(feature = "browser"), async_trait)]
143impl HandleRpc<AnswerOfferRequest, AnswerOfferResponse> for Processor {
144    async fn handle_rpc(&self, req: AnswerOfferRequest) -> Result<AnswerOfferResponse> {
145        if req.offer.is_empty() {
146            return Err(Error::invalid_params("Offer is empty"));
147        }
148        let encoded: Encoded = <Encoded as From<String>>::from(req.offer);
149
150        let offer_payload =
151            MessagePayload::from_encoded(&encoded).map_err(|_| ServerError::DecodeError)?;
152
153        let answer_payload = self
154            .swarm
155            .answer_offer(offer_payload)
156            .await
157            .map_err(ServerError::AnswerOffer)
158            .map_err(Error::from)?;
159
160        tracing::debug!("connect_peer_via_ice response: {:?}", answer_payload);
161        let encoded = answer_payload
162            .encode()
163            .map_err(|_| ServerError::EncodeError)?;
164
165        Ok(AnswerOfferResponse {
166            answer: encoded.to_string(),
167        })
168    }
169}
170
171#[cfg_attr(feature = "browser", async_trait(?Send))]
172#[cfg_attr(not(feature = "browser"), async_trait)]
173impl HandleRpc<AcceptAnswerRequest, AcceptAnswerResponse> for Processor {
174    async fn handle_rpc(&self, req: AcceptAnswerRequest) -> Result<AcceptAnswerResponse> {
175        if req.answer.is_empty() {
176            return Err(Error::invalid_params("Answer is empty"));
177        }
178        let encoded = Encoded::from(req.answer);
179
180        let answer_payload =
181            MessagePayload::from_encoded(&encoded).map_err(|_| ServerError::DecodeError)?;
182        answer_payload.transaction.signer();
183
184        self.swarm
185            .accept_answer(answer_payload)
186            .await
187            .map_err(ServerError::AcceptAnswer)?;
188
189        Ok(AcceptAnswerResponse {})
190    }
191}
192
193#[cfg_attr(feature = "browser", async_trait(?Send))]
194#[cfg_attr(not(feature = "browser"), async_trait)]
195impl HandleRpc<DisconnectRequest, DisconnectResponse> for Processor {
196    async fn handle_rpc(&self, req: DisconnectRequest) -> Result<DisconnectResponse> {
197        let did = s2d(&req.did)?;
198        self.disconnect(did).await?;
199        Ok(DisconnectResponse {})
200    }
201}
202
203#[cfg_attr(feature = "browser", async_trait(?Send))]
204#[cfg_attr(not(feature = "browser"), async_trait)]
205impl HandleRpc<SendBackendMessageRequest, SendBackendMessageResponse> for Processor {
206    async fn handle_rpc(
207        &self,
208        req: SendBackendMessageRequest,
209    ) -> Result<SendBackendMessageResponse> {
210        let destination = s2d(&req.destination_did)?;
211        let payload = base64::decode(req.data.as_str())
212            .map_err(|e| Error::invalid_params(format!("data is not valid base64: {e:?}")))?;
213        let envelope =
214            crate::extension::ext::Envelope::new(req.namespace, bytes::Bytes::from(payload));
215        self.send_envelope(destination, &envelope).await?;
216        Ok(SendBackendMessageResponse {})
217    }
218}
219
220#[cfg_attr(feature = "browser", async_trait(?Send))]
221#[cfg_attr(not(feature = "browser"), async_trait)]
222impl HandleRpc<SendE2eHandshakeRequest, SendE2eHandshakeResponse> for Processor {
223    async fn handle_rpc(&self, req: SendE2eHandshakeRequest) -> Result<SendE2eHandshakeResponse> {
224        let destination = s2d(&req.destination_did)?;
225        let tx_id = self.send_e2e_handshake(destination).await?;
226        Ok(SendE2eHandshakeResponse {
227            tx_id: tx_id.to_string(),
228        })
229    }
230}
231
232#[cfg_attr(feature = "browser", async_trait(?Send))]
233#[cfg_attr(not(feature = "browser"), async_trait)]
234impl HandleRpc<SendE2eMessageRequest, SendE2eMessageResponse> for Processor {
235    async fn handle_rpc(&self, req: SendE2eMessageRequest) -> Result<SendE2eMessageResponse> {
236        let destination = s2d(&req.destination_did)?;
237        let recipient_public_key = s2pk(&req.recipient_public_key)?;
238        let payload = base64::decode(req.data.as_str())
239            .map_err(|e| Error::invalid_params(format!("data is not valid base64: {e:?}")))?;
240        let frame_len = if req.max_plaintext_frame_len == 0 {
241            e2e::DEFAULT_E2E_PLAINTEXT_FRAME_LEN
242        } else {
243            req.max_plaintext_frame_len as usize
244        };
245
246        let stream_id = self
247            .send_e2e_message_with_frame_len(destination, recipient_public_key, &payload, frame_len)
248            .await?;
249        Ok(SendE2eMessageResponse {
250            stream_id: stream_id.to_string(),
251        })
252    }
253}
254
255#[cfg_attr(feature = "browser", async_trait(?Send))]
256#[cfg_attr(not(feature = "browser"), async_trait)]
257impl HandleRpc<PublishMessageToTopicRequest, PublishMessageToTopicResponse> for Processor {
258    async fn handle_rpc(
259        &self,
260        req: PublishMessageToTopicRequest,
261    ) -> Result<PublishMessageToTopicResponse> {
262        let encoded = req
263            .data
264            .encode()
265            .map_err(|e| Error::invalid_params(format!("Failed to encode data: {e:?}")))?;
266        self.storage_append_data(&req.topic, encoded).await?;
267        Ok(PublishMessageToTopicResponse {})
268    }
269}
270
271#[cfg_attr(feature = "browser", async_trait(?Send))]
272#[cfg_attr(not(feature = "browser"), async_trait)]
273impl HandleRpc<FetchTopicMessagesRequest, FetchTopicMessagesResponse> for Processor {
274    async fn handle_rpc(
275        &self,
276        req: FetchTopicMessagesRequest,
277    ) -> Result<FetchTopicMessagesResponse> {
278        let entry_key = Entry::gen_did(&req.topic)
279            .map_err(|_| Error::invalid_params("Failed to get id of topic"))?;
280
281        self.storage_fetch(entry_key).await?;
282        let result = self.storage_check_cache(entry_key).await;
283
284        let Some(entry) = result else {
285            return Ok(FetchTopicMessagesResponse { data: vec![] });
286        };
287
288        let data = entry
289            .data
290            .iter()
291            .skip(req.skip as usize)
292            .map(|v| v.decode())
293            .filter_map(|v| v.ok())
294            .collect::<Vec<String>>();
295
296        Ok(FetchTopicMessagesResponse { data })
297    }
298}
299
300#[cfg_attr(feature = "browser", async_trait(?Send))]
301#[cfg_attr(not(feature = "browser"), async_trait)]
302impl HandleRpc<RegisterServiceRequest, RegisterServiceResponse> for Processor {
303    async fn handle_rpc(&self, req: RegisterServiceRequest) -> Result<RegisterServiceResponse> {
304        self.register_service(&req.name).await?;
305        Ok(RegisterServiceResponse {})
306    }
307}
308
309#[cfg_attr(feature = "browser", async_trait(?Send))]
310#[cfg_attr(not(feature = "browser"), async_trait)]
311impl HandleRpc<LookupServiceRequest, LookupServiceResponse> for Processor {
312    async fn handle_rpc(&self, req: LookupServiceRequest) -> Result<LookupServiceResponse> {
313        let entry_key = Entry::gen_did(&req.name)
314            .map_err(|_| Error::invalid_params("Failed to get id of topic"))?;
315
316        self.storage_fetch(entry_key).await?;
317        let result = self.storage_check_cache(entry_key).await;
318
319        let Some(entry) = result else {
320            return Ok(LookupServiceResponse { dids: vec![] });
321        };
322
323        let dids = entry
324            .data
325            .iter()
326            .map(|v| v.decode())
327            .filter_map(|v| v.ok())
328            .collect::<Vec<String>>();
329
330        Ok(LookupServiceResponse { dids })
331    }
332}
333
334#[cfg_attr(feature = "browser", async_trait(?Send))]
335#[cfg_attr(not(feature = "browser"), async_trait)]
336impl HandleRpc<NodeInfoRequest, NodeInfoResponse> for Processor {
337    async fn handle_rpc(&self, _req: NodeInfoRequest) -> Result<NodeInfoResponse> {
338        self.get_node_info()
339            .await
340            .map_err(|_| Error::new(ErrorCode::InternalError))
341    }
342}
343
344#[cfg_attr(feature = "browser", async_trait(?Send))]
345#[cfg_attr(not(feature = "browser"), async_trait)]
346impl HandleRpc<NodeDidRequest, NodeDidResponse> for Processor {
347    async fn handle_rpc(&self, _req: NodeDidRequest) -> Result<NodeDidResponse> {
348        let did = self.did();
349        Ok(NodeDidResponse {
350            did: did.to_string(),
351        })
352    }
353}
354
355/// Get did from string or return InvalidParam Error
356fn s2d(s: &str) -> Result<Did> {
357    Did::from_str(s).map_err(|_| Error::invalid_params(format!("Invalid Did: {s}")))
358}
359
360fn s2pk(s: &str) -> Result<PublicKey<33>> {
361    PublicKey::try_from_b58m(s)
362        .or_else(|_| PublicKey::from_hex_string(s))
363        .map_err(|_| Error::invalid_params("Invalid secp256k1 public key"))
364}
rings_node/rpc_impl.rs

rings_node/
rpc_impl.rs
