arete_server/

http_health.rs

use crate::http::fork;
use crate::http::transactions::{self, TransactionState};
use crate::{
    config::{HttpLimitsConfig, RuntimePlan, TransactionConfig},
    health::HealthMonitor,
    ProgramRuntimeCatalog, ProgramRuntimeDefinition, SolanaGatewayUsageObservation,
    SolanaGatewayUsageObserver, SolanaGatewayUsageSurface,
};
use anyhow::Result;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use base64::Engine as _;
use dashmap::DashMap;
use http_body_util::BodyExt;
use http_body_util::Full;
use hyper::body::Bytes;
use hyper::header::{
    HeaderValue, ACCESS_CONTROL_ALLOW_HEADERS, ACCESS_CONTROL_ALLOW_METHODS,
    ACCESS_CONTROL_ALLOW_ORIGIN, ACCESS_CONTROL_EXPOSE_HEADERS, ACCESS_CONTROL_MAX_AGE,
};
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::{Method, Request, Response, StatusCode};
use hyper_util::rt::{TokioIo, TokioTimer};
use reqwest::Client;
use serde::Deserialize;
use serde_json::{json, Value};
use std::convert::Infallible;
use std::env;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::net::TcpListener;
use tokio::task::JoinSet;
use tokio_util::sync::CancellationToken;
use tracing::{error, info, warn};

use crate::websocket::auth::{AuthDecision, AuthDeny, ConnectionAuthRequest, WebSocketAuthPlugin};
use arete_auth::{SCOPE_FORK_READ, SCOPE_READ};

use crate::token_discovery::{OwnerTokenAccountsProvider, TokenDiscoveryError};
use arete_solana_contracts::{Contextual, OwnerTokenAccountsRequest, ReadContext, ReadOptions};

const HTTP_CONNECTION_DRAIN_TIMEOUT: Duration = Duration::from_secs(5);

/// Configuration for the HTTP health server
#[derive(Clone, Debug)]
pub struct HttpHealthConfig {
    pub bind_address: SocketAddr,
}

impl Default for HttpHealthConfig {
    fn default() -> Self {
        Self {
            bind_address: "[::]:8081".parse().expect("valid socket address"),
        }
    }
}

impl HttpHealthConfig {
    pub fn new(bind_address: impl Into<SocketAddr>) -> Self {
        Self {
            bind_address: bind_address.into(),
        }
    }
}

#[derive(Clone)]
struct HttpRequestState {
    health_monitor: Arc<Option<HealthMonitor>>,
    snapshot_runtime: Option<crate::snapshot::SnapshotRuntime>,
    runtime_plan: RuntimePlan,
    rpc_url: Arc<Option<String>>,
    rpc_client: ReadRpcClient,
    body_limits: BodyLimits,
    program_runtime_catalog: Arc<ProgramRuntimeCatalog>,
    auth_plugin: Arc<Option<Arc<dyn WebSocketAuthPlugin>>>,
    limit_state: Arc<HttpLimitState>,
    transaction_state: Arc<Option<TransactionState>>,
    token_accounts_provider: Option<Arc<dyn OwnerTokenAccountsProvider>>,
    solana_gateway_usage_observer: Option<Arc<dyn SolanaGatewayUsageObserver>>,
    solana_gateway_target_id: Arc<Option<String>>,
    program_read_binding_target_id: Arc<Option<String>>,
}

/// HTTP server that exposes health endpoints
pub struct HttpHealthServer {
    /// Cancelled to stop the accept loop; see [`HttpHealthServer::with_shutdown`].
    shutdown: Option<CancellationToken>,
    bind_addr: SocketAddr,
    health_monitor: Option<HealthMonitor>,
    snapshot_runtime: Option<crate::snapshot::SnapshotRuntime>,
    runtime_plan: RuntimePlan,
    program_runtime_catalog: ProgramRuntimeCatalog,
    auth_plugin: Option<Arc<dyn WebSocketAuthPlugin>>,
    transaction_config: Option<TransactionConfig>,
    token_accounts_provider: Option<Arc<dyn OwnerTokenAccountsProvider>>,
    solana_gateway_usage_observer: Option<Arc<dyn SolanaGatewayUsageObserver>>,
    solana_gateway_target_id: Option<String>,
    program_read_binding_target_id: Option<String>,
    /// Read RPC used instead of the process environment; set only by tests.
    read_rpc_url: Option<String>,
    limits: HttpLimitsConfig,
    #[cfg(feature = "otel")]
    metrics: Option<Arc<crate::metrics::Metrics>>,
}

impl HttpHealthServer {
    pub fn new(bind_addr: SocketAddr) -> Self {
        Self {
            bind_addr,
            shutdown: None,
            health_monitor: None,
            snapshot_runtime: None,
            runtime_plan: RuntimePlan::http(),
            program_runtime_catalog: ProgramRuntimeCatalog::default(),
            auth_plugin: None,
            transaction_config: None,
            token_accounts_provider: None,
            solana_gateway_usage_observer: None,
            solana_gateway_target_id: None,
            program_read_binding_target_id: None,
            read_rpc_url: None,
            limits: HttpLimitsConfig::default(),
            #[cfg(feature = "otel")]
            metrics: None,
        }
    }

    #[cfg(test)]
    pub(crate) fn with_read_rpc_url(mut self, rpc_url: impl Into<String>) -> Self {
        self.read_rpc_url = Some(rpc_url.into());
        self
    }

    /// Body, header and read RPC limits; see [`HttpLimitsConfig`].
    pub fn with_limits(mut self, limits: HttpLimitsConfig) -> Self {
        self.limits = limits;
        self
    }

    pub fn with_health_monitor(mut self, monitor: HealthMonitor) -> Self {
        self.health_monitor = Some(monitor);
        self
    }

    pub fn with_snapshot_runtime(
        mut self,
        snapshot_runtime: crate::snapshot::SnapshotRuntime,
    ) -> Self {
        self.snapshot_runtime = Some(snapshot_runtime);
        self
    }

    pub fn with_runtime_plan(mut self, runtime_plan: RuntimePlan) -> Self {
        self.runtime_plan = runtime_plan;
        self
    }

    pub fn with_program_runtime_catalog(mut self, catalog: ProgramRuntimeCatalog) -> Self {
        self.program_runtime_catalog = catalog;
        self
    }

    pub fn with_auth_plugin(mut self, plugin: Arc<dyn WebSocketAuthPlugin>) -> Self {
        self.auth_plugin = Some(plugin);
        self
    }

    pub fn with_transaction_config(mut self, config: TransactionConfig) -> Self {
        self.runtime_plan.transactions = config.enabled;
        self.transaction_config = Some(config);
        self
    }

    pub fn with_owner_token_accounts_provider(
        mut self,
        provider: Arc<dyn OwnerTokenAccountsProvider>,
    ) -> Self {
        self.token_accounts_provider = Some(provider);
        self
    }

    pub fn with_solana_gateway_usage_observer(
        mut self,
        observer: Arc<dyn SolanaGatewayUsageObserver>,
    ) -> Self {
        self.solana_gateway_usage_observer = Some(observer);
        self
    }

    pub fn with_solana_gateway_target(mut self, target_id: impl Into<String>) -> Self {
        self.solana_gateway_target_id = Some(target_id.into());
        self
    }

    pub fn with_program_read_binding_target(mut self, target_id: impl Into<String>) -> Self {
        self.program_read_binding_target_id = Some(target_id.into());
        self
    }

    #[cfg(feature = "otel")]
    pub fn with_metrics(mut self, metrics: Option<Arc<crate::metrics::Metrics>>) -> Self {
        self.metrics = metrics;
        self
    }

    /// Stop accepting and return from [`start`](Self::start) when `token` is
    /// cancelled. Without one the server accepts until the process exits.
    pub fn with_shutdown(mut self, token: CancellationToken) -> Self {
        self.shutdown = Some(token);
        self
    }

    pub async fn start(self) -> Result<()> {
        info!("Starting HTTP health server on {}", self.bind_addr);
        self.limits.validate()?;

        let listener = TcpListener::bind(&self.bind_addr).await?;
        info!("HTTP health server listening on {}", self.bind_addr);

        let transaction_state = self
            .transaction_config
            .filter(|config| config.enabled)
            .map(TransactionState::new)
            .transpose()?;
        let transaction_state = match (
            transaction_state,
            self.solana_gateway_usage_observer.as_ref(),
            self.solana_gateway_target_id.as_ref(),
        ) {
            (Some(state), Some(observer), Some(target_id)) => {
                Some(state.with_solana_gateway_usage_observer(observer.clone(), target_id.clone()))
            }
            (state, _, _) => state,
        };
        #[cfg(feature = "otel")]
        let transaction_state = transaction_state.map(|state| state.with_metrics(self.metrics));
        let request_state = HttpRequestState {
            health_monitor: Arc::new(self.health_monitor),
            snapshot_runtime: self.snapshot_runtime,
            runtime_plan: self.runtime_plan,
            rpc_url: Arc::new(self.read_rpc_url.or_else(resolve_rpc_url)),
            rpc_client: ReadRpcClient {
                client: Client::builder()
                    .connect_timeout(self.limits.upstream_connect_timeout)
                    .timeout(self.limits.upstream_timeout)
                    .build()?,
                max_response_bytes: self.limits.max_upstream_response_bytes,
            },
            body_limits: BodyLimits {
                max_bytes: self.limits.max_body_bytes,
                read_timeout: self.limits.body_read_timeout,
            },
            program_runtime_catalog: Arc::new(self.program_runtime_catalog),
            auth_plugin: Arc::new(self.auth_plugin),
            limit_state: Arc::new(HttpLimitState::default()),
            transaction_state: Arc::new(transaction_state),
            token_accounts_provider: self.token_accounts_provider,
            solana_gateway_usage_observer: self.solana_gateway_usage_observer,
            solana_gateway_target_id: Arc::new(self.solana_gateway_target_id),
            program_read_binding_target_id: Arc::new(self.program_read_binding_target_id),
        };

        let shutdown = self.shutdown.unwrap_or_default();
        let header_read_timeout = self.limits.header_read_timeout;
        let bind_addr = self.bind_addr;
        let mut connections = accept_connections(&listener, &shutdown, |stream, remote_addr| {
            let io = TokioIo::new(stream);
            let request_state = request_state.clone();
            let connection_shutdown = shutdown.clone();
            async move {
                let service = service_fn(move |req| {
                    let request_state = request_state.clone();
                    async move { handle_request(remote_addr, req, request_state).await }
                });
                // A client that never finishes its headers is dropped at the deadline.
                let connection = http1::Builder::new()
                    .timer(TokioTimer::new())
                    .header_read_timeout(header_read_timeout)
                    .serve_connection(io, service);
                tokio::pin!(connection);
                let result = tokio::select! {
                    result = &mut connection => result,
                    _ = connection_shutdown.cancelled() => {
                        connection.as_mut().graceful_shutdown();
                        connection.await
                    }
                };
                if let Err(e) = result {
                    error!("HTTP connection error: {}", e);
                }
            }
        })
        .await;
        info!("HTTP health server on {} stopping", bind_addr);

        if tokio::time::timeout(HTTP_CONNECTION_DRAIN_TIMEOUT, async {
            while connections.join_next().await.is_some() {}
        })
        .await
        .is_err()
        {
            warn!(
                "HTTP connections did not drain within {:?}; aborting them",
                HTTP_CONNECTION_DRAIN_TIMEOUT
            );
            connections.abort_all();
            while connections.join_next().await.is_some() {}
        }

        Ok(())
    }
}

/// Accept from `listener` until `shutdown`, serving each connection on its own
/// task, and return the tasks still running so the caller can drain them.
///
/// Finished connection tasks are reaped as they complete. A `JoinSet` keeps
/// every finished task until it is joined, so a server that only joined at
/// shutdown held one entry for every connection it had ever served.
async fn accept_connections<F, Fut>(
    listener: &TcpListener,
    shutdown: &CancellationToken,
    mut serve: F,
) -> JoinSet<()>
where
    F: FnMut(tokio::net::TcpStream, SocketAddr) -> Fut,
    Fut: std::future::Future<Output = ()> + Send + 'static,
{
    let mut connections = JoinSet::new();
    loop {
        let accepted = tokio::select! {
            _ = shutdown.cancelled() => break,
            Some(_) = connections.join_next(), if !connections.is_empty() => continue,
            accepted = listener.accept() => accepted,
        };
        match accepted {
            Ok((stream, remote_addr)) => {
                connections.spawn(serve(stream, remote_addr));
            }
            Err(e) => {
                error!("Failed to accept HTTP connection: {}", e);
            }
        }
    }
    connections
}

async fn handle_request(
    remote_addr: SocketAddr,
    mut req: Request<hyper::body::Incoming>,
    state: HttpRequestState,
) -> Result<Response<Full<Bytes>>, Infallible> {
    // Every body reader finds its limits on the request itself.
    req.extensions_mut().insert(state.body_limits);
    if req.method() == Method::OPTIONS {
        return Ok(with_cors(
            Response::builder()
                .status(StatusCode::NO_CONTENT)
                .body(Full::new(Bytes::new()))
                .unwrap(),
        ));
    }

    let response = handle_request_inner(remote_addr, req, state).await?;
    Ok(with_cors(response))
}

fn with_cors(mut response: Response<Full<Bytes>>) -> Response<Full<Bytes>> {
    let headers = response.headers_mut();
    headers.insert(ACCESS_CONTROL_ALLOW_ORIGIN, HeaderValue::from_static("*"));
    headers.insert(
        ACCESS_CONTROL_ALLOW_METHODS,
        HeaderValue::from_static("GET, POST, OPTIONS"),
    );
    headers.insert(
        ACCESS_CONTROL_ALLOW_HEADERS,
        HeaderValue::from_static("Authorization, Content-Type"),
    );
    headers.insert(
        ACCESS_CONTROL_EXPOSE_HEADERS,
        HeaderValue::from_static(
            "Retry-After, X-Error-Code, X-Request-Id, X-Arete-Upstream-Attempted, X-Arete-Program-Release-Hash, X-Arete-Idl-Content-Hash, X-Arete-Account-Address, X-Arete-Account-Exists",
        ),
    );
    headers.insert(ACCESS_CONTROL_MAX_AGE, HeaderValue::from_static("86400"));
    response
}

