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

rings_node/
rpc_impl.rs
