1use std::collections::HashMap;
15use std::sync::{Arc, Mutex};
16use std::time::{SystemTime, UNIX_EPOCH};
17
18use serde_json::Value;
19
20use crate::admin::AdminApiClient;
21use crate::config::{CachedToken, RpcConfig};
22use crate::errors::SdkError;
23use crate::SdkConfig;
24
25const DEFAULT_REFRESH_MARGIN_SECS: i64 = 60;
28
29#[derive(Clone)]
31pub struct RpcApiClient {
32 admin: AdminApiClient,
34 config: SdkConfig,
35 refresh_margin_secs: i64,
36 cache: Arc<Mutex<Option<CachedToken>>>,
39 refresh_lock: Arc<tokio::sync::Mutex<()>>,
43 networks: Arc<Mutex<Option<HashMap<String, String>>>>,
49 endpoint_url: Option<String>,
53}
54
55impl std::fmt::Debug for RpcApiClient {
56 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
57 f.debug_struct("RpcApiClient")
59 .field("refresh_margin_secs", &self.refresh_margin_secs)
60 .field(
61 "has_cached_token",
62 &self.cache.lock().is_ok_and(|c| c.is_some()),
63 )
64 .finish()
65 }
66}
67
68impl RpcApiClient {
69 pub fn new(config: SdkConfig, rpc_config: Option<&RpcConfig>) -> Self {
70 let refresh_margin_secs = rpc_config
71 .and_then(|c| c.refresh_margin_secs)
72 .filter(|&m| m >= 0)
73 .unwrap_or(DEFAULT_REFRESH_MARGIN_SECS);
74 let seed = rpc_config.and_then(|c| c.seed.clone());
77 let networks = rpc_config.and_then(|c| c.networks.clone());
78 let endpoint_url = rpc_config.and_then(|c| c.endpoint_url.clone());
79 Self {
80 admin: AdminApiClient::new(config.clone()),
81 config,
82 refresh_margin_secs,
83 cache: Arc::new(Mutex::new(seed)),
84 refresh_lock: Arc::new(tokio::sync::Mutex::new(())),
85 networks: Arc::new(Mutex::new(networks)),
86 endpoint_url,
87 }
88 }
89
90 pub fn set_networks(&self, networks: HashMap<String, String>) {
96 if let Ok(mut guard) = self.networks.lock() {
97 *guard = Some(networks);
98 }
99 }
100
101 pub fn current_token(&self) -> Option<CachedToken> {
105 self.cache.lock().ok().and_then(|c| c.clone())
106 }
107
108 pub fn clear_cached_token(&self) {
112 self.invalidate();
113 }
114
115 pub async fn call(
134 &self,
135 method: &str,
136 params: Option<Value>,
137 network: Option<String>,
138 endpoint_url: Option<String>,
139 ) -> Result<Value, SdkError> {
140 if endpoint_url.is_some() && network.is_some() {
145 return Err(SdkError::Config(
146 "`endpoint_url` and `network` are mutually exclusive: a custom \
147 URL is not multichain-routed"
148 .into(),
149 ));
150 }
151 let custom_url = endpoint_url.or_else(|| self.endpoint_url.clone());
152
153 if let Some(url) = custom_url {
156 let resp = self.send(None, &url, method, ¶ms).await?;
157 return Self::parse_rpc(resp);
158 }
159
160 let token = self.valid_token().await?;
162 let url = self.resolve_url(&token, network.as_deref())?;
163 let resp = self.send(Some(&token), &url, method, ¶ms).await?;
164
165 if resp.status == 401 {
169 self.invalidate();
170 let token = self.refresh().await?;
171 let url = self.resolve_url(&token, network.as_deref())?;
172 let retry = self.send(Some(&token), &url, method, ¶ms).await?;
173 return Self::parse_rpc(retry);
174 }
175 Self::parse_rpc(resp)
176 }
177
178 fn resolve_url(&self, token: &CachedToken, network: Option<&str>) -> Result<String, SdkError> {
182 let Some(key) = network else {
183 return Ok(token.endpoint_url.clone());
184 };
185 let guard = self
186 .networks
187 .lock()
188 .map_err(|_| SdkError::Config("network map lock poisoned".into()))?;
189 let Some(map) = guard.as_ref() else {
190 return Err(SdkError::Config(format!(
191 "network '{key}' requested but no network map is available; \
192 seed it via RpcConfig.networks or set_networks()"
193 )));
194 };
195 match map.get(key) {
196 Some(url) => Ok(url.clone()),
197 None => {
198 let mut keys: Vec<&str> = map.keys().map(String::as_str).collect();
199 keys.sort_unstable();
200 Err(SdkError::Config(format!(
201 "unknown network '{key}'. Available: {}",
202 keys.join(", ")
203 )))
204 }
205 }
206 }
207
208 async fn valid_token(&self) -> Result<CachedToken, SdkError> {
212 if let Some(tok) = self.cached_if_fresh() {
213 return Ok(tok);
214 }
215 self.refresh().await
216 }
217
218 fn cached_if_fresh(&self) -> Option<CachedToken> {
220 let now = now_unix();
221 let guard = self.cache.lock().ok()?;
222 guard
223 .as_ref()
224 .filter(|t| now + self.refresh_margin_secs < t.exp_unix)
225 .cloned()
226 }
227
228 async fn refresh(&self) -> Result<CachedToken, SdkError> {
231 let _guard = self.refresh_lock.lock().await;
232 if let Some(tok) = self.cached_if_fresh() {
234 return Ok(tok);
235 }
236 let fresh = self.admin.mint_tooling_token().await?;
237 if let Ok(mut guard) = self.cache.lock() {
238 *guard = Some(fresh.clone());
239 }
240 Ok(fresh)
241 }
242
243 fn invalidate(&self) {
244 if let Ok(mut guard) = self.cache.lock() {
245 *guard = None;
246 }
247 }
248
249 async fn send(
257 &self,
258 token: Option<&CachedToken>,
259 target_url: &str,
260 method: &str,
261 params: &Option<Value>,
262 ) -> Result<RawResponse, SdkError> {
263 let url = reqwest::Url::parse(target_url).map_err(|e| SdkError::Config(e.to_string()))?;
264 let body = serde_json::json!({
265 "jsonrpc": "2.0",
266 "id": 1,
267 "method": method,
268 "params": params.clone().unwrap_or_else(|| Value::Array(vec![])),
269 });
270 let mut req = self.config.rpc_http_client().post(url).json(&body);
271 if let Some(token) = token {
272 req = req.bearer_auth(&token.token);
273 }
274 let resp = req.send().await.map_err(SdkError::Http)?;
275 let status = resp.status().as_u16();
276 let text = resp.text().await.map_err(SdkError::Http)?;
277 Ok(RawResponse { status, text })
278 }
279
280 fn parse_rpc(resp: RawResponse) -> Result<Value, SdkError> {
283 let parsed: Result<JsonRpcEnvelope, _> = serde_json::from_str(&resp.text);
286 match parsed {
287 Ok(env) => {
288 if let Some(err) = env.error {
289 return Err(SdkError::Rpc {
290 code: err.code,
291 message: err.message,
292 });
293 }
294 if let Some(result) = env.result {
295 return Ok(result);
296 }
297 if !(200..300).contains(&resp.status) {
300 return Err(SdkError::Api {
301 status: status_code(resp.status),
302 body: resp.text,
303 });
304 }
305 Ok(Value::Null)
306 }
307 Err(source) => {
308 if !(200..300).contains(&resp.status) {
309 Err(SdkError::Api {
310 status: status_code(resp.status),
311 body: resp.text,
312 })
313 } else {
314 Err(SdkError::Decode {
315 source,
316 body: resp.text,
317 })
318 }
319 }
320 }
321 }
322}
323
324struct RawResponse {
325 status: u16,
326 text: String,
327}
328
329#[derive(serde::Deserialize)]
330struct JsonRpcEnvelope {
331 #[serde(default)]
332 result: Option<Value>,
333 #[serde(default)]
334 error: Option<JsonRpcError>,
335}
336
337#[derive(serde::Deserialize)]
338struct JsonRpcError {
339 code: i64,
340 message: String,
341}
342
343fn now_unix() -> i64 {
344 SystemTime::now()
345 .duration_since(UNIX_EPOCH)
346 .map_or(0, |d| d.as_secs() as i64)
349}
350
351fn status_code(status: u16) -> reqwest::StatusCode {
352 reqwest::StatusCode::from_u16(status).unwrap_or(reqwest::StatusCode::BAD_GATEWAY)
353}
354
355#[cfg(test)]
356#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
357mod tests {
358 use super::*;
359 use crate::config::{AdminConfig, SdkFullConfig};
360 use crate::QuicknodeSdk;
361 use std::sync::atomic::{AtomicUsize, Ordering};
362 use wiremock::matchers::{body_partial_json, header, method, path};
363 use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate};
364
365 fn future_exp() -> i64 {
367 now_unix() + 3600
368 }
369
370 fn token_body(endpoint_url: &str, exp: i64) -> serde_json::Value {
371 let _ = exp;
374 serde_json::json!({
375 "data": {
376 "endpoint_url": endpoint_url,
377 "token": "minted.jwt.value",
378 "expires_at": "2099-01-01T00:00:00.000Z"
379 },
380 "error": null
381 })
382 }
383
384 fn sdk_with_seed(admin_base: &str, rpc_endpoint: &str) -> QuicknodeSdk {
385 let mut cfg = SdkFullConfig::from_api_key("test-key".to_string());
386 cfg.admin = Some(AdminConfig {
387 base_url: Some(format!("{admin_base}/")),
388 });
389 cfg.rpc = Some(RpcConfig {
390 endpoint_url: None,
391 seed: Some(CachedToken {
392 endpoint_url: rpc_endpoint.to_string(),
393 token: "seeded.jwt".to_string(),
394 exp_unix: future_exp(),
395 }),
396 refresh_margin_secs: None,
397 networks: None,
398 });
399 QuicknodeSdk::new(&cfg).unwrap()
400 }
401
402 #[tokio::test]
403 async fn call_uses_seed_without_minting() {
404 let server = MockServer::start().await;
405 Mock::given(method("POST"))
407 .and(path("/"))
408 .and(body_partial_json(
409 serde_json::json!({ "method": "eth_blockNumber" }),
410 ))
411 .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
412 "jsonrpc": "2.0", "id": 1, "result": "0x1335f9a"
413 })))
414 .mount(&server)
415 .await;
416
417 let sdk = sdk_with_seed(&server.uri(), &server.uri());
420 let result = sdk
421 .rpc
422 .call("eth_blockNumber", None, None, None)
423 .await
424 .unwrap();
425 assert_eq!(result, serde_json::json!("0x1335f9a"));
426 }
427
428 #[tokio::test]
429 async fn call_sends_bearer_jwt_but_not_account_api_key() {
430 let server = MockServer::start().await;
431 Mock::given(method("POST"))
435 .and(path("/"))
436 .and(header("authorization", "Bearer seeded.jwt"))
437 .and(|req: &Request| !req.headers.contains_key("x-api-key"))
438 .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
439 "jsonrpc": "2.0", "id": 1, "result": "0xok"
440 })))
441 .mount(&server)
442 .await;
443
444 let sdk = sdk_with_seed(&server.uri(), &server.uri());
445 let result = sdk
446 .rpc
447 .call("eth_blockNumber", None, None, None)
448 .await
449 .unwrap();
450 assert_eq!(result, serde_json::json!("0xok"));
451 }
452
453 fn sdk_with_custom_url(endpoint_url: &str) -> QuicknodeSdk {
457 let mut cfg = SdkFullConfig::from_api_key("test-key".to_string());
458 cfg.admin = Some(AdminConfig {
459 base_url: Some("http://127.0.0.1:1/".to_string()),
460 });
461 cfg.rpc = Some(RpcConfig {
462 endpoint_url: Some(endpoint_url.to_string()),
463 seed: None,
464 refresh_margin_secs: None,
465 networks: None,
466 });
467 QuicknodeSdk::new(&cfg).unwrap()
468 }
469
470 #[tokio::test]
471 async fn config_endpoint_url_bypasses_jwt_and_minting() {
472 let server = MockServer::start().await;
473 Mock::given(method("POST"))
477 .and(path("/custom"))
478 .and(|req: &Request| {
479 !req.headers.contains_key("authorization") && !req.headers.contains_key("x-api-key")
480 })
481 .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
482 "jsonrpc": "2.0", "id": 1, "result": "0xcustom"
483 })))
484 .mount(&server)
485 .await;
486
487 let sdk = sdk_with_custom_url(&format!("{}/custom", server.uri()));
488 let result = sdk
489 .rpc
490 .call("eth_blockNumber", None, None, None)
491 .await
492 .unwrap();
493 assert_eq!(result, serde_json::json!("0xcustom"));
494 assert!(sdk.rpc.current_token().is_none());
496 }
497
498 #[tokio::test]
499 async fn per_call_endpoint_url_overrides_config_default() {
500 let server = MockServer::start().await;
501 Mock::given(method("POST"))
504 .and(path("/override"))
505 .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
506 "jsonrpc": "2.0", "id": 1, "result": "0xoverride"
507 })))
508 .mount(&server)
509 .await;
510
511 let sdk = sdk_with_custom_url(&format!("{}/wrong", server.uri()));
512 let result = sdk
513 .rpc
514 .call(
515 "eth_blockNumber",
516 None,
517 None,
518 Some(format!("{}/override", server.uri())),
519 )
520 .await
521 .unwrap();
522 assert_eq!(result, serde_json::json!("0xoverride"));
523 }
524
525 #[tokio::test]
526 async fn endpoint_url_and_network_together_is_config_error() {
527 let sdk = sdk_with_custom_url("https://example.invalid/rpc");
528 let err = sdk
529 .rpc
530 .call(
531 "eth_blockNumber",
532 None,
533 Some("solana-mainnet".to_string()),
534 Some("https://example.invalid/other".to_string()),
535 )
536 .await
537 .unwrap_err();
538 assert!(matches!(err, SdkError::Config(msg) if msg.contains("mutually exclusive")));
539 }
540
541 #[tokio::test]
542 async fn json_rpc_error_maps_to_rpc_error() {
543 let server = MockServer::start().await;
544 Mock::given(method("POST"))
545 .and(path("/"))
546 .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
547 "jsonrpc": "2.0", "id": 1,
548 "error": { "code": -32602, "message": "invalid params" }
549 })))
550 .mount(&server)
551 .await;
552
553 let sdk = sdk_with_seed(&server.uri(), &server.uri());
554 let err = sdk
555 .rpc
556 .call("eth_getBalance", None, None, None)
557 .await
558 .unwrap_err();
559 match err {
560 SdkError::Rpc { code, message } => {
561 assert_eq!(code, -32602);
562 assert!(message.contains("invalid params"));
563 }
564 other => panic!("expected Rpc error, got {other:?}"),
565 }
566 }
567
568 #[tokio::test]
569 async fn reactive_401_refreshes_and_retries_once() {
570 let server = MockServer::start().await;
571
572 struct Sequence {
574 calls: AtomicUsize,
575 }
576 impl Respond for Sequence {
577 fn respond(&self, _: &Request) -> ResponseTemplate {
578 let n = self.calls.fetch_add(1, Ordering::SeqCst);
579 if n == 0 {
580 ResponseTemplate::new(401).set_body_string("unauthorized")
581 } else {
582 ResponseTemplate::new(200).set_body_json(serde_json::json!({
583 "jsonrpc": "2.0", "id": 1, "result": "0xokay"
584 }))
585 }
586 }
587 }
588
589 Mock::given(method("POST"))
591 .and(path("/rpc"))
592 .respond_with(Sequence {
593 calls: AtomicUsize::new(0),
594 })
595 .mount(&server)
596 .await;
597 Mock::given(method("POST"))
598 .and(path("/tooling-access/token"))
599 .respond_with(
600 ResponseTemplate::new(200)
601 .set_body_json(token_body(&format!("{}/rpc", server.uri()), future_exp())),
602 )
603 .mount(&server)
604 .await;
605
606 let sdk = sdk_with_seed(&server.uri(), &format!("{}/rpc", server.uri()));
607 let result = sdk
608 .rpc
609 .call("eth_blockNumber", None, None, None)
610 .await
611 .unwrap();
612 assert_eq!(result, serde_json::json!("0xokay"));
613 }
614
615 #[tokio::test]
616 async fn second_401_surfaces_as_api_error() {
617 let server = MockServer::start().await;
618 Mock::given(method("POST"))
619 .and(path("/rpc"))
620 .respond_with(ResponseTemplate::new(401).set_body_string("nope"))
621 .mount(&server)
622 .await;
623 Mock::given(method("POST"))
624 .and(path("/tooling-access/token"))
625 .respond_with(
626 ResponseTemplate::new(200)
627 .set_body_json(token_body(&format!("{}/rpc", server.uri()), future_exp())),
628 )
629 .mount(&server)
630 .await;
631
632 let sdk = sdk_with_seed(&server.uri(), &format!("{}/rpc", server.uri()));
633 let err = sdk
634 .rpc
635 .call("eth_blockNumber", None, None, None)
636 .await
637 .unwrap_err();
638 assert!(matches!(err, SdkError::Api { status, .. } if status.as_u16() == 401));
639 }
640
641 #[tokio::test]
642 async fn expired_seed_triggers_mint() {
643 let server = MockServer::start().await;
644 Mock::given(method("POST"))
645 .and(path("/tooling-access/token"))
646 .respond_with(
647 ResponseTemplate::new(200)
648 .set_body_json(token_body(&format!("{}/rpc", server.uri()), future_exp())),
649 )
650 .mount(&server)
651 .await;
652 Mock::given(method("POST"))
653 .and(path("/rpc"))
654 .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
655 "jsonrpc": "2.0", "id": 1, "result": "0xfresh"
656 })))
657 .mount(&server)
658 .await;
659
660 let mut cfg = SdkFullConfig::from_api_key("test-key".to_string());
662 cfg.admin = Some(AdminConfig {
663 base_url: Some(format!("{}/", server.uri())),
664 });
665 cfg.rpc = Some(RpcConfig {
666 endpoint_url: None,
667 seed: Some(CachedToken {
668 endpoint_url: format!("{}/rpc", server.uri()),
669 token: "expired.jwt".to_string(),
670 exp_unix: now_unix() - 10,
671 }),
672 refresh_margin_secs: None,
673 networks: None,
674 });
675 let sdk = QuicknodeSdk::new(&cfg).unwrap();
676
677 let result = sdk
678 .rpc
679 .call("eth_blockNumber", None, None, None)
680 .await
681 .unwrap();
682 assert_eq!(result, serde_json::json!("0xfresh"));
683 assert_eq!(sdk.rpc.current_token().unwrap().token, "minted.jwt.value");
685 }
686
687 #[tokio::test]
688 async fn network_routes_to_mapped_url() {
689 let server = MockServer::start().await;
690 Mock::given(method("POST"))
693 .and(path("/solana"))
694 .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
695 "jsonrpc": "2.0", "id": 1, "result": "12345"
696 })))
697 .mount(&server)
698 .await;
699
700 let mut cfg = SdkFullConfig::from_api_key("test-key".to_string());
701 cfg.admin = Some(AdminConfig {
702 base_url: Some(format!("{}/", server.uri())),
703 });
704 let mut networks = std::collections::HashMap::new();
705 networks.insert(
706 "solana-mainnet".to_string(),
707 format!("{}/solana", server.uri()),
708 );
709 cfg.rpc = Some(RpcConfig {
710 endpoint_url: None,
711 seed: Some(CachedToken {
712 endpoint_url: format!("{}/default", server.uri()),
713 token: "seeded.jwt".to_string(),
714 exp_unix: future_exp(),
715 }),
716 refresh_margin_secs: None,
717 networks: Some(networks),
718 });
719 let sdk = QuicknodeSdk::new(&cfg).unwrap();
720
721 let result = sdk
722 .rpc
723 .call("getSlot", None, Some("solana-mainnet".to_string()), None)
724 .await
725 .unwrap();
726 assert_eq!(result, serde_json::json!("12345"));
727 }
728
729 #[tokio::test]
730 async fn unknown_network_is_config_error_listing_keys() {
731 let server = MockServer::start().await;
732 let sdk = sdk_with_seed(&server.uri(), &server.uri());
733 sdk.rpc.set_networks(std::collections::HashMap::from([(
735 "solana-mainnet".to_string(),
736 "https://x/solana".to_string(),
737 )]));
738 let err = sdk
739 .rpc
740 .call("getSlot", None, Some("polygon".to_string()), None)
741 .await
742 .unwrap_err();
743 match err {
744 SdkError::Config(msg) => {
745 assert!(msg.contains("unknown network 'polygon'"), "msg: {msg}");
746 assert!(
747 msg.contains("solana-mainnet"),
748 "msg should list keys: {msg}"
749 );
750 }
751 other => panic!("expected Config error, got {other:?}"),
752 }
753 }
754
755 #[tokio::test]
756 async fn network_without_seeded_map_errors() {
757 let server = MockServer::start().await;
758 let sdk = sdk_with_seed(&server.uri(), &server.uri());
759 let err = sdk
760 .rpc
761 .call("getSlot", None, Some("solana-mainnet".to_string()), None)
762 .await
763 .unwrap_err();
764 assert!(matches!(err, SdkError::Config(msg) if msg.contains("no network map")));
765 }
766}