async fn handle_request_inner(
    remote_addr: SocketAddr,
    req: Request<hyper::body::Incoming>,
    state: HttpRequestState,
) -> Result<Response<Full<Bytes>>, Infallible> {
    let HttpRequestState {
        health_monitor,
        snapshot_runtime,
        runtime_plan,
        rpc_url,
        rpc_client,
        body_limits,
        program_runtime_catalog,
        auth_plugin,
        limit_state,
        transaction_state,
        token_accounts_provider,
        solana_gateway_usage_observer,
        solana_gateway_target_id,
        program_read_binding_target_id,
    } = state;
    let path = req.uri().path().to_string();

    match path.as_str() {
        "/health" | "/healthz" if runtime_plan.health => {
            // Basic health check - server is running
            Ok(Response::builder()
                .status(StatusCode::OK)
                .header("Content-Type", "text/plain")
                .body(Full::new(Bytes::from("OK")))
                .unwrap())
        }
        "/ready" | "/readiness" if runtime_plan.health => {
            // Readiness check - stream must be healthy, and after a snapshot
            // resume the parser must have caught back up to the slot tip
            // before this pod should take traffic.
            let stream_ready = match health_monitor.as_ref() {
                Some(monitor) => monitor.is_healthy().await,
                // No health monitor configured, assume ready
                None => true,
            };
            let snapshot_ready = snapshot_runtime
                .as_ref()
                .is_none_or(crate::snapshot::SnapshotRuntime::resume_gate_ready);
            if stream_ready && snapshot_ready {
                Ok(Response::builder()
                    .status(StatusCode::OK)
                    .header("Content-Type", "text/plain")
                    .body(Full::new(Bytes::from("READY")))
                    .unwrap())
            } else {
                Ok(Response::builder()
                    .status(StatusCode::SERVICE_UNAVAILABLE)
                    .header("Content-Type", "text/plain")
                    .body(Full::new(Bytes::from("NOT READY")))
                    .unwrap())
            }
        }
        "/status" if runtime_plan.health => {
            // Detailed status endpoint
            if let Some(monitor) = health_monitor.as_ref() {
                let status = monitor.status().await;
                let error_count = monitor.error_count().await;
                let is_healthy = monitor.is_healthy().await;

                let status_json = serde_json::json!({
                    "healthy": is_healthy,
                    "status": format!("{:?}", status),
                    "error_count": error_count
                });

                let status_code = if is_healthy {
                    StatusCode::OK
                } else {
                    StatusCode::SERVICE_UNAVAILABLE
                };

                Ok(Response::builder()
                    .status(status_code)
                    .header("Content-Type", "application/json")
                    .body(Full::new(Bytes::from(status_json.to_string())))
                    .unwrap())
            } else {
                let status_json = serde_json::json!({
                    "healthy": true,
                    "status": "no_monitor",
                    "error_count": 0
                });

                Ok(Response::builder()
                    .status(StatusCode::OK)
                    .header("Content-Type", "application/json")
                    .body(Full::new(Bytes::from(status_json.to_string())))
                    .unwrap())
            }
        }
        _ if runtime_plan.transactions && path.starts_with("/transactions/") => {
            let Some(transaction_state) = transaction_state.as_ref() else {
                return Ok(error_response(StatusCode::NOT_FOUND, "Not Found"));
            };
            let client_addr = transaction_state.client_addr(remote_addr, req.headers());
            let auth_context = match authorize_http_request(
                client_addr,
                &req,
                auth_plugin.as_ref().as_ref(),
                &limit_state,
                None,
                false,
                solana_gateway_target_id.as_deref(),
            )
            .await
            {
                Ok(context) => context,
                Err(response) => return Ok(transaction_auth_error(response, path.as_str())),
            };
            Ok(transactions::handle(
                client_addr,
                req,
                auth_context,
                transaction_state.clone(),
                body_limits.read_timeout,
            )
            .await)
        }
        _ if runtime_plan.chain_reads && path.starts_with("/chain/") => {
            let auth_context = match authorize_http_request(
                remote_addr,
                &req,
                auth_plugin.as_ref().as_ref(),
                &limit_state,
                Some(SCOPE_READ),
                true,
                solana_gateway_target_id.as_deref(),
            )
            .await
            {
                Ok(context) => context,
                Err(response) => return Ok(response),
            };
            Ok(handle_chain_request(
                req,
                path.as_str(),
                ChainRequestServices {
                    rpc_url,
                    rpc_client,
                    auth_context,
                    usage_observer: solana_gateway_usage_observer.as_ref(),
                    solana_gateway_target_id: solana_gateway_target_id.as_deref(),
                    provider: token_accounts_provider.as_ref(),
                },
            )
            .await)
        }
        _ if runtime_plan.chain_reads && path == fork::FORK_RPC_PATH => {
            if req.method() != Method::POST {
                return Ok(error_response(
                    StatusCode::METHOD_NOT_ALLOWED,
                    "The fork endpoint accepts POST only",
                ));
            }
            let auth_context = match authorize_http_request(
                remote_addr,
                &req,
                auth_plugin.as_ref().as_ref(),
                &limit_state,
                Some(SCOPE_FORK_READ),
                true,
                solana_gateway_target_id.as_deref(),
            )
            .await
            {
                Ok(context) => context,
                Err(response) => return Ok(response),
            };
            let body =
                match tokio::time::timeout(body_limits.read_timeout, fork::read_body(req)).await {
                    Ok(Ok(body)) => body,
                    Ok(Err(response)) => return Ok(response),
                    Err(_) => return Ok(body_timeout_response()),
                };
            let payload = fork::parse_payload(&body);
            // A batch forwards up to one upstream call per entry, so it is charged per entry.
            // `authorize_http_request` already charged the first.
            if let Some(context) = auth_context.as_ref() {
                if let Err(deny) =
                    enforce_http_limits_weighted(context, &limit_state, payload.weight() - 1)
                {
                    return Ok(auth_deny_response(&deny));
                }
            }
            let services = fork::ForkServices {
                rpc_url: rpc_url.as_ref().as_deref(),
                client: &rpc_client,
                auth_context: auth_context.as_ref(),
                usage_observer: solana_gateway_usage_observer.as_ref(),
                target_id: solana_gateway_target_id.as_deref(),
                max_response_bytes: fork::MAX_FORK_RESPONSE_BYTES,
            };
            Ok(fork::execute(payload, &services).await)
        }
        _ if path.starts_with("/v1/releases/") => {
            if !runtime_plan.program_reads {
                return Ok(program_read_error_response(
                    ProgramReadError::ProgramReadsDisabled,
                ));
            }
            let auth_context = match authorize_http_request(
                remote_addr,
                &req,
                auth_plugin.as_ref().as_ref(),
                &limit_state,
                Some(SCOPE_READ),
                true,
                None,
            )
            .await
            {
                Ok(context) => context,
                Err(response) => return Ok(response),
            };
            Ok(handle_program_account_request(
                req,
                path.as_str(),
                rpc_url,
                rpc_client,
                program_runtime_catalog,
                auth_context,
                program_read_binding_target_id,
            )
            .await)
        }
        _ => Ok(Response::builder()
            .status(StatusCode::NOT_FOUND)
            .header("Content-Type", "text/plain")
            .body(Full::new(Bytes::from("Not Found")))
            .unwrap()),
    }
}

fn transaction_auth_error(response: Response<Full<Bytes>>, path: &str) -> Response<Full<Bytes>> {
    let status = response.status();
    let code = response
        .headers()
        .get("X-Error-Code")
        .and_then(|value| value.to_str().ok())
        .unwrap_or("authentication_failed")
        .to_string();
    let request_id = uuid::Uuid::new_v4().to_string();
    let mut value = json!({
        "code": code,
        "message": "Transaction request authentication failed",
        "retryable": status == StatusCode::TOO_MANY_REQUESTS || status == StatusCode::UNAUTHORIZED,
        "requestId": request_id,
    });
    if path == "/transactions/v1/send" {
        value["submissionState"] = json!("not_submitted");
    }
    Response::builder()
        .status(status)
        .header("Content-Type", "application/json")
        .header("X-Error-Code", code)
        .header("X-Request-Id", request_id)
        .header("X-Arete-Upstream-Attempted", "false")
        .body(Full::new(Bytes::from(value.to_string())))
        .expect("valid transaction authentication response")
}

fn capability_error(
    status: StatusCode,
    code: &str,
    message: impl Into<String>,
) -> Response<Full<Bytes>> {
    let mut response = json_response(status, json!({ "code": code, "error": message.into() }));
    response.headers_mut().insert(
        "X-Error-Code",
        HeaderValue::from_str(code).expect("static error code"),
    );
    response
}

async fn handle_owner_token_accounts(
    req: Request<hyper::body::Incoming>,
    provider: Option<&Arc<dyn OwnerTokenAccountsProvider>>,
    auth: Option<&crate::websocket::auth::AuthContext>,
) -> Response<Full<Bytes>> {
    let request = match read_json_body::<OwnerTokenAccountsRequest>(req).await {
        Ok(request) => request,
        Err(response) if response.status() != StatusCode::BAD_REQUEST => return response,
        Err(_) => {
            return capability_error(
                StatusCode::BAD_REQUEST,
                "invalid_request",
                "Invalid JSON request",
            )
        }
    };
    if let Err(error) = request.validate() {
        return capability_error(StatusCode::BAD_REQUEST, "invalid_request", error);
    }
    if request.limit as usize
        > batch_address_limit(auth, arete_solana_contracts::MAX_PAGE_SIZE as usize)
    {
        return capability_error(
            StatusCode::BAD_REQUEST,
            "invalid_request",
            "limit exceeds the session's address limit",
        );
    }
    let Some(provider) = provider else {
        return capability_error(
            StatusCode::NOT_IMPLEMENTED,
            "unsupported_capability",
            "Owner token discovery is not configured",
        );
    };
    match provider.owner_token_accounts(&request).await {
        Ok(page) => match page.validate(&request) {
            Ok(()) => json_response(
                StatusCode::OK,
                serde_json::to_value(page).expect("serializable page"),
            ),
            Err(error) => {
                capability_error(StatusCode::BAD_GATEWAY, "invalid_provider_response", error)
            }
        },
        Err(TokenDiscoveryError::InvalidCursor(error)) => {
            capability_error(StatusCode::BAD_REQUEST, "invalid_cursor", error)
        }
        Err(TokenDiscoveryError::Unsupported(error)) => {
            capability_error(StatusCode::NOT_IMPLEMENTED, "unsupported_capability", error)
        }
        Err(TokenDiscoveryError::Unavailable(error)) => {
            capability_error(StatusCode::BAD_GATEWAY, "provider_unavailable", error)
        }
    }
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct ContextualAccountBody {
    address: String,
    #[serde(default)]
    options: ReadOptions,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct ContextualAccountsBody {
    addresses: Vec<String>,
    #[serde(default)]
    options: ReadOptions,
}

async fn handle_contextual_accounts(
    req: Request<hyper::body::Incoming>,
    path: &str,
    rpc_url: Option<&str>,
    client: &ReadRpcClient,
    auth: Option<&crate::websocket::auth::AuthContext>,
) -> Response<Full<Bytes>> {
    let single = path == "/chain/v1/account";
    let (addresses, options) = if single {
        match read_json_body::<ContextualAccountBody>(req).await {
            Ok(body) => (vec![body.address], body.options),
            Err(response) if response.status() != StatusCode::BAD_REQUEST => return response,
            Err(_) => {
                return capability_error(
                    StatusCode::BAD_REQUEST,
                    "invalid_request",
                    "Invalid JSON request",
                )
            }
        }
    } else {
        match read_json_body::<ContextualAccountsBody>(req).await {
            Ok(body) => (body.addresses, body.options),
            Err(response) if response.status() != StatusCode::BAD_REQUEST => return response,
            Err(_) => {
                return capability_error(
                    StatusCode::BAD_REQUEST,
                    "invalid_request",
                    "Invalid JSON request",
                )
            }
        }
    };
    if addresses.len() > batch_address_limit(auth, MAX_CHAIN_BATCH_ADDRESSES) {
        return capability_error(
            StatusCode::BAD_REQUEST,
            "invalid_request",
            "addresses exceeds the batch limit",
        );
    }
    for address in &addresses {
        if let Err(error) = arete_solana_contracts::validate_address(address) {
            return capability_error(StatusCode::BAD_REQUEST, "invalid_request", error);
        }
    }
    if addresses.is_empty() {
        return json_response(StatusCode::OK, json!({ "context": null, "value": [] }));
    }
    let Some(rpc_url) = rpc_url else {
        return capability_error(
            StatusCode::SERVICE_UNAVAILABLE,
            "rpc_unavailable",
            "No RPC URL configured for chain reads",
        );
    };
    let (method, address) = if single {
        ("getAccountInfo", json!(addresses[0]))
    } else {
        ("getMultipleAccounts", json!(addresses))
    };
    match rpc_call(
        client,
        rpc_url,
        method,
        json!([address, options.rpc_config()]),
    )
    .await
    .and_then(|result| contextual_accounts_json(&addresses, single, options, &result))
    {
        Ok(value) => json_response(StatusCode::OK, value),
        Err(error) => {
            let (message, correlation_id) = upstream_error_message(&error);
            let code = "invalid_upstream_response";
            let mut response = json_response(
                StatusCode::BAD_GATEWAY,
                json!({ "code": code, "error": message, "correlationId": correlation_id }),
            );
            let headers = response.headers_mut();
            headers.insert("X-Error-Code", HeaderValue::from_static(code));
            if let Ok(value) = HeaderValue::from_str(&correlation_id) {
                headers.insert(CORRELATION_ID_HEADER, value);
            }
            response
        }
    }
}

fn contextual_accounts_json(
    addresses: &[String],
    single: bool,
    options: ReadOptions,
    result: &Value,
) -> anyhow::Result<Value> {
    let slot = result
        .pointer("/context/slot")
        .and_then(Value::as_u64)
        .ok_or_else(|| anyhow::anyhow!("Missing actual read context"))?;
    if options.min_context_slot.is_some_and(|min| slot < min) {
        anyhow::bail!("Read context is below minContextSlot");
    }
    let value = result
        .get("value")
        .ok_or_else(|| anyhow::anyhow!("Missing account result"))?;
    let values = if single {
        vec![value.clone()]
    } else {
        value
            .as_array()
            .ok_or_else(|| anyhow::anyhow!("Invalid batch result"))?
            .clone()
    };
    if values.len() != addresses.len() {
        anyhow::bail!("Account result count does not match addresses");
    }
    let mut items = Vec::new();
    for (address, account) in addresses.iter().zip(values) {
        if account.is_null() {
            items.push(Value::Null);
            continue;
        }
        if account.get("lamports").and_then(Value::as_u64).is_none()
            || account.get("owner").and_then(Value::as_str).is_none()
            || account.get("executable").and_then(Value::as_bool).is_none()
            || account.pointer("/data/0").and_then(Value::as_str).is_none()
            || account.pointer("/data/1").and_then(Value::as_str) != Some("base64")
        {
            anyhow::bail!("Malformed raw account result");
        }
        items.push(raw_account_json(address, &account));
    }
    Ok(serde_json::to_value(Contextual {
        context: Some(ReadContext { slot }),
        value: if single {
            items.remove(0)
        } else {
            json!(items)
        },
    })?)
}

#[derive(Debug, Deserialize)]
struct AddressesBody {
    addresses: Vec<String>,
}

#[derive(Debug, Deserialize)]
struct BalanceBody {
    owner: String,
    mint: String,
    #[serde(default, rename = "tokenProgram")]
    token_program: Option<String>,
    #[serde(default, rename = "minContextSlot")]
    min_context_slot: Option<String>,
}

#[derive(Debug, Deserialize)]
struct NativeBalanceBody {
    address: String,
    #[serde(default, rename = "minContextSlot")]
    min_context_slot: Option<String>,
}

#[derive(Debug, Deserialize)]
struct AccountsBody {
    addresses: Vec<String>,
}

#[derive(Default)]
struct HttpLimitState {
    per_subject_per_minute: DashMap<String, (u64, u32)>,
    /// Aggregate per-account buckets for the signed `account_limits`, kept
    /// apart from the per-session keys so the two can never collide.
    per_account_per_minute: DashMap<String, (u64, u32)>,
    /// The minute buckets were last swept in.
    swept_minute: std::sync::atomic::AtomicU64,
}

impl HttpLimitState {
    /// Drop the buckets of earlier minutes, at most once a minute.
    ///
    /// A bucket from an earlier minute is reset on its next use, so removing
    /// it changes no decision; without this, every session and account ever
    /// seen kept an entry for the life of the server.
    fn sweep(&self, now_bucket: u64) {
        let previous = self
            .swept_minute
            .swap(now_bucket, std::sync::atomic::Ordering::Relaxed);
        if previous == now_bucket {
            return;
        }
        for buckets in [&self.per_subject_per_minute, &self.per_account_per_minute] {
            buckets.retain(|_, (minute, _)| *minute >= now_bucket);
        }
    }
}

fn resolve_rpc_url() -> Option<String> {
    ["ARETE_READ_RPC_URL", "SOLANA_RPC_URL", "RPC_URL"]
        .iter()
        .find_map(|key| env::var(key).ok())
        .filter(|value| !value.is_empty())
}

fn json_response(status: StatusCode, value: Value) -> Response<Full<Bytes>> {
    Response::builder()
        .status(status)
        .header("Content-Type", "application/json")
        .body(Full::new(Bytes::from(value.to_string())))
        .unwrap()
}

fn error_response(status: StatusCode, message: impl Into<String>) -> Response<Full<Bytes>> {
    json_response(status, json!({ "error": message.into() }))
}

const CORRELATION_ID_HEADER: &str = "X-Correlation-Id";
const UPSTREAM_RPC_FAILED: &str = "Upstream RPC request failed";

/// A failure that came from the read RPC itself: transport, HTTP status, body or JSON-RPC error.
///
/// Its detail can name the upstream endpoint, which may carry a credential in its path or
/// query, so it is logged server-side and never returned to the caller.
#[derive(Debug)]
struct UpstreamRpcError(String);

impl std::fmt::Display for UpstreamRpcError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.0)
    }
}

