arete_server/
http_health.rs

1use crate::{health::HealthMonitor, ProgramAccountReaderFn};
2use anyhow::Result;
3use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
4use base64::Engine as _;
5use dashmap::DashMap;
6use http_body_util::BodyExt;
7use http_body_util::Full;
8use hyper::body::Bytes;
9use hyper::server::conn::http1;
10use hyper::service::service_fn;
11use hyper::{Request, Response, StatusCode};
12use hyper_util::rt::TokioIo;
13use reqwest::Client;
14use serde::Deserialize;
15use serde_json::{json, Value};
16use std::convert::Infallible;
17use std::env;
18use std::net::SocketAddr;
19use std::sync::Arc;
20use std::time::{Duration, SystemTime, UNIX_EPOCH};
21use tokio::net::TcpListener;
22use tracing::{error, info};
23
24use crate::websocket::auth::{AuthDecision, AuthDeny, ConnectionAuthRequest, WebSocketAuthPlugin};
25
26/// Configuration for the HTTP health server
27#[derive(Clone, Debug)]
28pub struct HttpHealthConfig {
29    pub bind_address: SocketAddr,
30}
31
32impl Default for HttpHealthConfig {
33    fn default() -> Self {
34        Self {
35            bind_address: "[::]:8081".parse().expect("valid socket address"),
36        }
37    }
38}
39
40impl HttpHealthConfig {
41    pub fn new(bind_address: impl Into<SocketAddr>) -> Self {
42        Self {
43            bind_address: bind_address.into(),
44        }
45    }
46}
47
48#[derive(Clone)]
49struct HttpRequestState {
50    health_monitor: Arc<Option<HealthMonitor>>,
51    rpc_url: Arc<Option<String>>,
52    rpc_client: Client,
53    program_account_reader: Arc<Option<ProgramAccountReaderFn>>,
54    auth_plugin: Arc<Option<Arc<dyn WebSocketAuthPlugin>>>,
55    limit_state: Arc<HttpLimitState>,
56}
57
58/// HTTP server that exposes health endpoints
59pub struct HttpHealthServer {
60    bind_addr: SocketAddr,
61    health_monitor: Option<HealthMonitor>,
62    program_account_reader: Option<ProgramAccountReaderFn>,
63    auth_plugin: Option<Arc<dyn WebSocketAuthPlugin>>,
64}
65
66impl HttpHealthServer {
67    pub fn new(bind_addr: SocketAddr) -> Self {
68        Self {
69            bind_addr,
70            health_monitor: None,
71            program_account_reader: None,
72            auth_plugin: None,
73        }
74    }
75
76    pub fn with_health_monitor(mut self, monitor: HealthMonitor) -> Self {
77        self.health_monitor = Some(monitor);
78        self
79    }
80
81    pub fn with_program_account_reader(mut self, reader: ProgramAccountReaderFn) -> Self {
82        self.program_account_reader = Some(reader);
83        self
84    }
85
86    pub fn with_auth_plugin(mut self, plugin: Arc<dyn WebSocketAuthPlugin>) -> Self {
87        self.auth_plugin = Some(plugin);
88        self
89    }
90
91    pub async fn start(self) -> Result<()> {
92        info!("Starting HTTP health server on {}", self.bind_addr);
93
94        let listener = TcpListener::bind(&self.bind_addr).await?;
95        info!("HTTP health server listening on {}", self.bind_addr);
96
97        let request_state = HttpRequestState {
98            health_monitor: Arc::new(self.health_monitor),
99            rpc_url: Arc::new(resolve_rpc_url()),
100            rpc_client: Client::builder().build()?,
101            program_account_reader: Arc::new(self.program_account_reader),
102            auth_plugin: Arc::new(self.auth_plugin),
103            limit_state: Arc::new(HttpLimitState::default()),
104        };
105
106        loop {
107            match listener.accept().await {
108                Ok((stream, remote_addr)) => {
109                    let io = TokioIo::new(stream);
110                    let request_state = request_state.clone();
111
112                    tokio::spawn(async move {
113                        let service = service_fn(move |req| {
114                            let request_state = request_state.clone();
115                            async move { handle_request(remote_addr, req, request_state).await }
116                        });
117
118                        if let Err(e) = http1::Builder::new().serve_connection(io, service).await {
119                            error!("HTTP connection error: {}", e);
120                        }
121                    });
122                }
123                Err(e) => {
124                    error!("Failed to accept HTTP connection: {}", e);
125                }
126            }
127        }
128    }
129}
130
131async fn handle_request(
132    remote_addr: SocketAddr,
133    req: Request<hyper::body::Incoming>,
134    state: HttpRequestState,
135) -> Result<Response<Full<Bytes>>, Infallible> {
136    let HttpRequestState {
137        health_monitor,
138        rpc_url,
139        rpc_client,
140        program_account_reader,
141        auth_plugin,
142        limit_state,
143    } = state;
144    let path = req.uri().path().to_string();
145
146    match path.as_str() {
147        "/health" | "/healthz" => {
148            // Basic health check - server is running
149            Ok(Response::builder()
150                .status(StatusCode::OK)
151                .header("Content-Type", "text/plain")
152                .body(Full::new(Bytes::from("OK")))
153                .unwrap())
154        }
155        "/ready" | "/readiness" => {
156            // Readiness check - check if stream is healthy
157            if let Some(monitor) = health_monitor.as_ref() {
158                if monitor.is_healthy().await {
159                    Ok(Response::builder()
160                        .status(StatusCode::OK)
161                        .header("Content-Type", "text/plain")
162                        .body(Full::new(Bytes::from("READY")))
163                        .unwrap())
164                } else {
165                    Ok(Response::builder()
166                        .status(StatusCode::SERVICE_UNAVAILABLE)
167                        .header("Content-Type", "text/plain")
168                        .body(Full::new(Bytes::from("NOT READY")))
169                        .unwrap())
170                }
171            } else {
172                // No health monitor configured, assume ready
173                Ok(Response::builder()
174                    .status(StatusCode::OK)
175                    .header("Content-Type", "text/plain")
176                    .body(Full::new(Bytes::from("READY")))
177                    .unwrap())
178            }
179        }
180        "/status" => {
181            // Detailed status endpoint
182            if let Some(monitor) = health_monitor.as_ref() {
183                let status = monitor.status().await;
184                let error_count = monitor.error_count().await;
185                let is_healthy = monitor.is_healthy().await;
186
187                let status_json = serde_json::json!({
188                    "healthy": is_healthy,
189                    "status": format!("{:?}", status),
190                    "error_count": error_count
191                });
192
193                let status_code = if is_healthy {
194                    StatusCode::OK
195                } else {
196                    StatusCode::SERVICE_UNAVAILABLE
197                };
198
199                Ok(Response::builder()
200                    .status(status_code)
201                    .header("Content-Type", "application/json")
202                    .body(Full::new(Bytes::from(status_json.to_string())))
203                    .unwrap())
204            } else {
205                let status_json = serde_json::json!({
206                    "healthy": true,
207                    "status": "no_monitor",
208                    "error_count": 0
209                });
210
211                Ok(Response::builder()
212                    .status(StatusCode::OK)
213                    .header("Content-Type", "application/json")
214                    .body(Full::new(Bytes::from(status_json.to_string())))
215                    .unwrap())
216            }
217        }
218        _ if path.starts_with("/chain/") => {
219            let auth_context = match authorize_http_request(
220                remote_addr,
221                &req,
222                auth_plugin.as_ref().as_ref(),
223                &limit_state,
224            )
225            .await
226            {
227                Ok(context) => context,
228                Err(response) => return Ok(response),
229            };
230            Ok(handle_chain_request(req, path.as_str(), rpc_url, rpc_client, auth_context).await)
231        }
232        _ if path.starts_with("/programs/") => {
233            let auth_context = match authorize_http_request(
234                remote_addr,
235                &req,
236                auth_plugin.as_ref().as_ref(),
237                &limit_state,
238            )
239            .await
240            {
241                Ok(context) => context,
242                Err(response) => return Ok(response),
243            };
244            Ok(handle_program_account_request(
245                req,
246                path.as_str(),
247                rpc_url,
248                rpc_client,
249                program_account_reader,
250                auth_context,
251            )
252            .await)
253        }
254        _ => Ok(Response::builder()
255            .status(StatusCode::NOT_FOUND)
256            .header("Content-Type", "text/plain")
257            .body(Full::new(Bytes::from("Not Found")))
258            .unwrap()),
259    }
260}
261
262#[derive(Debug, Deserialize)]
263struct AddressesBody {
264    addresses: Vec<String>,
265}
266
267#[derive(Debug, Deserialize)]
268struct BalanceBody {
269    owner: String,
270    mint: String,
271    #[serde(default, rename = "tokenProgram")]
272    token_program: Option<String>,
273}
274
275#[derive(Default)]
276struct HttpLimitState {
277    per_subject_per_minute: DashMap<String, (u64, u32)>,
278}
279
280fn resolve_rpc_url() -> Option<String> {
281    ["ARETE_READ_RPC_URL", "SOLANA_RPC_URL", "RPC_URL"]
282        .iter()
283        .find_map(|key| env::var(key).ok())
284        .filter(|value| !value.is_empty())
285}
286
287fn json_response(status: StatusCode, value: Value) -> Response<Full<Bytes>> {
288    Response::builder()
289        .status(status)
290        .header("Content-Type", "application/json")
291        .body(Full::new(Bytes::from(value.to_string())))
292        .unwrap()
293}
294
295fn error_response(status: StatusCode, message: impl Into<String>) -> Response<Full<Bytes>> {
296    json_response(status, json!({ "error": message.into() }))
297}
298
299fn auth_deny_response(deny: &AuthDeny) -> Response<Full<Bytes>> {
300    let mut builder = Response::builder()
301        .status(deny.http_status)
302        .header("Content-Type", "application/json")
303        .header("X-Error-Code", deny.code.as_str());
304
305    if let Some(reset_at) = deny.reset_at {
306        if let Ok(duration) = reset_at.duration_since(SystemTime::now()) {
307            builder = builder.header("Retry-After", duration.as_secs().to_string());
308        }
309    }
310
311    builder
312        .body(Full::new(Bytes::from(
313            json!({
314                "error": deny.reason,
315                "message": deny.reason,
316                "code": deny.code.as_str(),
317                "retryable": deny.code.should_retry(),
318                "fatal": !deny.code.should_retry() && !deny.code.should_refresh_token(),
319            })
320            .to_string(),
321        )))
322        .unwrap()
323}
324
325async fn authorize_http_request(
326    remote_addr: SocketAddr,
327    req: &Request<hyper::body::Incoming>,
328    auth_plugin: Option<&Arc<dyn WebSocketAuthPlugin>>,
329    limit_state: &HttpLimitState,
330) -> std::result::Result<Option<crate::websocket::auth::AuthContext>, Response<Full<Bytes>>> {
331    let Some(plugin) = auth_plugin else {
332        return Ok(None);
333    };
334
335    let mut auth_request = ConnectionAuthRequest::from_http_request(remote_addr, req);
336    // HTTP reads are bearer-only. Do not allow query-param session tokens here.
337    auth_request.query = None;
338    let decision = plugin.authorize(&auth_request).await;
339    let context = match decision {
340        AuthDecision::Allow(context) => context,
341        AuthDecision::Deny(deny) => return Err(auth_deny_response(&deny)),
342    };
343
344    enforce_http_limits(&context, limit_state).map_err(|deny| auth_deny_response(&deny))?;
345    Ok(Some(context))
346}
347
348fn enforce_http_limits(
349    context: &crate::websocket::auth::AuthContext,
350    limit_state: &HttpLimitState,
351) -> std::result::Result<(), Box<AuthDeny>> {
352    let Some(limit) = context.limits.max_http_requests_per_minute else {
353        return Ok(());
354    };
355
356    let now_bucket = SystemTime::now()
357        .duration_since(UNIX_EPOCH)
358        .unwrap_or(Duration::from_secs(0))
359        .as_secs()
360        / 60;
361    let key = format!("{}:{}", context.subject, context.metering_key);
362    let mut entry = limit_state
363        .per_subject_per_minute
364        .entry(key)
365        .or_insert((now_bucket, 0));
366    if entry.0 != now_bucket {
367        *entry = (now_bucket, 0);
368    }
369    if entry.1 >= limit {
370        return Err(Box::new(AuthDeny::rate_limited(
371            Duration::from_secs(60),
372            "http reads",
373        )));
374    }
375    entry.1 += 1;
376    Ok(())
377}
378
379async fn read_json_body<T: for<'de> Deserialize<'de>>(
380    req: Request<hyper::body::Incoming>,
381) -> std::result::Result<T, Response<Full<Bytes>>> {
382    let collected = req
383        .into_body()
384        .collect()
385        .await
386        .map_err(|err| error_response(StatusCode::BAD_REQUEST, err.to_string()))?;
387    serde_json::from_slice::<T>(&collected.to_bytes())
388        .map_err(|err| error_response(StatusCode::BAD_REQUEST, err.to_string()))
389}
390
391async fn handle_chain_request(
392    req: Request<hyper::body::Incoming>,
393    path: &str,
394    rpc_url: Arc<Option<String>>,
395    rpc_client: Client,
396    _auth_context: Option<crate::websocket::auth::AuthContext>,
397) -> Response<Full<Bytes>> {
398    let Some(rpc_url) = rpc_url.as_ref() else {
399        return error_response(
400            StatusCode::SERVICE_UNAVAILABLE,
401            "No RPC URL configured for chain reads",
402        );
403    };
404
405    match (req.method().as_str(), path) {
406        ("GET", path) if path.starts_with("/chain/exists/") => {
407            let address = path.trim_start_matches("/chain/exists/");
408            match rpc_get_account_info(&rpc_client, rpc_url, address).await {
409                Ok(value) => json_response(StatusCode::OK, json!({ "exists": !value.is_null() })),
410                Err(err) => error_response(StatusCode::BAD_GATEWAY, err.to_string()),
411            }
412        }
413        ("GET", path) if path.starts_with("/chain/lamports/") => {
414            let address = path.trim_start_matches("/chain/lamports/");
415            match rpc_call(
416                &rpc_client,
417                rpc_url,
418                "getBalance",
419                json!([address, { "commitment": "confirmed" }]),
420            )
421            .await
422            {
423                Ok(value) => json_response(
424                    StatusCode::OK,
425                    json!({ "lamports": value.pointer("/value").and_then(Value::as_u64).unwrap_or(0) }),
426                ),
427                Err(err) => error_response(StatusCode::BAD_GATEWAY, err.to_string()),
428            }
429        }
430        ("GET", path) if path.starts_with("/chain/rent-exemption/") => {
431            let raw_space = path.trim_start_matches("/chain/rent-exemption/");
432            let Ok(space) = raw_space.parse::<u64>() else {
433                return error_response(
434                    StatusCode::BAD_REQUEST,
435                    "rent-exemption space must be an integer",
436                );
437            };
438            match rpc_call(
439                &rpc_client,
440                rpc_url,
441                "getMinimumBalanceForRentExemption",
442                json!([space, { "commitment": "confirmed" }]),
443            )
444            .await
445            {
446                Ok(value) => json_response(
447                    StatusCode::OK,
448                    json!({ "lamports": value.as_u64().unwrap_or(0) }),
449                ),
450                Err(err) => error_response(StatusCode::BAD_GATEWAY, err.to_string()),
451            }
452        }
453        ("GET", "/chain/clock") => {
454            let slot = rpc_call(
455                &rpc_client,
456                rpc_url,
457                "getSlot",
458                json!([{ "commitment": "confirmed" }]),
459            )
460            .await;
461            let epoch_info = rpc_call(
462                &rpc_client,
463                rpc_url,
464                "getEpochInfo",
465                json!([{ "commitment": "confirmed" }]),
466            )
467            .await;
468            match (slot, epoch_info) {
469                (Ok(slot_value), Ok(epoch_value)) => {
470                    let slot_num = slot_value.as_u64().unwrap_or(0);
471                    let unix_timestamp =
472                        rpc_call(&rpc_client, rpc_url, "getBlockTime", json!([slot_num]))
473                            .await
474                            .ok()
475                            .and_then(|value| value.as_i64())
476                            .unwrap_or_default();
477                    json_response(
478                        StatusCode::OK,
479                        json!({
480                            "slot": slot_num,
481                            "epoch": epoch_value.get("epoch").and_then(Value::as_u64),
482                            "leaderScheduleEpoch": epoch_value.get("leaderScheduleSlotOffset").and_then(Value::as_u64),
483                            "unixTimestamp": unix_timestamp,
484                        }),
485                    )
486                }
487                (Err(err), _) | (_, Err(err)) => {
488                    error_response(StatusCode::BAD_GATEWAY, err.to_string())
489                }
490            }
491        }
492        ("GET", path) if path.starts_with("/chain/accounts/") => {
493            let address = path.trim_start_matches("/chain/accounts/");
494            match rpc_get_account_info(&rpc_client, rpc_url, address).await {
495                Ok(value) if value.is_null() => json_response(StatusCode::OK, Value::Null),
496                Ok(value) => json_response(StatusCode::OK, raw_account_json(address, &value)),
497                Err(err) => error_response(StatusCode::BAD_GATEWAY, err.to_string()),
498            }
499        }
500        ("GET", path) if path.starts_with("/chain/mints/") => {
501            let address = path.trim_start_matches("/chain/mints/");
502            match rpc_get_parsed_account_info(&rpc_client, rpc_url, address).await {
503                Ok(Some(value)) => json_response(StatusCode::OK, mint_info_json(address, &value)),
504                Ok(None) => json_response(StatusCode::OK, Value::Null),
505                Err(err) => error_response(StatusCode::BAD_GATEWAY, err.to_string()),
506            }
507        }
508        ("GET", path) if path.starts_with("/chain/token-accounts/") => {
509            let address = path.trim_start_matches("/chain/token-accounts/");
510            match rpc_get_parsed_account_info(&rpc_client, rpc_url, address).await {
511                Ok(Some(value)) => {
512                    json_response(StatusCode::OK, token_account_json(address, &value))
513                }
514                Ok(None) => json_response(StatusCode::OK, Value::Null),
515                Err(err) => error_response(StatusCode::BAD_GATEWAY, err.to_string()),
516            }
517        }
518        ("POST", "/chain/balances") => match read_json_body::<BalanceBody>(req).await {
519            Ok(body) => match rpc_get_token_balance(
520                &rpc_client,
521                rpc_url,
522                &body.owner,
523                &body.mint,
524                body.token_program.as_deref(),
525            )
526            .await
527            {
528                Ok(balance) => json_response(StatusCode::OK, balance),
529                Err(err) => error_response(StatusCode::BAD_GATEWAY, err.to_string()),
530            },
531            Err(response) => response,
532        },
533        _ => error_response(StatusCode::NOT_FOUND, "Not Found"),
534    }
535}
536
537async fn handle_program_account_request(
538    req: Request<hyper::body::Incoming>,
539    path: &str,
540    rpc_url: Arc<Option<String>>,
541    rpc_client: Client,
542    program_account_reader: Arc<Option<ProgramAccountReaderFn>>,
543    auth_context: Option<crate::websocket::auth::AuthContext>,
544) -> Response<Full<Bytes>> {
545    let Some(rpc_url) = rpc_url.as_ref() else {
546        return error_response(
547            StatusCode::SERVICE_UNAVAILABLE,
548            "No RPC URL configured for program account reads",
549        );
550    };
551    let Some(reader) = program_account_reader.as_ref() else {
552        return error_response(
553            StatusCode::NOT_IMPLEMENTED,
554            "Program account reader is not configured",
555        );
556    };
557
558    let segments: Vec<&str> = path.trim_start_matches('/').split('/').collect();
559    if segments.len() < 4
560        || segments.first() != Some(&"programs")
561        || segments.get(2) != Some(&"accounts")
562    {
563        return error_response(StatusCode::NOT_FOUND, "Not Found");
564    }
565    let program = segments[1];
566    let account = segments[3];
567
568    match (req.method().as_str(), segments.len()) {
569        ("GET", 5) => {
570            let address = segments[4];
571            match rpc_get_account_info(&rpc_client, rpc_url, address).await {
572                Ok(value) if value.is_null() => json_response(StatusCode::OK, Value::Null),
573                Ok(value) => match decode_account_bytes(&value)
574                    .and_then(|data| reader(program, account, &data).ok())
575                {
576                    Some(parsed) => json_response(StatusCode::OK, parsed),
577                    None => {
578                        error_response(StatusCode::BAD_REQUEST, "Unable to parse requested account")
579                    }
580                },
581                Err(err) => error_response(StatusCode::BAD_GATEWAY, err.to_string()),
582            }
583        }
584        ("GET", 6) if segments[5] == "exists" => {
585            let address = segments[4];
586            match rpc_get_account_info(&rpc_client, rpc_url, address).await {
587                Ok(value) => json_response(StatusCode::OK, json!({ "exists": !value.is_null() })),
588                Err(err) => error_response(StatusCode::BAD_GATEWAY, err.to_string()),
589            }
590        }
591        ("POST", 4) => match read_json_body::<AddressesBody>(req).await {
592            Ok(body) => {
593                if let Some(limit) = auth_context
594                    .as_ref()
595                    .and_then(|ctx| ctx.limits.max_http_batch_addresses)
596                {
597                    if body.addresses.len() > limit as usize {
598                        return auth_deny_response(&AuthDeny::rate_limited(
599                            Duration::from_secs(60),
600                            &format!(
601                                "http batch reads ({} addresses > {})",
602                                body.addresses.len(),
603                                limit
604                            ),
605                        ));
606                    }
607                }
608                match rpc_get_multiple_accounts(&rpc_client, rpc_url, &body.addresses).await {
609                    Ok(values) => {
610                        let parsed: Vec<Value> = values
611                            .iter()
612                            .map(|value| {
613                                if value.is_null() {
614                                    Value::Null
615                                } else if let Some(data) = decode_account_bytes(value) {
616                                    reader(program, account, &data).unwrap_or(Value::Null)
617                                } else {
618                                    Value::Null
619                                }
620                            })
621                            .collect();
622                        json_response(StatusCode::OK, Value::Array(parsed))
623                    }
624                    Err(err) => error_response(StatusCode::BAD_GATEWAY, err.to_string()),
625                }
626            }
627            Err(response) => response,
628        },
629        _ => error_response(StatusCode::NOT_FOUND, "Not Found"),
630    }
631}
632
633async fn rpc_call(
634    client: &Client,
635    rpc_url: &str,
636    method: &str,
637    params: Value,
638) -> anyhow::Result<Value> {
639    let response = client
640        .post(rpc_url)
641        .json(&json!({
642            "jsonrpc": "2.0",
643            "id": "arete-read",
644            "method": method,
645            "params": params,
646        }))
647        .send()
648        .await?
649        .error_for_status()?;
650    let value = response.json::<Value>().await?;
651    if let Some(error) = value.get("error") {
652        return Err(anyhow::anyhow!(error.to_string()));
653    }
654    Ok(value.get("result").cloned().unwrap_or(Value::Null))
655}
656
657async fn rpc_get_account_info(
658    client: &Client,
659    rpc_url: &str,
660    address: &str,
661) -> anyhow::Result<Value> {
662    let result = rpc_call(
663        client,
664        rpc_url,
665        "getAccountInfo",
666        json!([address, { "encoding": "base64", "commitment": "confirmed" }]),
667    )
668    .await?;
669    Ok(result.get("value").cloned().unwrap_or(Value::Null))
670}
671
672async fn rpc_get_multiple_accounts(
673    client: &Client,
674    rpc_url: &str,
675    addresses: &[String],
676) -> anyhow::Result<Vec<Value>> {
677    let result = rpc_call(
678        client,
679        rpc_url,
680        "getMultipleAccounts",
681        json!([addresses, { "encoding": "base64", "commitment": "confirmed" }]),
682    )
683    .await?;
684    Ok(result
685        .get("value")
686        .and_then(Value::as_array)
687        .cloned()
688        .unwrap_or_default())
689}
690
691async fn rpc_get_parsed_account_info(
692    client: &Client,
693    rpc_url: &str,
694    address: &str,
695) -> anyhow::Result<Option<Value>> {
696    let result = rpc_call(
697        client,
698        rpc_url,
699        "getAccountInfo",
700        json!([address, { "encoding": "jsonParsed", "commitment": "confirmed" }]),
701    )
702    .await?;
703    Ok(result
704        .get("value")
705        .cloned()
706        .filter(|value| !value.is_null()))
707}
708
709async fn rpc_get_token_balance(
710    client: &Client,
711    rpc_url: &str,
712    owner: &str,
713    mint: &str,
714    token_program: Option<&str>,
715) -> anyhow::Result<Value> {
716    let filter = token_program
717        .map(|program_id| json!({ "programId": program_id }))
718        .unwrap_or_else(|| json!({ "mint": mint }));
719    let result = rpc_call(
720        client,
721        rpc_url,
722        "getTokenAccountsByOwner",
723        json!([
724            owner,
725            filter,
726            { "encoding": "jsonParsed", "commitment": "confirmed" }
727        ]),
728    )
729    .await?;
730
731    let account = result
732        .get("value")
733        .and_then(Value::as_array)
734        .and_then(|items| {
735            items.iter().find(|item| {
736                item.pointer("/account/data/parsed/info/mint")
737                    .and_then(Value::as_str)
738                    == Some(mint)
739            })
740        })
741        .cloned();
742
743    if let Some(account) = account {
744        let pubkey = account.get("pubkey").and_then(Value::as_str);
745        let info = account.pointer("/account/data/parsed/info");
746        return Ok(json!({
747            "exists": true,
748            "address": pubkey,
749            "owner": owner,
750            "mint": mint,
751            "tokenProgram": token_program,
752            "amount": info.and_then(|value| value.pointer("/tokenAmount/amount")).and_then(Value::as_str).unwrap_or("0"),
753            "decimals": info.and_then(|value| value.pointer("/tokenAmount/decimals")).and_then(Value::as_u64),
754            "uiAmountString": info.and_then(|value| value.pointer("/tokenAmount/uiAmountString")).and_then(Value::as_str),
755        }));
756    }
757
758    Ok(json!({
759        "exists": false,
760        "address": Value::Null,
761        "owner": owner,
762        "mint": mint,
763        "tokenProgram": token_program,
764        "amount": "0",
765        "decimals": Value::Null,
766        "uiAmountString": Value::Null,
767    }))
768}
769
770fn decode_account_bytes(value: &Value) -> Option<Vec<u8>> {
771    let data = value.get("data")?.as_array()?;
772    let encoded = data.first()?.as_str()?;
773    BASE64_STANDARD.decode(encoded).ok()
774}
775
776fn raw_account_json(address: &str, value: &Value) -> Value {
777    json!({
778        "address": address,
779        "ownerProgram": value.get("owner").and_then(Value::as_str).unwrap_or_default(),
780        "lamports": value.get("lamports").and_then(Value::as_u64).unwrap_or(0),
781        "executable": value.get("executable").and_then(Value::as_bool).unwrap_or(false),
782        "data": value
783            .pointer("/data/0")
784            .and_then(Value::as_str)
785            .unwrap_or_default(),
786    })
787}
788
789fn mint_info_json(address: &str, value: &Value) -> Value {
790    let owner_program = value
791        .get("owner")
792        .and_then(Value::as_str)
793        .unwrap_or_default();
794    let info = value.pointer("/data/parsed/info");
795    json!({
796        "address": address,
797        "ownerProgram": owner_program,
798        "decimals": info.and_then(|v| v.get("decimals")).and_then(Value::as_u64),
799        "supply": info.and_then(|v| v.get("supply")).and_then(Value::as_str),
800        "mintAuthority": info.and_then(|v| v.get("mintAuthority")).and_then(Value::as_str),
801        "freezeAuthority": info.and_then(|v| v.get("freezeAuthority")).and_then(Value::as_str),
802    })
803}
804
805fn token_account_json(address: &str, value: &Value) -> Value {
806    let owner_program = value
807        .get("owner")
808        .and_then(Value::as_str)
809        .unwrap_or_default();
810    let info = value.pointer("/data/parsed/info");
811    json!({
812        "address": address,
813        "ownerProgram": owner_program,
814        "mint": info.and_then(|v| v.get("mint")).and_then(Value::as_str),
815        "owner": info.and_then(|v| v.get("owner")).and_then(Value::as_str),
816        "amount": info.and_then(|v| v.pointer("/tokenAmount/amount")).and_then(Value::as_str),
817        "uiAmountString": info.and_then(|v| v.pointer("/tokenAmount/uiAmountString")).and_then(Value::as_str),
818    })
819}
arete_server/http_health.rs

arete_server/
http_health.rs