impl std::error::Error for UpstreamRpcError {}

impl From<reqwest::Error> for UpstreamRpcError {
    fn from(error: reqwest::Error) -> Self {
        // reqwest's Display includes the request URL; drop it before anything formats it.
        let error = error.without_url();
        Self(format!(
            "{error} (timeout: {}, connect: {}, status: {:?})",
            error.is_timeout(),
            error.is_connect(),
            error.status().map(|status| status.as_u16()),
        ))
    }
}

/// The caller-safe message for a failed chain read, plus a correlation id for the server log.
fn upstream_error_message(error: &anyhow::Error) -> (String, String) {
    let correlation_id = uuid::Uuid::new_v4().to_string();
    warn!(
        correlation_id = %correlation_id,
        error = %error,
        "Chain read upstream request failed"
    );
    let message = if error.downcast_ref::<UpstreamRpcError>().is_some() {
        UPSTREAM_RPC_FAILED.to_string()
    } else {
        // Our own validation of the upstream result; these messages are static.
        error.to_string()
    };
    (message, correlation_id)
}

fn upstream_error_response(error: &anyhow::Error) -> Response<Full<Bytes>> {
    let (message, correlation_id) = upstream_error_message(error);
    let mut response = json_response(
        StatusCode::BAD_GATEWAY,
        json!({ "error": message, "correlationId": correlation_id }),
    );
    if let Ok(value) = HeaderValue::from_str(&correlation_id) {
        response.headers_mut().insert(CORRELATION_ID_HEADER, value);
    }
    response
}

fn parse_min_context_slot(value: Option<&str>) -> std::result::Result<Option<u64>, &'static str> {
    value
        .map(|slot| {
            slot.parse::<u64>()
                .map_err(|_| "minContextSlot must be a decimal u64 string")
        })
        .transpose()
}

fn auth_deny_response(deny: &AuthDeny) -> Response<Full<Bytes>> {
    let mut builder = Response::builder()
        .status(deny.http_status)
        .header("Content-Type", "application/json")
        .header("X-Error-Code", deny.code.as_str());

    if let Some(reset_at) = deny.reset_at {
        if let Ok(duration) = reset_at.duration_since(SystemTime::now()) {
            builder = builder.header("Retry-After", duration.as_secs().to_string());
        }
    }

    builder
        .body(Full::new(Bytes::from(
            json!({
                "error": deny.reason,
                "message": deny.reason,
                "code": deny.code.as_str(),
                "retryable": deny.code.should_retry(),
                "fatal": !deny.code.should_retry() && !deny.code.should_refresh_token(),
            })
            .to_string(),
        )))
        .unwrap()
}

#[allow(clippy::result_large_err)]
async fn authorize_http_request(
    remote_addr: SocketAddr,
    req: &Request<hyper::body::Incoming>,
    auth_plugin: Option<&Arc<dyn WebSocketAuthPlugin>>,
    limit_state: &HttpLimitState,
    required_scope: Option<&str>,
    enforce_read_limits: bool,
    solana_gateway_target_id: Option<&str>,
) -> std::result::Result<Option<crate::websocket::auth::AuthContext>, Response<Full<Bytes>>> {
    let Some(plugin) = auth_plugin else {
        return Ok(None);
    };

    let mut auth_request = ConnectionAuthRequest::from_http_request(remote_addr, req);
    // HTTP reads are bearer-only. Do not allow query-param session tokens here.
    auth_request.query = None;
    let decision = plugin.authorize(&auth_request).await;
    let context = match decision {
        AuthDecision::Allow(context) => context,
        AuthDecision::Deny(deny) => return Err(auth_deny_response(&deny)),
    };

    if let Some(target_id) = solana_gateway_target_id {
        if let Err(error) =
            arete_auth::SolanaGatewayAuthorization::validate_target(&context, target_id)
        {
            return Err(Response::builder()
                .status(StatusCode::FORBIDDEN)
                .header("Content-Type", "application/json")
                .header("X-Error-Code", "invalid_gateway_target")
                .body(Full::new(Bytes::from(
                    json!({
                        "error": "invalid_gateway_target",
                        "message": error.to_string(),
                        "code": "invalid_gateway_target",
                        "retryable": false,
                        "fatal": true
                    })
                    .to_string(),
                )))
                .expect("valid gateway target error response"));
        }
    }

    if let Some(required_scope) = required_scope {
        if !context.has_scope(required_scope) {
            return Err(json_response(
                StatusCode::FORBIDDEN,
                json!({
                    "error": "insufficient_scope",
                    "message": format!("Required scope: {required_scope}"),
                    "code": "insufficient_scope",
                    "retryable": false,
                    "fatal": true
                }),
            ));
        }
    }

    if enforce_read_limits {
        enforce_http_limits(&context, limit_state).map_err(|deny| auth_deny_response(&deny))?;
    }
    Ok(Some(context))
}

fn enforce_http_limits(
    context: &crate::websocket::auth::AuthContext,
    limit_state: &HttpLimitState,
) -> std::result::Result<(), Box<AuthDeny>> {
    enforce_http_limits_weighted(context, limit_state, 1)
}

/// Charge `weight` requests against the session's per-minute HTTP limit and,
/// for a token that names an account, the account's aggregate limit, all or
/// nothing.
///
/// The account limit is what bounds an account's reads across all of its
/// consumers and sessions; without it each new session would start with a
/// fresh allowance.
fn enforce_http_limits_weighted(
    context: &crate::websocket::auth::AuthContext,
    limit_state: &HttpLimitState,
    weight: usize,
) -> std::result::Result<(), Box<AuthDeny>> {
    if weight == 0 {
        return Ok(());
    }
    let weight = u32::try_from(weight).unwrap_or(u32::MAX);
    let now_bucket = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or(Duration::from_secs(0))
        .as_secs()
        / 60;
    limit_state.sweep(now_bucket);
    let denied = || {
        Box::new(AuthDeny::rate_limited(
            Duration::from_secs(60),
            "http reads",
        ))
    };

    let session = context.limits.max_http_requests_per_minute.map(|limit| {
        (
            format!("{}:{}", context.subject, context.metering_key),
            limit,
        )
    });
    if let Some((key, limit)) = &session {
        if !charge_bucket(
            &limit_state.per_subject_per_minute,
            key,
            *limit,
            weight,
            now_bucket,
        ) {
            return Err(denied());
        }
    }

    let account = (!context.is_legacy_policy() && context.account_key.is_some())
        .then_some(context.account_limits.max_http_requests_per_minute)
        .flatten();
    if let Some(limit) = account {
        if !charge_bucket(
            &limit_state.per_account_per_minute,
            context.account_key(),
            limit,
            weight,
            now_bucket,
        ) {
            // Refund the session charge so a refusal costs nothing.
            if let Some((key, _)) = &session {
                if let Some(mut entry) = limit_state.per_subject_per_minute.get_mut(key) {
                    if entry.0 == now_bucket {
                        entry.1 = entry.1.saturating_sub(weight);
                    }
                }
            }
            return Err(denied());
        }
    }
    Ok(())
}

/// Add `weight` to the current minute's bucket at `key` unless that would pass `limit`.
fn charge_bucket(
    buckets: &DashMap<String, (u64, u32)>,
    key: &str,
    limit: u32,
    weight: u32,
    now_bucket: u64,
) -> bool {
    let mut entry = buckets.entry(key.to_string()).or_insert((now_bucket, 0));
    if entry.0 != now_bucket {
        *entry = (now_bucket, 0);
    }
    if entry.1.saturating_add(weight) > limit {
        return false;
    }
    entry.1 += weight;
    true
}

/// How much request body a handler may read, and for how long. Attached to every request by
/// [`handle_request`].
#[derive(Clone, Copy, Debug)]
struct BodyLimits {
    max_bytes: usize,
    read_timeout: Duration,
}

impl Default for BodyLimits {
    fn default() -> Self {
        let limits = HttpLimitsConfig::default();
        Self {
            max_bytes: limits.max_body_bytes,
            read_timeout: limits.body_read_timeout,
        }
    }
}

fn body_too_large_response() -> Response<Full<Bytes>> {
    error_response(
        StatusCode::PAYLOAD_TOO_LARGE,
        "Request body exceeds the endpoint limit",
    )
}

fn body_timeout_response() -> Response<Full<Bytes>> {
    error_response(
        StatusCode::REQUEST_TIMEOUT,
        "Request body was not received in time",
    )
}

/// Read and parse a JSON body within the request's [`BodyLimits`]: `413` past the size limit,
/// `408` past the read deadline, `400` for anything that is not the expected JSON.
#[allow(clippy::result_large_err)]
async fn read_json_body<T: for<'de> Deserialize<'de>>(
    req: Request<hyper::body::Incoming>,
) -> std::result::Result<T, Response<Full<Bytes>>> {
    let limits = req
        .extensions()
        .get::<BodyLimits>()
        .copied()
        .unwrap_or_default();
    if req
        .headers()
        .get(hyper::header::CONTENT_LENGTH)
        .and_then(|value| value.to_str().ok())
        .and_then(|value| value.parse::<u64>().ok())
        .is_some_and(|length| length > limits.max_bytes as u64)
    {
        return Err(body_too_large_response());
    }
    let body = http_body_util::Limited::new(req.into_body(), limits.max_bytes);
    let collected = match tokio::time::timeout(limits.read_timeout, body.collect()).await {
        Ok(Ok(collected)) => collected,
        Ok(Err(error)) if error.is::<http_body_util::LengthLimitError>() => {
            return Err(body_too_large_response())
        }
        Ok(Err(_)) => {
            return Err(error_response(
                StatusCode::BAD_REQUEST,
                "Request body could not be read",
            ))
        }
        Err(_) => return Err(body_timeout_response()),
    };
    serde_json::from_slice::<T>(&collected.to_bytes())
        .map_err(|err| error_response(StatusCode::BAD_REQUEST, err.to_string()))
}

#[derive(Clone, Copy)]
enum ChainUsageOperation {
    Exists,
    Lamports,
    NativeBalance,
    RentExemption,
    Clock,
    Account,
    Mint,
    TokenAccount,
    Accounts,
    Balances,
    OwnerTokenAccounts,
    ContextualAccounts,
}

impl ChainUsageOperation {
    fn from_request(method: &Method, path: &str) -> Option<Self> {
        match (method.as_str(), path) {
            ("GET", path) if path.starts_with("/chain/exists/") => Some(Self::Exists),
            ("GET", path) if path.starts_with("/chain/lamports/") => Some(Self::Lamports),
            ("POST", "/chain/native-balance") => Some(Self::NativeBalance),
            ("GET", path) if path.starts_with("/chain/rent-exemption/") => {
                Some(Self::RentExemption)
            }
            ("GET", "/chain/clock") => Some(Self::Clock),
            ("GET", path) if path.starts_with("/chain/accounts/") => Some(Self::Account),
            ("GET", path) if path.starts_with("/chain/mints/") => Some(Self::Mint),
            ("GET", path) if path.starts_with("/chain/token-accounts/") => Some(Self::TokenAccount),
            ("POST", "/chain/accounts") => Some(Self::Accounts),
            ("POST", "/chain/v1/account") => Some(Self::Account),
            ("POST", "/chain/v1/accounts") => Some(Self::ContextualAccounts),
            ("POST", "/chain/v1/owner-token-accounts") => Some(Self::OwnerTokenAccounts),
            ("POST", "/chain/balances") => Some(Self::Balances),
            _ => None,
        }
    }

    fn name(self) -> &'static str {
        match self {
            Self::Exists => "exists",
            Self::Lamports => "lamports",
            Self::NativeBalance => "native_balance",
            Self::RentExemption => "rent_exemption",
            Self::Clock => "clock",
            Self::Account => "account",
            Self::Mint => "mint",
            Self::TokenAccount => "token_account",
            Self::Accounts => "accounts",
            Self::Balances => "balances",
            Self::OwnerTokenAccounts => "owner_token_accounts",
            Self::ContextualAccounts => "contextual_accounts",
        }
    }

    fn address_count(self, status: StatusCode, body: &[u8]) -> u64 {
        if !status.is_success() {
            return 0;
        }
        match self {
            Self::RentExemption | Self::Clock => 0,
            Self::Balances => 2,
            Self::ContextualAccounts | Self::OwnerTokenAccounts => {
                serde_json::from_slice::<Value>(body)
                    .ok()
                    .and_then(|value| {
                        value
                            .get(if matches!(self, Self::ContextualAccounts) {
                                "value"
                            } else {
                                "items"
                            })?
                            .as_array()
                            .map(|items| items.len() as u64)
                    })
                    .unwrap_or_default()
            }
            Self::Accounts => serde_json::from_slice::<Value>(body)
                .ok()
                .and_then(|value| value.get("items")?.as_array().map(|items| items.len()))
                .and_then(|count| count.try_into().ok())
                .unwrap_or_default(),
            _ => 1,
        }
    }
}

struct ChainRequestServices<'a> {
    rpc_url: Arc<Option<String>>,
    rpc_client: ReadRpcClient,
    auth_context: Option<crate::websocket::auth::AuthContext>,
    usage_observer: Option<&'a Arc<dyn SolanaGatewayUsageObserver>>,
    solana_gateway_target_id: Option<&'a str>,
    provider: Option<&'a Arc<dyn OwnerTokenAccountsProvider>>,
}

async fn handle_chain_request(
    req: Request<hyper::body::Incoming>,
    path: &str,
    services: ChainRequestServices<'_>,
) -> Response<Full<Bytes>> {
    let ChainRequestServices {
        rpc_url,
        rpc_client,
        auth_context,
        usage_observer,
        solana_gateway_target_id,
        provider,
    } = services;
    let operation = ChainUsageOperation::from_request(req.method(), path);
    let response = handle_chain_request_inner(
        req,
        path,
        rpc_url,
        rpc_client,
        auth_context.clone(),
        provider,
    )
    .await;
    let status = response.status();
    let (parts, body) = response.into_parts();
    let collected = match body.collect().await {
        Ok(collected) => collected,
        Err(error) => match error {},
    };
    let bytes = collected.to_bytes();
    if let (Some(operation), Some(observer), Some(target_id)) =
        (operation, usage_observer, solana_gateway_target_id)
    {
        observer
            .observe(SolanaGatewayUsageObservation {
                target_id: target_id.to_string(),
                auth_context,
                surface: SolanaGatewayUsageSurface::ChainRead,
                operation: operation.name(),
                result: if status.is_success() { "ok" } else { "error" },
                address_count: operation.address_count(status, &bytes),
                response_bytes: bytes.len().try_into().unwrap_or(u64::MAX),
            })
            .await;
    }
    Response::from_parts(parts, Full::new(bytes))
}

async fn handle_chain_request_inner(
    req: Request<hyper::body::Incoming>,
    path: &str,
    rpc_url: Arc<Option<String>>,
    rpc_client: ReadRpcClient,
    auth_context: Option<crate::websocket::auth::AuthContext>,
    provider: Option<&Arc<dyn OwnerTokenAccountsProvider>>,
) -> Response<Full<Bytes>> {
    if req.method() == Method::POST && path == "/chain/v1/owner-token-accounts" {
        return handle_owner_token_accounts(req, provider, auth_context.as_ref()).await;
    }
    if req.method() == Method::POST && matches!(path, "/chain/v1/account" | "/chain/v1/accounts") {
        return handle_contextual_accounts(
            req,
            path,
            rpc_url.as_ref().as_deref(),
            &rpc_client,
            auth_context.as_ref(),
        )
        .await;
    }
    let Some(rpc_url) = rpc_url.as_ref() else {
        return error_response(
            StatusCode::SERVICE_UNAVAILABLE,
            "No RPC URL configured for chain reads",
        );
    };

    match (req.method().as_str(), path) {
        ("GET", path) if path.starts_with("/chain/exists/") => {
            let address = path.trim_start_matches("/chain/exists/");
            match rpc_get_account_info(&rpc_client, rpc_url, address).await {
                Ok(value) => json_response(StatusCode::OK, json!({ "exists": !value.is_null() })),
                Err(err) => upstream_error_response(&err),
            }
        }
        ("GET", path) if path.starts_with("/chain/lamports/") => {
            let address = path.trim_start_matches("/chain/lamports/");
            match rpc_call(
                &rpc_client,
                rpc_url,
                "getBalance",
                json!([address, { "commitment": "confirmed" }]),
            )
            .await
            {
                Ok(value) => json_response(
                    StatusCode::OK,
                    json!({ "lamports": value.pointer("/value").and_then(Value::as_u64).unwrap_or(0) }),
                ),
                Err(err) => upstream_error_response(&err),
            }
        }
        ("POST", "/chain/native-balance") => match read_json_body::<NativeBalanceBody>(req).await {
            Ok(body) => {
                let min_context_slot =
                    match parse_min_context_slot(body.min_context_slot.as_deref()) {
                        Ok(slot) => slot,
                        Err(message) => return error_response(StatusCode::BAD_REQUEST, message),
                    };
                match rpc_get_native_balance(&rpc_client, rpc_url, &body.address, min_context_slot)
                    .await
                {
                    Ok(balance) => json_response(StatusCode::OK, balance),
                    Err(err) => upstream_error_response(&err),
                }
            }
            Err(response) => response,
        },
        ("GET", path) if path.starts_with("/chain/rent-exemption/") => {
            let raw_space = path.trim_start_matches("/chain/rent-exemption/");
            let Ok(space) = raw_space.parse::<u64>() else {
                return error_response(
                    StatusCode::BAD_REQUEST,
                    "rent-exemption space must be an integer",
                );
            };
            match rpc_call(
                &rpc_client,
                rpc_url,
                "getMinimumBalanceForRentExemption",
                json!([space, { "commitment": "confirmed" }]),
            )
            .await
            {
                Ok(value) => json_response(
                    StatusCode::OK,
                    json!({ "lamports": value.as_u64().unwrap_or(0) }),
                ),
                Err(err) => upstream_error_response(&err),
            }
        }
        ("GET", "/chain/clock") => {
            let slot = rpc_call(
                &rpc_client,
                rpc_url,
                "getSlot",
                json!([{ "commitment": "confirmed" }]),
            )
            .await;
            let epoch_info = rpc_call(
                &rpc_client,
                rpc_url,
                "getEpochInfo",
                json!([{ "commitment": "confirmed" }]),
            )
            .await;
            match (slot, epoch_info) {
                (Ok(slot_value), Ok(epoch_value)) => {
                    let slot_num = slot_value.as_u64().unwrap_or(0);
                    let unix_timestamp =
                        rpc_call(&rpc_client, rpc_url, "getBlockTime", json!([slot_num]))
                            .await
                            .ok()
                            .and_then(|value| value.as_i64())
                            .unwrap_or_default();
                    json_response(
                        StatusCode::OK,
                        json!({
                            "slot": slot_num,
                            "epoch": epoch_value.get("epoch").and_then(Value::as_u64),
                            "leaderScheduleEpoch": epoch_value.get("leaderScheduleSlotOffset").and_then(Value::as_u64),
                            "unixTimestamp": unix_timestamp,
                        }),
                    )
                }
                (Err(err), _) | (_, Err(err)) => upstream_error_response(&err),
            }
        }
        ("GET", path) if path.starts_with("/chain/accounts/") => {
            let address = path.trim_start_matches("/chain/accounts/");
            match rpc_get_account_info(&rpc_client, rpc_url, address).await {
                Ok(value) if value.is_null() => json_response(StatusCode::OK, Value::Null),
                Ok(value) => json_response(StatusCode::OK, raw_account_json(address, &value)),
                Err(err) => upstream_error_response(&err),
            }
        }
        ("GET", path) if path.starts_with("/chain/mints/") => {
            let address = path.trim_start_matches("/chain/mints/");
            match rpc_get_parsed_account_info(&rpc_client, rpc_url, address).await {
                Ok(Some(value)) => json_response(StatusCode::OK, mint_info_json(address, &value)),
                Ok(None) => json_response(StatusCode::OK, Value::Null),
                Err(err) => upstream_error_response(&err),
            }
        }
        ("GET", path) if path.starts_with("/chain/token-accounts/") => {
            let address = path.trim_start_matches("/chain/token-accounts/");
            match rpc_get_parsed_account_info(&rpc_client, rpc_url, address).await {
                Ok(Some(value)) => {
                    json_response(StatusCode::OK, token_account_json(address, &value))
                }
                Ok(None) => json_response(StatusCode::OK, Value::Null),
                Err(err) => upstream_error_response(&err),
            }
        }
        ("POST", "/chain/accounts") => match read_json_body::<AccountsBody>(req).await {
            Ok(body) => {
                let configured_limit =
                    batch_address_limit(auth_context.as_ref(), MAX_CHAIN_BATCH_ADDRESSES);
                if body.addresses.len() > configured_limit {
                    return error_response(
                        StatusCode::BAD_REQUEST,
                        format!(
                            "addresses exceeds the {configured_limit}-address limit for one batch"
                        ),
                    );
                }
                if body.addresses.is_empty() {
                    return json_response(StatusCode::OK, json!({ "items": [] }));
                }
                match rpc_get_multiple_accounts(&rpc_client, rpc_url, &body.addresses).await {
                    // A short array would silently misalign items with addresses, so treat a
                    // length mismatch as an upstream fault rather than padding it.
                    Ok(values) if values.len() == body.addresses.len() => json_response(
                        StatusCode::OK,
                        json!({ "items": batch_accounts_json(&body.addresses, &values) }),
                    ),
                    Ok(values) => error_response(
                        StatusCode::BAD_GATEWAY,
                        format!(
                            "getMultipleAccounts returned {} entries for {} addresses",
                            values.len(),
                            body.addresses.len()
                        ),
                    ),
                    Err(err) => upstream_error_response(&err),
                }
            }
            Err(response) => response,
        },
        ("POST", "/chain/balances") => match read_json_body::<BalanceBody>(req).await {
            Ok(body) => {
                let min_context_slot =
                    match parse_min_context_slot(body.min_context_slot.as_deref()) {
                        Ok(slot) => slot,
                        Err(message) => return error_response(StatusCode::BAD_REQUEST, message),
                    };
                match rpc_get_token_balance(
                    &rpc_client,
                    rpc_url,
                    &body.owner,
                    &body.mint,
                    body.token_program.as_deref(),
                    min_context_slot,
                )
                .await
                {
                    Ok(balance) => json_response(StatusCode::OK, balance),
                    Err(err) => upstream_error_response(&err),
                }
            }
            Err(response) => response,
        },
        _ => error_response(StatusCode::NOT_FOUND, "Not Found"),
    }
}

async fn handle_program_account_request(
    req: Request<hyper::body::Incoming>,
    path: &str,
    rpc_url: Arc<Option<String>>,
    rpc_client: ReadRpcClient,
    program_runtime_catalog: Arc<ProgramRuntimeCatalog>,
    auth_context: Option<crate::websocket::auth::AuthContext>,
    program_read_binding_target_id: Arc<Option<String>>,
) -> Response<Full<Bytes>> {
    let route = match ProgramReadRoute::parse(req.method(), path) {
        Ok(route) => route,
        Err(error) => return program_read_error_response(error),
    };
    let release_hash = match route.release_hash.parse() {
        Ok(release_hash) => release_hash,
        Err(_) => return program_read_error_response(ProgramReadError::InvalidReleaseHash),
    };
    let Some(definition) = program_runtime_catalog.get(&release_hash).cloned() else {
        return program_read_error_response(ProgramReadError::ReleaseNotFound);
    };
    if let Err(error) = authorize_program_read(
        auth_context.as_ref(),
        program_read_binding_target_id.as_deref(),
        &definition,
    ) {
        return with_release_metadata(program_read_error_response(error), &definition, None, None);
    }
    if matches!(route.operation, ProgramReadOperation::NativeQuery) {
        return with_release_metadata(
            capability_error(
                StatusCode::NOT_IMPLEMENTED,
                "unsupported_capability",
                "Native position query requires a platform service for this release/type/filter",
            ),
            &definition,
            None,
            None,
        );
    }
    let Some(rpc_url) = rpc_url.as_ref() else {
        return with_release_metadata(
            program_read_error_response(ProgramReadError::RpcNotConfigured),
            &definition,
            None,
            None,
        );
    };

    match route.operation {
        ProgramReadOperation::NativeQuery => unreachable!("handled before RPC admission"),
        ProgramReadOperation::ContextualFetch { address } => {
            let options = match read_json_body::<ProgramContextOptionsBody>(req).await {
                Ok(body) => body.options,
                Err(response) if response.status() != StatusCode::BAD_REQUEST => return response,
                Err(_) => return program_read_error_response(ProgramReadError::InvalidRequest),
            };
            handle_contextual_program_accounts(
                &definition,
                &route.account,
                ContextualProgramRead {
                    addresses: vec![address],
                    single: true,
                    options,
                },
                rpc_url,
                &rpc_client,
                auth_context.as_ref(),
            )
            .await
        }
        ProgramReadOperation::ContextualBatch => {
            let body = match read_json_body::<ContextualAccountsBody>(req).await {
                Ok(body) => body,
                Err(response) if response.status() != StatusCode::BAD_REQUEST => return response,
                Err(_) => return program_read_error_response(ProgramReadError::InvalidRequest),
            };
            handle_contextual_program_accounts(
                &definition,
                &route.account,
                ContextualProgramRead {
                    addresses: body.addresses,
                    single: false,
                    options: body.options,
                },
                rpc_url,
                &rpc_client,
                auth_context.as_ref(),
            )
            .await
        }
        ProgramReadOperation::Fetch { address } => {
            let account_value = match rpc_get_account_info(&rpc_client, rpc_url, &address).await {
                Ok(value) => value,
                Err(error) => {
                    error!("Program account RPC read failed: {}", error);
                    return with_release_metadata(
                        program_read_error_response(ProgramReadError::RpcFailed),
                        &definition,
                        Some(&address),
                        None,
                    );
                }
            };
            match decode_release_account(&definition, &route.account, &account_value).await {
                AccountReadOutcome::Missing => with_release_metadata(
                    json_response(StatusCode::OK, Value::Null),
                    &definition,
                    Some(&address),
                    Some(false),
                ),
                AccountReadOutcome::Value(value) => with_release_metadata(
                    json_response(StatusCode::OK, value),
                    &definition,
                    Some(&address),
                    Some(true),
                ),
                AccountReadOutcome::Error(error) => with_release_metadata(
                    program_read_error_response(error),
                    &definition,
                    Some(&address),
                    Some(true),
                ),
            }
        }
        ProgramReadOperation::Exists { address } => {
            let account_value = match rpc_get_account_info(&rpc_client, rpc_url, &address).await {
                Ok(value) => value,
                Err(error) => {
                    error!("Program account existence RPC read failed: {}", error);
                    return with_release_metadata(
                        program_read_error_response(ProgramReadError::RpcFailed),
                        &definition,
                        Some(&address),
                        None,
                    );
                }
            };
            let exists = !account_value.is_null();
            if exists && !account_owner_matches(&definition, &account_value) {
                return with_release_metadata(
                    program_read_error_response(ProgramReadError::AccountOwnerMismatch),
                    &definition,
                    Some(&address),
                    Some(true),
                );
            }
            with_release_metadata(
                json_response(StatusCode::OK, json!({ "exists": exists })),
                &definition,
                Some(&address),
                Some(exists),
            )
        }
        ProgramReadOperation::Batch => {
            let body = match read_json_body::<AddressesBody>(req).await {
                Ok(body) => body,
                Err(response) if response.status() != StatusCode::BAD_REQUEST => return response,
                Err(_) => return program_read_error_response(ProgramReadError::InvalidRequest),
            };
            let configured_limit =
                batch_address_limit(auth_context.as_ref(), MAX_PROGRAM_BATCH_ADDRESSES);
            if body.addresses.len() > configured_limit {
                return with_release_metadata(
                    program_read_error_response(ProgramReadError::BatchLimitExceeded),
                    &definition,
                    None,
                    None,
                );
            }
            if body.addresses.is_empty() {
                return with_release_metadata(
                    json_response(StatusCode::OK, json!({ "items": [] })),
                    &definition,
                    None,
                    None,
                );
            }

            let values =
                match rpc_get_multiple_accounts(&rpc_client, rpc_url, &body.addresses).await {
                    Ok(values) if values.len() == body.addresses.len() => values,
                    Ok(_) => {
                        return with_release_metadata(
                            program_read_error_response(ProgramReadError::RpcResponseInvalid),
                            &definition,
                            None,
                            None,
                        )
                    }
                    Err(error) => {
                        error!("Program account batch RPC read failed: {}", error);
                        return with_release_metadata(
                            program_read_error_response(ProgramReadError::RpcFailed),
                            &definition,
                            None,
                            None,
                        );
                    }
                };

            let outcomes = futures_util::future::join_all(
                values
                    .iter()
                    .map(|value| decode_release_account(&definition, &route.account, value)),
            )
            .await;
            let mut items = Vec::with_capacity(outcomes.len());
            for (address, outcome) in body.addresses.iter().zip(outcomes) {
                let item = match outcome {
                    AccountReadOutcome::Missing => {
                        json!({ "address": address, "status": "missing" })
                    }
                    AccountReadOutcome::Value(value) => {
                        json!({ "address": address, "status": "ok", "value": value })
                    }
                    AccountReadOutcome::Error(error) => json!({
                        "address": address,
                        "status": "error",
                        "error": { "code": error.code() }
                    }),
                };
                items.push(item);
            }
            with_release_metadata(
                json_response(StatusCode::OK, json!({ "items": items })),
                &definition,
                None,
                None,
            )
        }
    }
}

const MAX_PROGRAM_BATCH_ADDRESSES: usize = 100;
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct ProgramContextOptionsBody {
    #[serde(default)]
    options: ReadOptions,
}

struct ContextualProgramRead {
    addresses: Vec<String>,
    single: bool,
    options: ReadOptions,
}

async fn handle_contextual_program_accounts(
    definition: &ProgramRuntimeDefinition,
    account: &str,
    read: ContextualProgramRead,
    rpc_url: &str,
    client: &ReadRpcClient,
    auth: Option<&crate::websocket::auth::AuthContext>,
) -> Response<Full<Bytes>> {
    let ContextualProgramRead {
        addresses,
        single,
        options,
    } = read;
    if addresses.len() > batch_address_limit(auth, MAX_PROGRAM_BATCH_ADDRESSES) {
        return with_release_metadata(
            program_read_error_response(ProgramReadError::BatchLimitExceeded),
            definition,
            None,
            None,
        );
    }
    for address in &addresses {
        if arete_solana_contracts::validate_address(address).is_err() {
            return program_read_error_response(ProgramReadError::InvalidRequest);
        }
    }
    if addresses.is_empty() {
        return with_release_metadata(
            json_response(
                StatusCode::OK,
                json!({ "context": null, "value": { "items": [] } }),
            ),
            definition,
            None,
            None,
        );
    }
    let method = if single {
        "getAccountInfo"
    } else {
        "getMultipleAccounts"
    };
    let input = if single {
        json!(addresses[0])
    } else {
        json!(addresses)
    };
    let result = match rpc_call(
        client,
        rpc_url,
        method,
        json!([input, options.rpc_config()]),
    )
    .await
    {
        Ok(result) => result,
        Err(_) => {
            return with_release_metadata(
                program_read_error_response(ProgramReadError::RpcFailed),
                definition,
                None,
                None,
            )
        }
    };
    let context = match result.pointer("/context/slot").and_then(Value::as_u64) {
        Some(slot) if options.min_context_slot.is_none_or(|min| slot >= min) => {
            ReadContext { slot }
        }
        _ => {
            return with_release_metadata(
                program_read_error_response(ProgramReadError::RpcResponseInvalid),
                definition,
                None,
                None,
            )
        }
    };
    let values = match result.get("value") {
        Some(value) if single => vec![value.clone()],
        Some(Value::Array(values)) if values.len() == addresses.len() => values.clone(),
        _ => {
            return with_release_metadata(
                program_read_error_response(ProgramReadError::RpcResponseInvalid),
                definition,
                None,
                None,
            )
        }
    };
    let outcomes = futures_util::future::join_all(
        values
            .iter()
            .map(|value| decode_release_account(definition, account, value)),
    )
    .await;
    let mut items = Vec::new();
    for (address, outcome) in addresses.iter().zip(outcomes) {
        if single {
            let (value, exists) = match outcome {
                AccountReadOutcome::Missing => (Value::Null, false),
                AccountReadOutcome::Value(value) => (value, true),
                AccountReadOutcome::Error(error) => {
                    return with_release_metadata(
                        program_read_error_response(error),
                        definition,
                        Some(address),
                        Some(true),
                    )
                }
            };
            return with_release_metadata(
                json_response(
                    StatusCode::OK,
                    json!({ "context": context, "value": value }),
                ),
                definition,
                Some(address),
                Some(exists),
            );
        }
        items.push(match outcome {
            AccountReadOutcome::Missing => json!({ "address": address, "status": "missing" }),
            AccountReadOutcome::Value(value) => {
                json!({ "address": address, "status": "ok", "value": value })
            }
            AccountReadOutcome::Error(error) => {
                json!({ "address": address, "status": "error", "error": { "code": error.code() } })
            }
        });
    }
    with_release_metadata(
        json_response(
            StatusCode::OK,
            json!({ "context": context, "value": { "items": items } }),
        ),
        definition,
        None,
        None,
    )
}

const MAX_PROGRAM_ACCOUNT_BYTES: usize = 10 * 1024 * 1024;
const PROGRAM_DECODE_TIMEOUT: Duration = Duration::from_secs(2);

fn authorize_program_read(
    auth_context: Option<&crate::websocket::auth::AuthContext>,
    expected_target_id: Option<&str>,
    definition: &ProgramRuntimeDefinition,
) -> std::result::Result<(), ProgramReadError> {
    let Some(context) = auth_context else {
        return Ok(());
    };
    let Some(expected_target_id) = expected_target_id.filter(|target_id| !target_id.is_empty())
    else {
        return Err(ProgramReadError::AuthorizationNotConfigured);
    };

    arete_auth::ProgramReadAuthorization::try_from_context(
        context,
        expected_target_id,
        &definition.program_id,
        &definition.program_release_hash.to_string(),
    )
    .map(|_| ())
    .map_err(|_| ProgramReadError::Unauthorized)
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum ProgramReadError {
    NotFound,
    ProgramReadsDisabled,
    InvalidReleaseHash,
    ReleaseNotFound,
    InvalidRequest,
    Unauthorized,
    AuthorizationNotConfigured,
    BatchLimitExceeded,
    RpcNotConfigured,
    RpcFailed,
    RpcResponseInvalid,
    AccountOwnerMismatch,
    AccountDataInvalid,
    AccountDataTooLarge,
    AccountDecodeFailed,
    AccountDecodeTimeout,
}

impl ProgramReadError {
    fn code(self) -> &'static str {
        match self {
            Self::NotFound => "NOT_FOUND",
            Self::ProgramReadsDisabled => "PROGRAM_READS_DISABLED",
            Self::InvalidReleaseHash => "INVALID_PROGRAM_RELEASE_HASH",
            Self::ReleaseNotFound => "PROGRAM_RELEASE_NOT_FOUND",
            Self::InvalidRequest => "INVALID_REQUEST",
            Self::Unauthorized => "PROGRAM_READ_UNAUTHORIZED",
            Self::AuthorizationNotConfigured => "PROGRAM_READ_AUTH_NOT_CONFIGURED",
            Self::BatchLimitExceeded => "BATCH_LIMIT_EXCEEDED",
            Self::RpcNotConfigured => "READ_RPC_NOT_CONFIGURED",
            Self::RpcFailed => "RPC_REQUEST_FAILED",
            Self::RpcResponseInvalid => "RPC_RESPONSE_INVALID",
            Self::AccountOwnerMismatch => "ACCOUNT_OWNER_MISMATCH",
            Self::AccountDataInvalid => "ACCOUNT_DATA_INVALID",
            Self::AccountDataTooLarge => "ACCOUNT_DATA_TOO_LARGE",
            Self::AccountDecodeFailed => "ACCOUNT_DECODE_FAILED",
            Self::AccountDecodeTimeout => "ACCOUNT_DECODE_TIMEOUT",
        }
    }

    fn status(self) -> StatusCode {
        match self {
            Self::NotFound | Self::ReleaseNotFound => StatusCode::NOT_FOUND,
            Self::InvalidReleaseHash | Self::InvalidRequest => StatusCode::BAD_REQUEST,
            Self::Unauthorized => StatusCode::FORBIDDEN,
            Self::AuthorizationNotConfigured => StatusCode::SERVICE_UNAVAILABLE,
            Self::BatchLimitExceeded | Self::AccountDataTooLarge => StatusCode::PAYLOAD_TOO_LARGE,
            Self::ProgramReadsDisabled | Self::RpcNotConfigured => StatusCode::SERVICE_UNAVAILABLE,
            Self::RpcFailed | Self::RpcResponseInvalid | Self::AccountDataInvalid => {
                StatusCode::BAD_GATEWAY
            }
            Self::AccountOwnerMismatch | Self::AccountDecodeFailed => {
                StatusCode::UNPROCESSABLE_ENTITY
            }
            Self::AccountDecodeTimeout => StatusCode::GATEWAY_TIMEOUT,
        }
    }
}

fn program_read_error_response(error: ProgramReadError) -> Response<Full<Bytes>> {
    Response::builder()
        .status(error.status())
        .header("Content-Type", "application/json")
        .header("X-Error-Code", error.code())
        .body(Full::new(Bytes::from(
            json!({ "error": { "code": error.code() } }).to_string(),
        )))
        .expect("valid program read error response")
}

#[derive(Debug)]
struct ProgramReadRoute {
    release_hash: String,
    account: String,
    operation: ProgramReadOperation,
}

#[derive(Debug)]
enum ProgramReadOperation {
    ContextualFetch { address: String },
    ContextualBatch,
    NativeQuery,
    Fetch { address: String },
    Batch,
    Exists { address: String },
}

impl ProgramReadRoute {
    fn parse(method: &Method, path: &str) -> std::result::Result<Self, ProgramReadError> {
        let segments: Vec<&str> = path.trim_start_matches('/').split('/').collect();
        if segments.len() < 5
            || segments[0] != "v1"
            || segments[1] != "releases"
            || segments[2].is_empty()
            || segments[3] != "accounts"
            || segments[4].is_empty()
        {
            return Err(ProgramReadError::NotFound);
        }
        let operation = match (method, segments.as_slice()) {
            (&Method::POST, [_, _, _, _, _]) => ProgramReadOperation::Batch,
            (&Method::POST, [_, _, _, _, _, "query"]) => ProgramReadOperation::NativeQuery,
            (&Method::POST, [_, _, _, _, _, "context"]) => ProgramReadOperation::ContextualBatch,
            (&Method::POST, [_, _, _, _, _, address, "context"]) if !address.is_empty() => {
                ProgramReadOperation::ContextualFetch {
                    address: (*address).to_string(),
                }
            }
            (&Method::GET, [_, _, _, _, _, address]) if !address.is_empty() => {
                ProgramReadOperation::Fetch {
                    address: (*address).to_string(),
                }
            }
            (&Method::GET, [_, _, _, _, _, address, "exists"]) if !address.is_empty() => {
                ProgramReadOperation::Exists {
                    address: (*address).to_string(),
                }
            }
            _ => return Err(ProgramReadError::NotFound),
        };
        Ok(Self {
            release_hash: segments[2].to_string(),
            account: segments[4].to_string(),
            operation,
        })
    }
}

enum AccountReadOutcome {
    Missing,
    Value(Value),
    Error(ProgramReadError),
}

fn account_owner_matches(definition: &ProgramRuntimeDefinition, value: &Value) -> bool {
    value.get("owner").and_then(Value::as_str) == Some(definition.program_id.as_str())
}

async fn decode_release_account(
    definition: &ProgramRuntimeDefinition,
    account: &str,
    value: &Value,
) -> AccountReadOutcome {
    if value.is_null() {
        return AccountReadOutcome::Missing;
    }
    if !account_owner_matches(definition, value) {
        return AccountReadOutcome::Error(ProgramReadError::AccountOwnerMismatch);
    }
    let data = match decode_account_bytes(value) {
        Some(data) => data,
        None => return AccountReadOutcome::Error(ProgramReadError::AccountDataInvalid),
    };
    if data.len() > MAX_PROGRAM_ACCOUNT_BYTES {
        return AccountReadOutcome::Error(ProgramReadError::AccountDataTooLarge);
    }

    let reader = definition.account_reader.clone();
    let account = account.to_string();
    let decode = tokio::task::spawn_blocking(move || reader(&account, &data));
    match tokio::time::timeout(PROGRAM_DECODE_TIMEOUT, decode).await {
        Ok(Ok(Ok(value))) => AccountReadOutcome::Value(value),
        Ok(Ok(Err(_))) | Ok(Err(_)) => {
            AccountReadOutcome::Error(ProgramReadError::AccountDecodeFailed)
        }
        Err(_) => AccountReadOutcome::Error(ProgramReadError::AccountDecodeTimeout),
    }
}

fn with_release_metadata(
    mut response: Response<Full<Bytes>>,
    definition: &ProgramRuntimeDefinition,
    address: Option<&str>,
    exists: Option<bool>,
) -> Response<Full<Bytes>> {
    let headers = response.headers_mut();
    if let Ok(value) = HeaderValue::from_str(&definition.program_release_hash.to_string()) {
        headers.insert("X-Arete-Program-Release-Hash", value);
    }
    if let Ok(value) = HeaderValue::from_str(&definition.idl_content_hash.to_string()) {
        headers.insert("X-Arete-Idl-Content-Hash", value);
    }
    if let Some(address) = address {
        if let Ok(value) = HeaderValue::from_str(address) {
            headers.insert("X-Arete-Account-Address", value);
        }
    }
    if let Some(exists) = exists {
        headers.insert(
            "X-Arete-Account-Exists",
            HeaderValue::from_static(if exists { "true" } else { "false" }),
        );
    }
    response
}

/// The read RPC client and the largest response body it accepts. Connect and total timeouts
/// are set on the client itself.
#[derive(Clone)]
struct ReadRpcClient {
    client: Client,
    max_response_bytes: usize,
}

impl std::ops::Deref for ReadRpcClient {
    type Target = Client;

    fn deref(&self) -> &Client {
        &self.client
    }
}

/// Read a response body, failing once it passes `max_bytes` instead of buffering the rest.
async fn read_capped_response(
    response: reqwest::Response,
    max_bytes: usize,
) -> std::result::Result<Vec<u8>, UpstreamRpcError> {
    let too_large = || UpstreamRpcError(format!("response exceeds {max_bytes} bytes"));
    if response
        .content_length()
        .is_some_and(|length| length > max_bytes as u64)
    {
        return Err(too_large());
    }
    let mut stream = response.bytes_stream();
    let mut bytes = Vec::new();
    while let Some(chunk) = futures_util::StreamExt::next(&mut stream).await {
        let chunk = chunk.map_err(UpstreamRpcError::from)?;
        if bytes.len().saturating_add(chunk.len()) > max_bytes {
            return Err(too_large());
        }
        bytes.extend_from_slice(&chunk);
    }
    Ok(bytes)
}

async fn rpc_call(
    client: &ReadRpcClient,
    rpc_url: &str,
    method: &str,
    params: Value,
) -> anyhow::Result<Value> {
    let response = client
        .post(rpc_url)
        .json(&json!({
            "jsonrpc": "2.0",
            "id": "arete-read",
            "method": method,
            "params": params,
        }))
        .send()
        .await
        .and_then(reqwest::Response::error_for_status)
        .map_err(UpstreamRpcError::from)?;
    let body = read_capped_response(response, client.max_response_bytes).await?;
    let value = serde_json::from_slice::<Value>(&body)
        .map_err(|error| UpstreamRpcError(format!("invalid JSON-RPC response: {error}")))?;
    if let Some(error) = value.get("error") {
        return Err(UpstreamRpcError(format!("JSON-RPC error: {error}")).into());
    }
    Ok(value.get("result").cloned().unwrap_or(Value::Null))
}

fn rpc_read_config(encoding: Option<&str>, min_context_slot: Option<u64>) -> Value {
    let mut config = json!({ "commitment": "confirmed" });
    let object = config
        .as_object_mut()
        .expect("RPC read config is always an object");
    if let Some(encoding) = encoding {
        object.insert("encoding".to_string(), json!(encoding));
    }
    if let Some(min_context_slot) = min_context_slot {
        object.insert("minContextSlot".to_string(), json!(min_context_slot));
    }
    config
}

async fn rpc_get_native_balance(
    client: &ReadRpcClient,
    rpc_url: &str,
    address: &str,
    min_context_slot: Option<u64>,
) -> anyhow::Result<Value> {
    let result = rpc_call(
        client,
        rpc_url,
        "getBalance",
        json!([address, rpc_read_config(None, min_context_slot)]),
    )
    .await?;
    contextual_native_balance_json(&result)
}

fn contextual_native_balance_json(result: &Value) -> anyhow::Result<Value> {
    let lamports = result
        .get("value")
        .and_then(Value::as_u64)
        .ok_or_else(|| anyhow::anyhow!("getBalance response is missing a u64 value"))?;
    let context_slot = result
        .pointer("/context/slot")
        .and_then(Value::as_u64)
        .ok_or_else(|| anyhow::anyhow!("getBalance response is missing a u64 context slot"))?;
    Ok(json!({
        "lamports": lamports.to_string(),
        "contextSlot": context_slot.to_string(),
    }))
}

async fn rpc_get_account_info(
    client: &ReadRpcClient,
    rpc_url: &str,
    address: &str,
) -> anyhow::Result<Value> {
    let result = rpc_call(
        client,
        rpc_url,
        "getAccountInfo",
        json!([address, { "encoding": "base64", "commitment": "confirmed" }]),
    )
    .await?;
    Ok(result.get("value").cloned().unwrap_or(Value::Null))
}

async fn rpc_get_multiple_accounts(
    client: &ReadRpcClient,
    rpc_url: &str,
    addresses: &[String],
) -> anyhow::Result<Vec<Value>> {
    let result = rpc_call(
        client,
        rpc_url,
        "getMultipleAccounts",
        json!([addresses, { "encoding": "base64", "commitment": "confirmed" }]),
    )
    .await?;
    Ok(result
        .get("value")
        .and_then(Value::as_array)
        .cloned()
        .unwrap_or_default())
}

async fn rpc_get_parsed_account_info(
    client: &ReadRpcClient,
    rpc_url: &str,
    address: &str,
) -> anyhow::Result<Option<Value>> {
    let result = rpc_call(
        client,
        rpc_url,
        "getAccountInfo",
        json!([address, { "encoding": "jsonParsed", "commitment": "confirmed" }]),
    )
    .await?;
    Ok(result
        .get("value")
        .cloned()
        .filter(|value| !value.is_null()))
}

async fn rpc_get_token_balance(
    client: &ReadRpcClient,
    rpc_url: &str,
    owner: &str,
    mint: &str,
    token_program: Option<&str>,
    min_context_slot: Option<u64>,
) -> anyhow::Result<Value> {
    let filter = token_program
        .map(|program_id| json!({ "programId": program_id }))
        .unwrap_or_else(|| json!({ "mint": mint }));
    let result = rpc_call(
        client,
        rpc_url,
        "getTokenAccountsByOwner",
        json!([
            owner,
            filter,
            rpc_read_config(Some("jsonParsed"), min_context_slot)
        ]),
    )
    .await?;

    contextual_token_balance_json(&result, owner, mint, token_program)
}

fn contextual_token_balance_json(
    result: &Value,
    owner: &str,
    mint: &str,
    token_program: Option<&str>,
) -> anyhow::Result<Value> {
    let context_slot = result
        .pointer("/context/slot")
        .and_then(Value::as_u64)
        .ok_or_else(|| {
            anyhow::anyhow!("getTokenAccountsByOwner response is missing a u64 context slot")
        })?;

    let account = result
        .get("value")
        .and_then(Value::as_array)
        .and_then(|items| {
            items.iter().find(|item| {
                item.pointer("/account/data/parsed/info/mint")
                    .and_then(Value::as_str)
                    == Some(mint)
            })
        })
        .cloned();

    if let Some(account) = account {
        let pubkey = account.get("pubkey").and_then(Value::as_str);
        let info = account.pointer("/account/data/parsed/info");
        return Ok(json!({
            "exists": true,
            "address": pubkey,
            "owner": owner,
            "mint": mint,
            "tokenProgram": token_program,
            "amount": info.and_then(|value| value.pointer("/tokenAmount/amount")).and_then(Value::as_str).unwrap_or("0"),
            "decimals": info.and_then(|value| value.pointer("/tokenAmount/decimals")).and_then(Value::as_u64),
            "uiAmountString": info.and_then(|value| value.pointer("/tokenAmount/uiAmountString")).and_then(Value::as_str),
            "contextSlot": context_slot.to_string(),
        }));
    }

    Ok(json!({
        "exists": false,
        "address": Value::Null,
        "owner": owner,
        "mint": mint,
        "tokenProgram": token_program,
        "amount": "0",
        "decimals": Value::Null,
        "uiAmountString": Value::Null,
        "contextSlot": context_slot.to_string(),
    }))
}

fn decode_account_bytes(value: &Value) -> Option<Vec<u8>> {
    let data = value.get("data")?.as_array()?;
    let encoded = data.first()?.as_str()?;
    BASE64_STANDARD.decode(encoded).ok()
}

/// `lamports` is a decimal string, not a JSON number.
///
/// A `u64` above 2^53 does not survive a JSON number in every client: JavaScript rounds
/// `9007199254740993` to `...92` during `JSON.parse`, before any SDK can intervene. The
/// native-balance route already answers with a string for this reason, and the batch route makes it
/// matter — a custody sweep reading many accounts at once is exactly where a silently rounded
/// balance would be acted on.
fn raw_account_json(address: &str, value: &Value) -> Value {
    json!({
        "address": address,
        "ownerProgram": value.get("owner").and_then(Value::as_str).unwrap_or_default(),
        "lamports": value
            .get("lamports")
            .and_then(Value::as_u64)
            .unwrap_or(0)
            .to_string(),
        "executable": value.get("executable").and_then(Value::as_bool).unwrap_or(false),
        "data": value
            .pointer("/data/0")
            .and_then(Value::as_str)
            .unwrap_or_default(),
    })
}

/// How many addresses one batch may carry: the session's own ceiling, bounded by the protocol's.
///
/// Shared by every batch route rather than repeated at each one. Rate limiting charges a batch as a
/// single HTTP request, so a route that consulted only the protocol limit would let a session
/// capped at ten addresses read a hundred for the same cost. That is exactly what
/// `POST /chain/accounts` did while the program-account batch computed it correctly, so the two
/// call sites now derive it from one place and cannot drift apart again.
pub(crate) fn batch_address_limit(
    auth_context: Option<&crate::websocket::auth::AuthContext>,
    protocol_max: usize,
) -> usize {
    auth_context
        .and_then(|ctx| ctx.limits.max_http_batch_addresses)
        .map(|limit| limit as usize)
        .unwrap_or(protocol_max)
        .min(protocol_max)
}

/// Solana's own `getMultipleAccounts` ceiling, so one batch is one upstream call.
const MAX_CHAIN_BATCH_ADDRESSES: usize = 100;

/// Positionally aligned with `addresses`; `null` where the account is absent, matching
/// both `getMultipleAccounts` and the single-address route.
fn batch_accounts_json(addresses: &[String], values: &[Value]) -> Vec<Value> {
    addresses
        .iter()
        .zip(values)
        .map(|(address, value)| {
            if value.is_null() {
                Value::Null
            } else {
                raw_account_json(address, value)
            }
        })
        .collect()
}

fn mint_info_json(address: &str, value: &Value) -> Value {
    let owner_program = value
        .get("owner")
        .and_then(Value::as_str)
        .unwrap_or_default();
    let info = value.pointer("/data/parsed/info");
    json!({
        "address": address,
        "ownerProgram": owner_program,
        "decimals": info.and_then(|v| v.get("decimals")).and_then(Value::as_u64),
        "supply": info.and_then(|v| v.get("supply")).and_then(Value::as_str),
        "mintAuthority": info.and_then(|v| v.get("mintAuthority")).and_then(Value::as_str),
        "freezeAuthority": info.and_then(|v| v.get("freezeAuthority")).and_then(Value::as_str),
    })
}

fn token_account_json(address: &str, value: &Value) -> Value {
    let owner_program = value
        .get("owner")
        .and_then(Value::as_str)
        .unwrap_or_default();
    let info = value.pointer("/data/parsed/info");
    json!({
        "address": address,
        "ownerProgram": owner_program,
        "mint": info.and_then(|v| v.get("mint")).and_then(Value::as_str),
        "owner": info.and_then(|v| v.get("owner")).and_then(Value::as_str),
        "amount": info.and_then(|v| v.pointer("/tokenAmount/amount")).and_then(Value::as_str),
        "uiAmountString": info.and_then(|v| v.pointer("/tokenAmount/uiAmountString")).and_then(Value::as_str),
    })
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn cors_headers_allow_browser_sdk_reads() {
        let response = with_cors(
            Response::builder()
                .status(StatusCode::OK)
                .body(Full::new(Bytes::new()))
                .unwrap(),
        );

        assert_eq!(response.headers()[ACCESS_CONTROL_ALLOW_ORIGIN], "*");
        assert_eq!(
            response.headers()[ACCESS_CONTROL_ALLOW_METHODS],
            "GET, POST, OPTIONS"
        );
        assert_eq!(
            response.headers()[ACCESS_CONTROL_ALLOW_HEADERS],
            "Authorization, Content-Type"
        );
    }

    #[test]
    fn native_balance_serializes_u64_values_as_decimal_strings() {
        let value = contextual_native_balance_json(&json!({
            "context": { "slot": 9_007_199_254_740_995_u64 },
            "value": 9_007_199_254_740_993_u64,
        }))
        .unwrap();

        assert_eq!(value["lamports"], "9007199254740993");
        assert_eq!(value["contextSlot"], "9007199254740995");
    }

    /// A missing account must hold its slot, or every later address is attributed to the
    /// wrong account — silently, and with real money downstream.
    #[test]
    fn batch_accounts_keep_position_when_an_account_is_absent() {
        let addresses = vec!["a".to_string(), "b".to_string(), "c".to_string()];
        let values = vec![
            json!({ "owner": "prog", "lamports": 7u64, "executable": false, "data": ["AQI=", "base64"] }),
            Value::Null,
            json!({ "owner": "prog", "lamports": 9u64, "executable": false, "data": ["AwQ=", "base64"] }),
        ];

        let items = batch_accounts_json(&addresses, &values);

        assert_eq!(items.len(), 3);
        assert_eq!(items[0]["address"], "a");
        assert_eq!(items[0]["lamports"], "7");
        assert_eq!(items[0]["data"], "AQI=");
        assert_eq!(items[1], Value::Null, "absent account keeps its slot");
        assert_eq!(items[2]["address"], "c");
        assert_eq!(items[2]["lamports"], "9");
    }

    /// The batch is where a rounded balance would be acted on, so the wire must carry a `u64`
    /// no JavaScript client can round. 9007199254740993 is the first integer a JSON number
    /// cannot represent: `JSON.parse` returns ...92.
    #[test]
    fn batch_accounts_serialize_lamports_as_decimal_strings() {
        let addresses = vec!["big".to_string()];
        let values = vec![json!({
            "owner": "prog",
            "lamports": 9_007_199_254_740_993_u64,
            "executable": false,
            "data": ["AQI=", "base64"],
        })];

        let items = batch_accounts_json(&addresses, &values);

        assert_eq!(items[0]["lamports"], "9007199254740993");
    }

    fn auth_context_allowing(
        max_batch_addresses: Option<u32>,
    ) -> crate::websocket::auth::AuthContext {
        crate::websocket::auth::AuthContext {
            subject: "test".to_string(),
            issuer: "test-issuer".to_string(),
            audience: "test-audience".to_string(),
            key_class: arete_auth::KeyClass::Publishable,
            metering_key: "meter-test".to_string(),
            deployment_id: None,
            target_kind: None,
            target_id: None,
            program_id: None,
            program_release_hash: None,
            expires_at: u64::MAX,
            scope: "read".to_string(),
            limits: arete_auth::Limits {
                max_http_batch_addresses: max_batch_addresses,
                ..Default::default()
            },
            plan: None,
            origin: None,
            client_ip: None,
            jti: "test-jti".to_string(),
            actor_key: None,
            account_key: None,
            consumer_key: None,
            policy_version: None,
            account_limits: arete_auth::Limits::default(),
            row_scopes: None,
        }
    }

    /// A batch is charged as one HTTP request, so a session's address ceiling is the only thing
    /// standing between a ten-address plan and a hundred-address read. `POST /chain/accounts`
    /// ignored it while the program-account batch honoured it; both now share this computation.
    #[test]
    fn a_session_ceiling_bounds_the_batch_below_the_protocol_limit() {
        let context = auth_context_allowing(Some(10));

        assert_eq!(
            batch_address_limit(Some(&context), MAX_CHAIN_BATCH_ADDRESSES),
            10
        );
    }

    /// A generous claim must not raise the protocol ceiling: one batch is still one upstream call.
    #[test]
    fn a_session_ceiling_cannot_exceed_the_protocol_limit() {
        let context = auth_context_allowing(Some(1_000));

        assert_eq!(
            batch_address_limit(Some(&context), MAX_CHAIN_BATCH_ADDRESSES),
            MAX_CHAIN_BATCH_ADDRESSES
        );
    }

    /// No claim, and no auth at all, both fall back to the protocol ceiling.
    #[test]
    fn an_absent_ceiling_falls_back_to_the_protocol_limit() {
        let context = auth_context_allowing(None);

        assert_eq!(
            batch_address_limit(Some(&context), MAX_CHAIN_BATCH_ADDRESSES),
            MAX_CHAIN_BATCH_ADDRESSES
        );
        assert_eq!(
            batch_address_limit(None, MAX_PROGRAM_BATCH_ADDRESSES),
            MAX_PROGRAM_BATCH_ADDRESSES
        );
    }

    #[test]
    fn batch_accounts_json_is_empty_for_no_addresses() {
        assert!(batch_accounts_json(&[], &[]).is_empty());
    }

    /// The batch item shape must stay identical to the single-address route's, or a client
    /// needs two parsers for one concept.
    #[test]
    fn batch_item_matches_the_single_address_shape() {
        let value = json!({ "owner": "prog", "lamports": 5u64, "executable": true, "data": ["BQY=", "base64"] });
        let addresses = vec!["solo".to_string()];

        assert_eq!(
            batch_accounts_json(&addresses, std::slice::from_ref(&value))[0],
            raw_account_json("solo", &value)
        );
    }

    #[test]
    fn rpc_read_config_propagates_min_context_slot() {
        let config = rpc_read_config(Some("jsonParsed"), Some(9_007_199_254_740_997));

        assert_eq!(config["commitment"], "confirmed");
        assert_eq!(config["encoding"], "jsonParsed");
        assert_eq!(config["minContextSlot"], 9_007_199_254_740_997_u64);
    }

    #[test]
    fn token_balance_preserves_raw_amount_and_stringifies_context_slot() {
        let value = contextual_token_balance_json(
            &json!({
                "context": { "slot": 9_007_199_254_740_995_u64 },
                "value": [{
                    "pubkey": "token-account",
                    "account": {
                        "data": {
                            "parsed": {
                                "info": {
                                    "mint": "mint",
                                    "tokenAmount": {
                                        "amount": "18446744073709551615",
                                        "decimals": 9,
                                        "uiAmountString": "18446744073.709551615"
                                    }
                                }
                            }
                        }
                    }
                }]
            }),
            "owner",
            "mint",
            None,
        )
        .unwrap();

        assert_eq!(value["amount"], "18446744073709551615");
        assert_eq!(value["contextSlot"], "9007199254740995");
    }

    #[test]
    fn balance_bodies_accept_decimal_string_min_context_slots() {
        let native: NativeBalanceBody = serde_json::from_value(json!({
            "address": "owner",
            "minContextSlot": "9007199254740997",
        }))
        .unwrap();
        let token: BalanceBody = serde_json::from_value(json!({
            "owner": "owner",
            "mint": "mint",
            "minContextSlot": "9007199254740999",
        }))
        .unwrap();

        assert_eq!(
            parse_min_context_slot(native.min_context_slot.as_deref()).unwrap(),
            Some(9_007_199_254_740_997)
        );
        assert_eq!(
            parse_min_context_slot(token.min_context_slot.as_deref()).unwrap(),
            Some(9_007_199_254_740_999)
        );
    }

    fn runtime_definition(reader: crate::ProgramAccountReaderFn) -> ProgramRuntimeDefinition {
        let program_spec_hash = crate::ProgramSpecHash::from_digest([1; 32]);
        let idl_content_hash = crate::IdlContentHash::from_digest([2; 32]);
        let normalized_idl_hash = crate::NormalizedIdlHash::from_digest([3; 32]);
        let program_release_hash = arete_hash::OssGeneratedProgramReleaseV1::new(
            "Program111",
            program_spec_hash,
            idl_content_hash,
            normalized_idl_hash,
        )
        .hash()
        .unwrap();
        ProgramRuntimeDefinition {
            program_id: "Program111".to_string(),
            program_spec_hash,
            idl_content_hash,
            normalized_idl_hash,
            program_release_hash,
            account_reader: reader,
        }
    }

    fn program_read_context(
        target_id: &str,
        program_id: &str,
        release_hash: &str,
    ) -> crate::websocket::auth::AuthContext {
        crate::websocket::auth::AuthContext::from_claims(
            arete_auth::SessionClaims::program_read_builder(
                "issuer",
                "user:1",
                target_id,
                program_id,
                release_hash,
            )
            .build(),
        )
    }

    #[test]
    fn program_read_auth_is_bound_to_target_program_and_release() {
        let definition = runtime_definition(Arc::new(|_, _| Ok(Value::Null)));
        let release_hash = definition.program_release_hash.to_string();
        let valid = program_read_context("binding-1", &definition.program_id, &release_hash);

        assert_eq!(
            authorize_program_read(Some(&valid), Some("binding-1"), &definition),
            Ok(())
        );

        let wrong_target = program_read_context("binding-2", &definition.program_id, &release_hash);
        let wrong_program = program_read_context("binding-1", "Program222", &release_hash);
        let wrong_release = program_read_context(
            "binding-1",
            &definition.program_id,
            "arete:h1:program-release:sha256:different",
        );

        for context in [&wrong_target, &wrong_program, &wrong_release] {
            assert_eq!(
                authorize_program_read(Some(context), Some("binding-1"), &definition),
                Err(ProgramReadError::Unauthorized)
            );
        }
        assert_eq!(
            authorize_program_read(Some(&valid), None, &definition),
            Err(ProgramReadError::AuthorizationNotConfigured)
        );
        assert_eq!(authorize_program_read(None, None, &definition), Ok(()));
    }

    #[test]
    fn release_routes_are_exact_and_legacy_program_routes_are_not_accepted() {
        let hash = crate::ProgramReleaseHash::from_digest([4; 32]);
        let fetch_path = format!("/v1/releases/{hash}/accounts/Vault/address");
        let exists_path = format!("{fetch_path}/exists");
        let batch_path = format!("/v1/releases/{hash}/accounts/Vault");

        assert!(matches!(
            ProgramReadRoute::parse(&Method::GET, &fetch_path)
                .unwrap()
                .operation,
            ProgramReadOperation::Fetch { .. }
        ));
        assert!(matches!(
            ProgramReadRoute::parse(&Method::GET, &exists_path)
                .unwrap()
                .operation,
            ProgramReadOperation::Exists { .. }
        ));
        assert!(matches!(
            ProgramReadRoute::parse(&Method::POST, &batch_path)
                .unwrap()
                .operation,
            ProgramReadOperation::Batch
        ));
        assert_eq!(
            ProgramReadRoute::parse(&Method::GET, "/programs/demo/accounts/Vault/address")
                .unwrap_err(),
            ProgramReadError::NotFound
        );
    }

    #[tokio::test]
    async fn owner_mismatch_is_rejected_before_decoder_execution() {
        use std::sync::atomic::{AtomicBool, Ordering};

        let called = Arc::new(AtomicBool::new(false));
        let called_by_reader = called.clone();
        let definition = runtime_definition(Arc::new(move |_, _| {
            called_by_reader.store(true, Ordering::SeqCst);
            Ok(json!({ "decoded": true }))
        }));
        let value = json!({
            "owner": "DifferentProgram",
            "data": [BASE64_STANDARD.encode([1, 2, 3]), "base64"]
        });

        assert!(matches!(
            decode_release_account(&definition, "Vault", &value).await,
            AccountReadOutcome::Error(ProgramReadError::AccountOwnerMismatch)
        ));
        assert!(!called.load(Ordering::SeqCst));
    }

    #[tokio::test]
    async fn decode_failures_remain_typed_errors_and_metadata_is_public_only() {
        let definition = runtime_definition(Arc::new(|_, _| anyhow::bail!("private diagnostic")));
        let value = json!({
            "owner": "Program111",
            "data": [BASE64_STANDARD.encode([1, 2, 3]), "base64"]
        });
        assert!(matches!(
            decode_release_account(&definition, "Vault", &value).await,
            AccountReadOutcome::Error(ProgramReadError::AccountDecodeFailed)
        ));

        let response = with_release_metadata(
            program_read_error_response(ProgramReadError::AccountDecodeFailed),
            &definition,
            Some("address"),
            Some(true),
        );
        assert_eq!(response.headers()["X-Error-Code"], "ACCOUNT_DECODE_FAILED");
        assert_eq!(
            response.headers()["X-Arete-Program-Release-Hash"],
            definition.program_release_hash.to_string()
        );
        assert_eq!(
            response.headers()["X-Arete-Idl-Content-Hash"],
            definition.idl_content_hash.to_string()
        );
        let body = response.into_body().collect().await.unwrap().to_bytes();
        let body = std::str::from_utf8(&body).unwrap();
        assert_eq!(body, r#"{"error":{"code":"ACCOUNT_DECODE_FAILED"}}"#);
        assert!(!body.contains("private diagnostic"));
        assert!(!body.contains("decoder"));
    }

    /// Maps a bearer token to a gateway session carrying exactly the scopes it names.
    struct ScopeTokenPlugin;

    #[async_trait::async_trait]
    impl WebSocketAuthPlugin for ScopeTokenPlugin {
        async fn authorize(&self, request: &ConnectionAuthRequest) -> AuthDecision {
            let Some(scope) = request.bearer_token() else {
                return AuthDecision::Deny(AuthDeny::token_missing());
            };
            let mut claims =
                arete_auth::SessionClaims::solana_gateway_builder("issuer", "user:1", "gateway-1")
                    .with_scope(scope.replace('+', " "))
                    .with_metering_key("account:1")
                    .build();
            if scope.contains("limited") {
                claims.scope = SCOPE_FORK_READ.to_string();
                claims.limits = Some(arete_auth::Limits {
                    max_http_requests_per_minute: Some(3),
                    ..Default::default()
                });
            }
            if scope.contains("narrow") {
                claims.scope = SCOPE_FORK_READ.to_string();
                claims.limits = Some(arete_auth::Limits {
                    max_http_batch_addresses: Some(2),
                    ..Default::default()
                });
            }
            AuthDecision::Allow(crate::websocket::auth::AuthContext::from_claims(claims))
        }

        fn as_any(&self) -> &dyn std::any::Any {
            self
        }
    }

    async fn start_gateway(rpc_url: &str) -> (String, CancellationToken) {
        start_gateway_with_limits(rpc_url, HttpLimitsConfig::default()).await
    }

    async fn start_gateway_with_limits(
        rpc_url: &str,
        limits: HttpLimitsConfig,
    ) -> (String, CancellationToken) {
        let port = std::net::TcpListener::bind("127.0.0.1:0")
            .unwrap()
            .local_addr()
            .unwrap()
            .port();
        let addr: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap();
        let shutdown = CancellationToken::new();
        let server = HttpHealthServer::new(addr)
            .with_runtime_plan(RuntimePlan::solana_gateway())
            .with_auth_plugin(Arc::new(ScopeTokenPlugin))
            .with_solana_gateway_target("gateway-1")
            .with_read_rpc_url(rpc_url)
            .with_limits(limits)
            .with_shutdown(shutdown.clone());
        tokio::spawn(server.start());
        for _ in 0..100 {
            if tokio::net::TcpStream::connect(addr).await.is_ok() {
                break;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        (format!("http://{addr}"), shutdown)
    }

    #[tokio::test]
    async fn the_fork_route_requires_the_exact_fork_read_scope() {
        let rpc = crate::http::fork::tests::mock_rpc(crate::http::fork::tests::result_responder(
            json!(77),
        ))
        .await;
        let (base, shutdown) = start_gateway(&rpc.url).await;
        let client = Client::new();
        let body = json!({ "jsonrpc": "2.0", "id": 1, "method": "getSlot" });
        let post = |token: Option<&str>| {
            let mut request = client.post(format!("{base}/rpc/v1/fork")).json(&body);
            if let Some(token) = token {
                request = request.bearer_auth(token);
            }
            request.send()
        };

        let missing = post(None).await.unwrap();
        assert_eq!(missing.status(), StatusCode::UNAUTHORIZED);

        for scope in [
            "read",
            "read+transaction:inspect+transaction:send",
            "fork:read-extra",
        ] {
            let denied = post(Some(scope)).await.unwrap();
            assert_eq!(denied.status(), StatusCode::FORBIDDEN, "{scope}");
            let value: Value = denied.json().await.unwrap();
            assert_eq!(value["code"], "insufficient_scope");
        }
        assert!(rpc.requests.lock().unwrap().is_empty());

        let allowed = post(Some("fork:read")).await.unwrap();
        assert_eq!(allowed.status(), StatusCode::OK);
        let value: Value = allowed.json().await.unwrap();
        assert_eq!(value, json!({ "jsonrpc": "2.0", "id": 1, "result": 77 }));

        // fork:read does not open the chain-read routes.
        let chain = client
            .get(format!("{base}/chain/clock"))
            .bearer_auth("fork:read")
            .send()
            .await
            .unwrap();
        assert_eq!(chain.status(), StatusCode::FORBIDDEN);

        let get = client
            .get(format!("{base}/rpc/v1/fork"))
            .bearer_auth("fork:read")
            .send()
            .await
            .unwrap();
        assert_eq!(get.status(), StatusCode::METHOD_NOT_ALLOWED);
        shutdown.cancel();
    }

    #[tokio::test]
    async fn a_fork_batch_is_charged_per_entry() {
        let rpc = crate::http::fork::tests::mock_rpc(crate::http::fork::tests::result_responder(
            json!(1),
        ))
        .await;
        let (base, shutdown) = start_gateway(&rpc.url).await;
        let client = Client::new();
        let call = |id: u64| json!({ "jsonrpc": "2.0", "id": id, "method": "getSlot" });
        let batch = json!([call(1), call(2), call(3), call(4)]);

        let refused = client
            .post(format!("{base}/rpc/v1/fork"))
            .bearer_auth("limited")
            .json(&batch)
            .send()
            .await
            .unwrap();
        assert_eq!(refused.status(), StatusCode::TOO_MANY_REQUESTS);
        assert!(rpc.requests.lock().unwrap().is_empty());

        let within = client
            .post(format!("{base}/rpc/v1/fork"))
            .bearer_auth("limited")
            .json(&json!([call(1), call(2)]))
            .send()
            .await
            .unwrap();
        assert_eq!(within.status(), StatusCode::OK);
        assert_eq!(rpc.requests.lock().unwrap().len(), 2);
        shutdown.cancel();
    }

    /// The fork route's `getMultipleAccounts` honours the session's batch
    /// address ceiling, as the chain-read batch routes do: a session capped at
    /// two addresses cannot read a hundred in one call.
    #[tokio::test]
    async fn a_fork_multiple_accounts_call_is_bounded_by_the_session_ceiling() {
        let rpc = crate::http::fork::tests::mock_rpc(crate::http::fork::tests::result_responder(
            json!({ "context": { "slot": 1 }, "value": [null, null] }),
        ))
        .await;
        let (base, shutdown) = start_gateway(&rpc.url).await;
        let client = Client::new();
        let key = "11111111111111111111111111111111";
        let read = |count: usize| {
            json!({
                "jsonrpc": "2.0",
                "id": 1,
                "method": "getMultipleAccounts",
                "params": [vec![key; count]],
            })
        };

        let refused: Value = client
            .post(format!("{base}/rpc/v1/fork"))
            .bearer_auth("narrow")
            .json(&read(3))
            .send()
            .await
            .unwrap()
            .json()
            .await
            .unwrap();
        assert_eq!(refused["error"]["code"], -32602);
        assert!(rpc.requests.lock().unwrap().is_empty());

        let within: Value = client
            .post(format!("{base}/rpc/v1/fork"))
            .bearer_auth("narrow")
            .json(&read(2))
            .send()
            .await
            .unwrap()
            .json()
            .await
            .unwrap();
        assert!(within.get("result").is_some(), "{within}");
        assert_eq!(rpc.requests.lock().unwrap().len(), 1);
        shutdown.cancel();
    }

    /// A slow body must not hold a relay admission permit. With one inspect
    /// permit, a request trickling its body used to make every other inspect
    /// request a `429` until it finished; the body is now read, under the
    /// HTTP body deadline, before any admission is taken.
    #[tokio::test]
    async fn a_stalled_transaction_body_holds_no_admission_permit() {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        let port = std::net::TcpListener::bind("127.0.0.1:0")
            .unwrap()
            .local_addr()
            .unwrap()
            .port();
        let addr: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap();
        let shutdown = CancellationToken::new();
        let server = HttpHealthServer::new(addr)
            .with_transaction_config(TransactionConfig {
                enabled: true,
                rpc_url: Some("http://127.0.0.1:9".to_string()),
                allow_unauthenticated: true,
                inspect_concurrency: 1,
                send_concurrency: 1,
                ..TransactionConfig::default()
            })
            .with_limits(HttpLimitsConfig {
                body_read_timeout: Duration::from_millis(500),
                ..HttpLimitsConfig::default()
            })
            .with_shutdown(shutdown.clone());
        tokio::spawn(server.start());
        let mut stalled = loop {
            if let Ok(stream) = tokio::net::TcpStream::connect(addr).await {
                break stream;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        };
        stalled
            .write_all(
                b"POST /transactions/v1/fee HTTP/1.1\r\nHost: test\r\nContent-Type: application/json\r\nContent-Length: 64\r\n\r\n{",
            )
            .await
            .unwrap();
        tokio::time::sleep(Duration::from_millis(100)).await;

        let response = Client::new()
            .post(format!("http://{addr}/transactions/v1/fee"))
            .header("Content-Type", "application/json")
            .body("not json")
            .send()
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::BAD_REQUEST);

        let mut answer = Vec::new();
        tokio::time::timeout(Duration::from_secs(5), async {
            let mut buffer = [0; 1024];
            while !String::from_utf8_lossy(&answer).contains("\r\n\r\n") {
                let read = stalled.read(&mut buffer).await.unwrap();
                assert_ne!(read, 0, "the stalled request gets an answer");
                answer.extend_from_slice(&buffer[..read]);
            }
        })
        .await
        .expect("the stalled body is cut off at the deadline");
        assert!(
            String::from_utf8_lossy(&answer).starts_with("HTTP/1.1 408"),
            "{}",
            String::from_utf8_lossy(&answer)
        );
        shutdown.cancel();
    }

    /// Finished connections leave the server's connection set as they end,
    /// rather than all of them staying in it until shutdown.
    #[tokio::test]
    async fn finished_http_connections_are_reaped_while_serving() {
        use std::sync::atomic::{AtomicUsize, Ordering};

        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let shutdown = CancellationToken::new();
        let served = Arc::new(AtomicUsize::new(0));
        let accept = {
            let shutdown = shutdown.clone();
            let served = served.clone();
            tokio::spawn(async move {
                accept_connections(&listener, &shutdown, |_stream, _addr| {
                    let served = served.clone();
                    async move {
                        served.fetch_add(1, Ordering::SeqCst);
                    }
                })
                .await
            })
        };

        for _ in 0..20 {
            drop(tokio::net::TcpStream::connect(addr).await.unwrap());
        }
        tokio::time::timeout(Duration::from_secs(5), async {
            while served.load(Ordering::SeqCst) < 20 {
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("every connection is served");
        tokio::time::sleep(Duration::from_millis(50)).await;
        shutdown.cancel();

        let connections = accept.await.unwrap();
        assert_eq!(connections.len(), 0);
    }

    fn account_session(
        consumer: &str,
        account_limit: Option<u32>,
    ) -> crate::websocket::auth::AuthContext {
        crate::websocket::auth::AuthContext {
            subject: consumer.to_string(),
            metering_key: consumer.to_string(),
            limits: arete_auth::Limits {
                max_http_requests_per_minute: Some(100),
                ..Default::default()
            },
            actor_key: Some("user:1".to_string()),
            account_key: Some("account:1".to_string()),
            consumer_key: Some(consumer.to_string()),
            policy_version: Some(1),
            account_limits: arete_auth::Limits {
                max_http_requests_per_minute: account_limit,
                ..Default::default()
            },
            ..auth_context_allowing(None)
        }
    }

    /// Buckets of earlier minutes are dropped once a minute, so sessions and
    /// accounts that stop reading do not keep an entry for the server's life.
    #[test]
    fn http_rate_buckets_of_earlier_minutes_are_swept() {
        let limit_state = HttpLimitState::default();
        limit_state
            .per_subject_per_minute
            .insert("old-session".into(), (10, 3));
        limit_state
            .per_account_per_minute
            .insert("old-account".into(), (10, 3));
        limit_state
            .per_account_per_minute
            .insert("current-account".into(), (11, 1));

        limit_state.sweep(11);

        assert!(limit_state.per_subject_per_minute.is_empty());
        assert_eq!(limit_state.per_account_per_minute.len(), 1);
        assert!(limit_state
            .per_account_per_minute
            .contains_key("current-account"));
    }

    /// HTTP reads count against the account's signed aggregate quota, shared
    /// by every consumer and session of the account; a refusal charges
    /// nothing to the session's own bucket.
    #[test]
    fn http_reads_count_against_the_account_quota() {
        let limit_state = HttpLimitState::default();
        let first = account_session("consumer:a", Some(3));
        let second = account_session("consumer:b", Some(3));

        enforce_http_limits_weighted(&first, &limit_state, 2).unwrap();
        enforce_http_limits_weighted(&second, &limit_state, 1).unwrap();
        let deny = enforce_http_limits_weighted(&second, &limit_state, 1).unwrap_err();
        assert_eq!(
            deny.code,
            crate::websocket::auth::AuthErrorCode::RateLimitExceeded
        );
        let session = limit_state
            .per_subject_per_minute
            .get("consumer:b:consumer:b")
            .unwrap()
            .1;
        assert_eq!(session, 1, "the refused read is refunded");

        // No account claim: only the session's own limit applies.
        let unclaimed = HttpLimitState::default();
        let open = account_session("consumer:a", None);
        enforce_http_limits_weighted(&open, &unclaimed, 50).unwrap();
        assert!(unclaimed.per_account_per_minute.is_empty());
    }

    #[tokio::test]
    async fn chain_read_upstream_failures_never_reveal_the_rpc_url() {
        type Responder = crate::http::fork::tests::Responder;
        let leaky: [(&str, Responder); 3] = [
            (
                "http status",
                Arc::new(|_| (429, "rate limited at /secret-token".into())),
            ),
            (
                "json-rpc error",
                Arc::new(|_| {
                    (
                        200,
                        json!({
                            "jsonrpc": "2.0",
                            "id": "arete-read",
                            "error": { "code": -32005, "message": "see /secret-token" }
                        })
                        .to_string(),
                    )
                }),
            ),
            (
                "malformed body",
                Arc::new(|_| (200, "<html>/secret-token</html>".into())),
            ),
        ];
        let unreachable = {
            let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
            let url = format!("http://{}", listener.local_addr().unwrap());
            drop(listener);
            url
        };
        let mut upstreams = Vec::new();
        for (label, responder) in leaky {
            let rpc = crate::http::fork::tests::mock_rpc(responder).await;
            upstreams.push((label, rpc.url));
        }
        upstreams.push(("unreachable", unreachable));

        let client = Client::new();
        for (label, upstream) in upstreams {
            let rpc_url = format!("{upstream}/secret-token?api-key=hunter2");
            let port = upstream.rsplit(':').next().unwrap().to_string();
            let (base, shutdown) = start_gateway(&rpc_url).await;
            for request in [
                client.get(format!("{base}/chain/clock")),
                client.get(format!("{base}/chain/exists/{}", "1".repeat(32))),
                client
                    .post(format!("{base}/chain/v1/account"))
                    .json(&json!({ "address": "11111111111111111111111111111111" })),
            ] {
                let response = request.bearer_auth("read").send().await.unwrap();
                assert_eq!(response.status(), StatusCode::BAD_GATEWAY, "{label}");
                let correlation_id = response.headers()[CORRELATION_ID_HEADER]
                    .to_str()
                    .unwrap()
                    .to_string();
                let text = response.text().await.unwrap();
                for secret in ["secret-token", "hunter2", "api-key", "127.0.0.1", &port] {
                    assert!(!text.contains(secret), "{label}: {secret} in {text}");
                }
                let value: Value = serde_json::from_str(&text).unwrap();
                assert_eq!(value["error"], UPSTREAM_RPC_FAILED, "{label}");
                assert!(text.contains(&correlation_id), "{label}");
            }
            shutdown.cancel();
        }
    }

    fn small_limits() -> HttpLimitsConfig {
        HttpLimitsConfig {
            max_body_bytes: 1024,
            header_read_timeout: Duration::from_millis(300),
            body_read_timeout: Duration::from_millis(300),
            upstream_connect_timeout: Duration::from_millis(300),
            upstream_timeout: Duration::from_millis(300),
            max_upstream_response_bytes: 1024,
        }
    }

    #[tokio::test]
    async fn oversized_request_bodies_are_refused_with_413() {
        let rpc = crate::http::fork::tests::mock_rpc(crate::http::fork::tests::result_responder(
            json!(1),
        ))
        .await;
        let (base, shutdown) = start_gateway_with_limits(&rpc.url, small_limits()).await;
        let client = Client::new();
        let padding = " ".repeat(4096);
        let oversized = format!(r#"{{"address":"11111111111111111111111111111111"{padding}}}"#);
        for path in [
            "/chain/v1/account",
            "/chain/native-balance",
            "/chain/accounts",
        ] {
            // Declared length over the limit.
            let response = client
                .post(format!("{base}{path}"))
                .bearer_auth("read")
                .header("Content-Type", "application/json")
                .body(oversized.clone())
                .send()
                .await
                .unwrap();
            assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE, "{path}");

            // No declared length: refused once the streamed bytes pass the limit.
            let chunks = futures_util::stream::iter(
                oversized
                    .as_bytes()
                    .chunks(512)
                    .map(|chunk| Ok::<_, std::io::Error>(Bytes::copy_from_slice(chunk)))
                    .collect::<Vec<_>>(),
            );
            let response = client
                .post(format!("{base}{path}"))
                .bearer_auth("read")
                .body(reqwest::Body::wrap_stream(chunks))
                .send()
                .await
                .unwrap();
            assert_eq!(
                response.status(),
                StatusCode::PAYLOAD_TOO_LARGE,
                "{path} chunked"
            );
        }
        assert!(rpc.requests.lock().unwrap().is_empty());
        shutdown.cancel();
    }

    #[tokio::test]
    async fn a_body_that_stalls_is_answered_with_408() {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        let rpc = crate::http::fork::tests::mock_rpc(crate::http::fork::tests::result_responder(
            json!(1),
        ))
        .await;
        let (base, shutdown) = start_gateway_with_limits(&rpc.url, small_limits()).await;
        let addr = base.trim_start_matches("http://");
        let mut stream = tokio::net::TcpStream::connect(addr).await.unwrap();
        stream
            .write_all(
                format!(
                    "POST /chain/v1/account HTTP/1.1\r\nHost: {addr}\r\nAuthorization: Bearer read\r\nContent-Type: application/json\r\nContent-Length: 200\r\n\r\n{{\"addr"
                )
                .as_bytes(),
            )
            .await
            .unwrap();
        let mut response = vec![0; 1024];
        let read = tokio::time::timeout(Duration::from_secs(5), stream.read(&mut response))
            .await
            .expect("the server answers a stalled body before the test deadline")
            .unwrap();
        let response = String::from_utf8_lossy(&response[..read]);
        assert!(response.starts_with("HTTP/1.1 408"), "{response}");
        shutdown.cancel();
    }

    #[tokio::test]
    async fn a_client_that_never_finishes_its_headers_is_disconnected() {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        let (base, shutdown) =
            start_gateway_with_limits("http://127.0.0.1:9/unused", small_limits()).await;
        let addr = base.trim_start_matches("http://");
        let mut stream = tokio::net::TcpStream::connect(addr).await.unwrap();
        stream.write_all(b"GET /health HTTP/1.1\r\n").await.unwrap();
        let mut buffer = vec![0; 1024];
        let closed = tokio::time::timeout(Duration::from_secs(5), async {
            loop {
                match stream.read(&mut buffer).await {
                    Ok(0) | Err(_) => break,
                    Ok(_) => continue,
                }
            }
        })
        .await;
        assert!(
            closed.is_ok(),
            "the connection is closed at the header deadline"
        );
        shutdown.cancel();
    }

    #[tokio::test]
    async fn a_slow_upstream_times_out_as_a_502() {
        // Accepts connections and never answers them.
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let upstream = format!("http://{}", listener.local_addr().unwrap());
        tokio::spawn(async move {
            let mut held = Vec::new();
            while let Ok((stream, _)) = listener.accept().await {
                held.push(stream);
            }
        });
        let (base, shutdown) = start_gateway_with_limits(&upstream, small_limits()).await;
        let started = std::time::Instant::now();
        let response = Client::new()
            .get(format!("{base}/chain/clock"))
            .bearer_auth("read")
            .send()
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::BAD_GATEWAY);
        assert!(started.elapsed() < Duration::from_secs(5));
        let value: Value = response.json().await.unwrap();
        assert_eq!(value["error"], UPSTREAM_RPC_FAILED);
        shutdown.cancel();
    }

    #[tokio::test]
    async fn an_oversized_upstream_response_is_refused() {
        let rpc = crate::http::fork::tests::mock_rpc(crate::http::fork::tests::result_responder(
            json!("x".repeat(4096)),
        ))
        .await;
        let (base, shutdown) = start_gateway_with_limits(&rpc.url, small_limits()).await;
        let response = Client::new()
            .get(format!("{base}/chain/clock"))
            .bearer_auth("read")
            .send()
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::BAD_GATEWAY);
        let value: Value = response.json().await.unwrap();
        assert_eq!(value["error"], UPSTREAM_RPC_FAILED);
        shutdown.cancel();
    }
}