nautilus_lighter/http/

client.rs

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

//! Raw and domain HTTP clients for Lighter REST endpoints.

use std::{collections::HashMap, sync::Arc, time::Duration};

use jiff::Timestamp;
use nautilus_core::{
    AtomicTime, UnixNanos, string::secret::SecretString, time::get_atomic_clock_realtime,
};
use nautilus_model::{
    data::{Bar, BarType, FundingRateUpdate, OrderBookDeltas, TradeTick},
    identifiers::InstrumentId,
    instruments::{Instrument, InstrumentAny},
};
use nautilus_network::{
    http::{
        HttpClient, HttpRedirectPolicy, HttpResponse, Method, create_standard_nautilus_headers,
        parse_retry_after,
    },
    ratelimiter::quota::Quota,
    retry::{RetryManager, create_http_retry_manager},
};
use rust_decimal::Decimal;
use serde::{Serialize, de::DeserializeOwned};
use url::form_urlencoded;
use zeroize::Zeroizing;

use crate::{
    common::{
        enums::{
            LighterCandleResolution, LighterEnvironment, LighterFundingResolution,
            LighterMarketStatus,
        },
        rate_limit::{
            LIGHTER_REST_BUCKET, LIGHTER_REST_QUOTA, LighterTxRateLimiter, await_tx_quota,
        },
        symbol::MarketRegistry,
        urls::lighter_http_base_url,
    },
    http::{
        error::{
            LighterHttpError, LighterHttpResult, create_lighter_http_timeout_error,
            should_retry_lighter_http_error,
        },
        models::{
            LighterAccountDetail, LighterAccountsResponse, LighterCandle, LighterCandles,
            LighterFundings, LighterMakerOnlyApiKeys, LighterNextNonce, LighterOrderBookDetails,
            LighterOrderBookOrders, LighterOrderBooks, LighterOrders, LighterResultCode,
            LighterSendTxBatchRequest, LighterSendTxBatchResponse, LighterSendTxRequest,
            LighterSendTxResponse, LighterTrade, LighterTrades, LighterTx,
        },
        parse::{
            parse_candle_bar, parse_funding_rate_update,
            parse_order_book_details_instruments_with_status, parse_order_book_snapshot,
            parse_trade_tick, register_order_books,
        },
        query::{
            LighterAccountActiveOrdersQuery, LighterAccountInactiveOrdersQuery,
            LighterAccountLookup, LighterAccountQuery, LighterCandlesQuery, LighterFundingsQuery,
            LighterMakerOnlyApiKeysQuery, LighterNextNonceQuery, LighterOrderBookDetailsQuery,
            LighterOrderBookOrdersQuery, LighterOrderBooksQuery, LighterRecentTradesQuery,
            LighterTradesQuery, LighterTxLookup, LighterTxQuery,
        },
    },
};

const API_V1: &str = "/api/v1";
const ENDPOINT_ACCOUNT: &str = "/api/v1/account";
const ENDPOINT_ACCOUNT_ACTIVE_ORDERS: &str = "/api/v1/accountActiveOrders";
const ENDPOINT_ACCOUNT_INACTIVE_ORDERS: &str = "/api/v1/accountInactiveOrders";
const ENDPOINT_CANDLES: &str = "/api/v1/candles";
const ENDPOINT_FUNDINGS: &str = "/api/v1/fundings";
const ENDPOINT_MAKER_ONLY_API_KEYS: &str = "/api/v1/getMakerOnlyApiKeys";
const ENDPOINT_NEXT_NONCE: &str = "/api/v1/nextNonce";
const ENDPOINT_ORDER_BOOK_DETAILS: &str = "/api/v1/orderBookDetails";
const ENDPOINT_ORDER_BOOK_ORDERS: &str = "/api/v1/orderBookOrders";
const ENDPOINT_ORDER_BOOKS: &str = "/api/v1/orderBooks";
const ENDPOINT_RECENT_TRADES: &str = "/api/v1/recentTrades";
const ENDPOINT_REFERRAL_USE: &str = "/api/v1/referral/use";
const ENDPOINT_SEND_TX: &str = "/api/v1/sendTx";
const ENDPOINT_SEND_TX_BATCH: &str = "/api/v1/sendTxBatch";
const ENDPOINT_TRADES: &str = "/api/v1/trades";
const ENDPOINT_TX: &str = "/api/v1/tx";
const HEADER_AUTHORIZATION: &str = "authorization";
const HEADER_RETRY_AFTER: &str = "retry-after";
const MULTIPART_BOUNDARY: &str = "nautilus-lighter-form-boundary";

/// Maximum page size accepted by Lighter REST list endpoints (`/api/v1/trades`,
/// `/api/v1/accountInactiveOrders`). Values above this trigger `20001 invalid
/// param` from the venue, so reconciliation paginates at this cap and follows
/// `next_cursor` until the response is empty.
pub const LIGHTER_REST_PAGE_SIZE: u16 = 100;
pub const LIGHTER_CANDLES_MAX_LIMIT: u16 = 500;

/// Maximum rows returned per `/api/v1/fundings` call (the venue per-call cap).
pub const LIGHTER_FUNDINGS_MAX_LIMIT: u16 = 100;

const DEFAULT_BARS_LIMIT: usize = LIGHTER_CANDLES_MAX_LIMIT as usize;
const DEFAULT_FUNDING_RATES_LIMIT: usize = 100;
const MAX_BAR_REQUEST_PAGES: usize = 500;
const MAX_FUNDING_REQUEST_PAGES: usize = 500;

trait LighterResponseCheck {
    fn response_code(&self) -> i32;
    fn response_message(&self) -> Option<&str>;
}

macro_rules! impl_lighter_response_check {
    ($($ty:ty),+ $(,)?) => {
        $(
            impl LighterResponseCheck for $ty {
                fn response_code(&self) -> i32 {
                    self.code
                }

                fn response_message(&self) -> Option<&str> {
                    self.message.as_deref()
                }
            }
        )+
    };
}

impl_lighter_response_check!(
    LighterAccountsResponse,
    LighterCandles,
    LighterFundings,
    LighterMakerOnlyApiKeys,
    LighterNextNonce,
    LighterOrderBookDetails,
    LighterOrderBookOrders,
    LighterOrderBooks,
    LighterOrders,
    LighterResultCode,
    LighterSendTxBatchResponse,
    LighterSendTxResponse,
    LighterTrades,
    LighterTx,
);

/// Raw HTTP client for Lighter REST API operations.
///
/// This client owns the transport, base URL, default headers, and rate limit. Methods map directly
/// to venue endpoints and return venue response models without converting to Nautilus domain types.
#[derive(Clone, Debug)]
pub struct LighterRawHttpClient {
    base_url: String,
    environment: LighterEnvironment,
    client: HttpClient,
    retry_manager: RetryManager<LighterHttpError>,
    tx_rate_limiter: Option<Arc<LighterTxRateLimiter>>,
}

impl Default for LighterRawHttpClient {
    fn default() -> Self {
        Self::new(LighterEnvironment::Mainnet, None, 60, None)
            .expect("failed to create default Lighter raw HTTP client")
    }
}

impl LighterRawHttpClient {
    /// Creates a new [`LighterRawHttpClient`].
    ///
    /// # Errors
    ///
    /// Returns an error if the underlying HTTP client cannot be created.
    pub fn new(
        environment: LighterEnvironment,
        base_url: Option<String>,
        timeout_secs: u64,
        proxy_url: Option<String>,
    ) -> LighterHttpResult<Self> {
        Self::new_with_quotas(
            environment,
            base_url,
            timeout_secs,
            proxy_url,
            *LIGHTER_REST_QUOTA,
            None,
        )
    }

    /// Creates a new [`LighterRawHttpClient`] with an explicit read quota and an
    /// optional shared transaction limiter.
    ///
    /// `default_quota` governs the REST read bucket ([`LIGHTER_REST_BUCKET`]),
    /// resolved from the client's configured `rest_quota_per_min` (detected tier
    /// only logs hints). `tx_rate_limiter` paces `sendTx` / `sendTxBatch`; the
    /// execution client shares one limiter across this and the WebSocket
    /// `sendTx` path so their combined rate honors the single venue tx bucket.
    /// The data client passes `None` (it sends no transactions).
    ///
    /// # Errors
    ///
    /// Returns an error if the underlying HTTP client cannot be created.
    pub fn new_with_quotas(
        environment: LighterEnvironment,
        base_url: Option<String>,
        timeout_secs: u64,
        proxy_url: Option<String>,
        default_quota: Quota,
        tx_rate_limiter: Option<Arc<LighterTxRateLimiter>>,
    ) -> LighterHttpResult<Self> {
        let base_url = base_url
            .unwrap_or_else(|| lighter_http_base_url(environment).to_string())
            .trim_end_matches('/')
            .to_string();

        Ok(Self {
            base_url,
            environment,
            client: HttpClient::builder()
                .redirect_policy(HttpRedirectPolicy::Reject)
                .header_keys(vec![HEADER_RETRY_AFTER.to_string()])
                .headers(Self::default_headers())
                .default_quota(default_quota)
                .timeout_secs(timeout_secs)
                .maybe_proxy_url(proxy_url)
                .build()?,
            retry_manager: create_http_retry_manager(),
            tx_rate_limiter,
        })
    }

    /// Returns the configured REST base URL.
    #[must_use]
    pub fn base_url(&self) -> &str {
        self.base_url.as_str()
    }

    /// Returns the configured Lighter environment.
    #[must_use]
    pub const fn environment(&self) -> LighterEnvironment {
        self.environment
    }

    /// Overrides the REST base URL. Intended for mock-server tests.
    pub fn set_base_url(&mut self, base_url: &str) {
        self.base_url = base_url.trim_end_matches('/').to_string();
    }

    /// Overrides the retry manager. Intended for mock-server tests that need
    /// shorter backoff than [`create_http_retry_manager`] produces.
    pub fn set_retry_manager(&mut self, retry_manager: RetryManager<LighterHttpError>) {
        self.retry_manager = retry_manager;
    }

    /// Calls `GET /api/v1/orderBooks`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_order_books(
        &self,
        query: &LighterOrderBooksQuery,
    ) -> LighterHttpResult<LighterOrderBooks> {
        self.send_get_request(ENDPOINT_ORDER_BOOKS, Some(query))
            .await
    }

    /// Calls `GET /api/v1/orderBookDetails`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_order_book_details(
        &self,
        query: &LighterOrderBookDetailsQuery,
    ) -> LighterHttpResult<LighterOrderBookDetails> {
        self.send_get_request(ENDPOINT_ORDER_BOOK_DETAILS, Some(query))
            .await
    }

    /// Calls `GET /api/v1/orderBookOrders`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_order_book_orders(
        &self,
        query: &LighterOrderBookOrdersQuery,
    ) -> LighterHttpResult<LighterOrderBookOrders> {
        self.send_get_request(ENDPOINT_ORDER_BOOK_ORDERS, Some(query))
            .await
    }

    /// Calls `GET /api/v1/recentTrades`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_recent_trades(
        &self,
        query: &LighterRecentTradesQuery,
    ) -> LighterHttpResult<LighterTrades> {
        self.send_get_request(ENDPOINT_RECENT_TRADES, Some(query))
            .await
    }

    /// Calls `GET /api/v1/trades`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_trades(&self, query: &LighterTradesQuery) -> LighterHttpResult<LighterTrades> {
        self.send_get_request(ENDPOINT_TRADES, Some(query)).await
    }

    /// Calls `GET /api/v1/candles`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_candles(
        &self,
        query: &LighterCandlesQuery,
    ) -> LighterHttpResult<LighterCandles> {
        self.send_get_request(ENDPOINT_CANDLES, Some(query)).await
    }

    /// Calls `GET /api/v1/fundings`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_fundings(
        &self,
        query: &LighterFundingsQuery,
    ) -> LighterHttpResult<LighterFundings> {
        self.send_get_request(ENDPOINT_FUNDINGS, Some(query)).await
    }

    /// Calls `GET /api/v1/account`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_account(
        &self,
        query: &LighterAccountQuery,
    ) -> LighterHttpResult<LighterAccountsResponse> {
        self.send_get_request(ENDPOINT_ACCOUNT, Some(query)).await
    }

    /// Calls `GET /api/v1/accountActiveOrders`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_account_active_orders(
        &self,
        query: &LighterAccountActiveOrdersQuery,
    ) -> LighterHttpResult<LighterOrders> {
        self.send_get_request(ENDPOINT_ACCOUNT_ACTIVE_ORDERS, Some(query))
            .await
    }

    /// Calls `GET /api/v1/accountInactiveOrders`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_account_inactive_orders(
        &self,
        query: &LighterAccountInactiveOrdersQuery,
    ) -> LighterHttpResult<LighterOrders> {
        self.send_get_request(ENDPOINT_ACCOUNT_INACTIVE_ORDERS, Some(query))
            .await
    }

    /// Calls `GET /api/v1/nextNonce`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_next_nonce(
        &self,
        query: &LighterNextNonceQuery,
    ) -> LighterHttpResult<LighterNextNonce> {
        self.send_get_request(ENDPOINT_NEXT_NONCE, Some(query))
            .await
    }

    /// Calls `GET /api/v1/tx`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_tx(&self, query: &LighterTxQuery) -> LighterHttpResult<LighterTx> {
        self.send_get_request(ENDPOINT_TX, Some(query)).await
    }

    /// Calls `GET /api/v1/getMakerOnlyApiKeys`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_maker_only_api_keys(
        &self,
        query: &LighterMakerOnlyApiKeysQuery,
    ) -> LighterHttpResult<LighterMakerOnlyApiKeys> {
        let params = LighterMakerOnlyApiKeysParams {
            account_index: query.account_index,
        };
        let headers = query
            .authorization
            .as_ref()
            .or(query.auth.as_ref())
            .map(|auth| {
                HashMap::from([(
                    HEADER_AUTHORIZATION.to_string(),
                    auth.expose_secret().to_owned(),
                )])
            });
        self.send_get_request_with_headers(ENDPOINT_MAKER_ONLY_API_KEYS, Some(&params), headers)
            .await
    }

    /// Calls `POST /api/v1/referral/use`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn use_referral(
        &self,
        l1_address: &str,
        referral_code: &str,
        auth_token: &str,
    ) -> LighterHttpResult<LighterResultCode> {
        let fields = [("l1_address", l1_address), ("referral_code", referral_code)];
        self.send_post_urlencoded(ENDPOINT_REFERRAL_USE, &fields, auth_token)
            .await
    }

    /// Calls `POST /api/v1/sendTx`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn send_tx(
        &self,
        request: &LighterSendTxRequest,
    ) -> LighterHttpResult<LighterSendTxResponse> {
        let fields = request.form_fields();
        self.send_post_form(ENDPOINT_SEND_TX, &fields).await
    }

    /// Calls `POST /api/v1/sendTxBatch`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn send_tx_batch(
        &self,
        request: &LighterSendTxBatchRequest,
    ) -> LighterHttpResult<LighterSendTxBatchResponse> {
        let fields = request.form_fields();
        self.send_post_form(ENDPOINT_SEND_TX_BATCH, &fields).await
    }

    async fn send_get_request<T, P>(
        &self,
        endpoint: &str,
        params: Option<&P>,
    ) -> LighterHttpResult<T>
    where
        T: DeserializeOwned + LighterResponseCheck,
        P: Serialize,
    {
        self.send_get_request_with_headers(endpoint, params, None)
            .await
    }

    async fn send_get_request_with_headers<T, P>(
        &self,
        endpoint: &str,
        params: Option<&P>,
        headers: Option<HashMap<String, String>>,
    ) -> LighterHttpResult<T>
    where
        T: DeserializeOwned + LighterResponseCheck,
        P: Serialize,
    {
        let url = self.url(endpoint);
        let rate_limit_keys = Self::rate_limit_keys(endpoint);
        self.retry_manager
            .invocation(
                endpoint,
                || {
                    let url = url.clone();
                    let rate_limit_keys = rate_limit_keys.clone();
                    let headers = headers.clone();

                    async move {
                        let response = self
                            .client
                            .request_with_params(
                                Method::GET,
                                url,
                                params,
                                headers,
                                None,
                                None,
                                Some(rate_limit_keys),
                            )
                            .await?;
                        Self::parse_response(&response)
                    }
                },
                should_retry_lighter_http_error,
                |e| create_lighter_http_timeout_error(e.to_string()),
            )
            .retry_delay(&LighterHttpError::retry_after)
            .execute()
            .await
    }

    // Single-shot: sendTx / sendTxBatch carry a signed nonce; transport-layer
    // retry could double-submit if the original landed and only the ack was lost.
    async fn send_post_form<T>(
        &self,
        endpoint: &str,
        fields: &[(&str, String)],
    ) -> LighterHttpResult<T>
    where
        T: DeserializeOwned + LighterResponseCheck,
    {
        // send_post_form only carries sendTx / sendTxBatch. With a shared tx
        // limiter (execution client) pace on it and pass no REST keys, so the
        // read quota does not also cap transactions. Without one (e.g. the
        // integrator-revoke utility) fall back to the internal REST limiter so
        // the request is still bounded rather than unthrottled.
        let rate_keys = match &self.tx_rate_limiter {
            Some(limiter) => {
                await_tx_quota(limiter).await;
                None
            }
            None => Some(Self::rate_limit_keys(endpoint)),
        };

        let response = self
            .client
            .request(
                Method::POST,
                self.url(endpoint),
                None,
                Some(multipart_headers()),
                Some(multipart_form_bytes(fields)),
                None,
                rate_keys,
            )
            .await?;

        Self::parse_response(&response)
    }

    // Single-shot because referral use changes account-level state and the API
    // does not document idempotency for a response lost after submission.
    async fn send_post_urlencoded<T>(
        &self,
        endpoint: &str,
        fields: &[(&str, &str)],
        auth_token: &str,
    ) -> LighterHttpResult<T>
    where
        T: DeserializeOwned + LighterResponseCheck,
    {
        let mut serializer = form_urlencoded::Serializer::new(String::new());
        serializer.extend_pairs(fields.iter().copied());
        let body = serializer.finish().into_bytes();

        let headers = HashMap::from([
            ("Accept".to_string(), "application/json".to_string()),
            (
                "Content-Type".to_string(),
                "application/x-www-form-urlencoded".to_string(),
            ),
            (HEADER_AUTHORIZATION.to_string(), auth_token.to_string()),
        ]);

        let response = self
            .client
            .request(
                Method::POST,
                self.url(endpoint),
                None,
                Some(headers),
                Some(body),
                None,
                Some(Self::rate_limit_keys(endpoint)),
            )
            .await?;

        Self::parse_response(&response)
    }

    fn parse_response<T>(response: &HttpResponse) -> LighterHttpResult<T>
    where
        T: DeserializeOwned + LighterResponseCheck,
    {
        let retry_after = response.headers.get(HEADER_RETRY_AFTER).and_then(|value| {
            let now = get_atomic_clock_realtime().get_time_ns().to_datetime_utc();
            parse_retry_after(value, now)
        });

        if !response.status.is_success() {
            let status = response.status.as_u16();
            let body = String::from_utf8_lossy(&response.body).to_string();

            // Status-first: a `{code,message}` body must not override the
            // retry decision for 5xx / 429.
            if status >= 500 {
                return Err(LighterHttpError::Http {
                    status,
                    body,
                    retry_after,
                });
            }

            // HTTP 405 is a Lighter rate-limit status like 429 per the docs, not a method error
            if status == 429 || status == 405 {
                return Err(LighterHttpError::RateLimit {
                    message: body,
                    retry_after,
                });
            }

            if let Ok(result) = serde_json::from_slice::<LighterResultCode>(&response.body)
                && result.code != 200
            {
                Self::result_code_error(
                    result.code,
                    result.message.as_deref(),
                    "HTTP error",
                    retry_after,
                )?;
            }

            return Err(LighterHttpError::Http {
                status,
                body,
                retry_after,
            });
        }

        let payload: T = match serde_json::from_slice(&response.body) {
            Ok(payload) => payload,
            Err(payload_error) => {
                if let Ok(result) = serde_json::from_slice::<LighterResultCode>(&response.body) {
                    Self::check_response(&result, retry_after)?;
                }
                return Err(payload_error.into());
            }
        };

        Self::check_response(&payload, retry_after)?;
        Ok(payload)
    }

    fn check_response<T: LighterResponseCheck>(
        payload: &T,
        retry_after: Option<Duration>,
    ) -> LighterHttpResult<()> {
        Self::result_code_error(
            payload.response_code(),
            payload.response_message(),
            "Lighter request failed",
            retry_after,
        )
    }

    fn result_code_error(
        code: i32,
        message: Option<&str>,
        default_message: &str,
        retry_after: Option<Duration>,
    ) -> LighterHttpResult<()> {
        match code {
            200 => Ok(()),
            405 | 429 => Err(LighterHttpError::RateLimit {
                message: message.unwrap_or("Lighter rate limit exceeded").to_string(),
                retry_after,
            }),
            code => Err(venue_error(code, message, default_message)),
        }
    }

    fn url(&self, endpoint: &str) -> String {
        format!("{}{}", self.base_url, endpoint)
    }

    fn rate_limit_keys(endpoint: &str) -> Vec<String> {
        let route = endpoint.strip_prefix(API_V1).unwrap_or(endpoint);
        vec![
            LIGHTER_REST_BUCKET.to_string(),
            format!("lighter:{}", route.trim_start_matches('/')),
        ]
    }

    fn default_headers() -> HashMap<String, String> {
        create_standard_nautilus_headers().into_iter().collect()
    }
}

#[derive(Serialize)]
struct LighterMakerOnlyApiKeysParams {
    account_index: i64,
}

fn multipart_headers() -> HashMap<String, String> {
    HashMap::from([
        ("Accept".to_string(), "application/json".to_string()),
        (
            "Content-Type".to_string(),
            format!("multipart/form-data; boundary={MULTIPART_BOUNDARY}"),
        ),
    ])
}

fn multipart_form_bytes(fields: &[(&str, String)]) -> Vec<u8> {
    let mut body = String::new();
    for (name, value) in fields {
        body.push_str("--");
        body.push_str(MULTIPART_BOUNDARY);
        body.push_str("\r\nContent-Disposition: form-data; name=\"");
        body.push_str(name);
        body.push_str("\"\r\n\r\n");
        body.push_str(value);
        body.push_str("\r\n");
    }
    body.push_str("--");
    body.push_str(MULTIPART_BOUNDARY);
    body.push_str("--\r\n");
    body.into_bytes()
}

/// Domain HTTP client for Lighter REST operations.
///
/// This client wraps [`LighterRawHttpClient`] and converts selected endpoint responses into
/// Nautilus domain data. Market metadata calls also populate the shared [`MarketRegistry`].
#[derive(Clone, Debug)]
pub struct LighterHttpClient {
    pub(crate) inner: Arc<LighterRawHttpClient>,
    market_registry: Arc<MarketRegistry>,
    clock: &'static AtomicTime,
}

impl Default for LighterHttpClient {
    fn default() -> Self {
        Self::new(LighterEnvironment::Mainnet, None, 60, None)
            .expect("failed to create default Lighter HTTP client")
    }
}

impl LighterHttpClient {
    /// Creates a new [`LighterHttpClient`].
    ///
    /// # Errors
    ///
    /// Returns an error if the underlying raw HTTP client cannot be created.
    pub fn new(
        environment: LighterEnvironment,
        base_url: Option<String>,
        timeout_secs: u64,
        proxy_url: Option<String>,
    ) -> LighterHttpResult<Self> {
        let raw_client = LighterRawHttpClient::new(environment, base_url, timeout_secs, proxy_url)?;
        Ok(Self::from_raw(raw_client))
    }

    /// Wraps an existing raw HTTP client.
    #[must_use]
    pub fn from_raw(raw_client: LighterRawHttpClient) -> Self {
        Self::from_raw_with_registry(raw_client, Arc::new(MarketRegistry::new()))
    }

    /// Wraps an existing raw HTTP client and shared market registry.
    #[must_use]
    pub fn from_raw_with_registry(
        raw_client: LighterRawHttpClient,
        market_registry: Arc<MarketRegistry>,
    ) -> Self {
        Self {
            inner: Arc::new(raw_client),
            market_registry,
            clock: get_atomic_clock_realtime(),
        }
    }

    /// Returns the configured REST base URL.
    #[must_use]
    pub fn base_url(&self) -> &str {
        self.inner.base_url()
    }

    /// Returns the configured Lighter environment.
    #[must_use]
    pub fn environment(&self) -> LighterEnvironment {
        self.inner.environment()
    }

    /// Returns the shared market registry used by this client.
    #[must_use]
    pub fn market_registry(&self) -> Arc<MarketRegistry> {
        self.market_registry.clone()
    }

    /// Overrides the REST base URL. Intended for mock-server tests.
    ///
    /// # Panics
    ///
    /// Panics if the raw client is shared by another [`Arc`].
    pub fn set_base_url(&mut self, base_url: &str) {
        Arc::get_mut(&mut self.inner)
            .expect("cannot override URL: raw client is shared")
            .set_base_url(base_url);
    }

    /// Calls `GET /api/v1/orderBooks` and registers returned markets.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_order_books(
        &self,
        query: &LighterOrderBooksQuery,
    ) -> LighterHttpResult<LighterOrderBooks> {
        let response = self.inner.get_order_books(query).await?;
        register_order_books(&self.market_registry, &response.order_books);
        Ok(response)
    }

    /// Calls `GET /api/v1/orderBookDetails` and registers returned markets.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_order_book_details(
        &self,
        query: &LighterOrderBookDetailsQuery,
    ) -> LighterHttpResult<LighterOrderBookDetails> {
        let response = self.inner.get_order_book_details(query).await?;
        parse_order_book_details_instruments_with_status(
            &self.market_registry,
            &response.order_book_details,
            &response.spot_order_book_details,
            self.generate_ts_init(),
        )?;
        Ok(response)
    }

    /// Calls `GET /api/v1/orderBookOrders`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_order_book_orders(
        &self,
        query: &LighterOrderBookOrdersQuery,
    ) -> LighterHttpResult<LighterOrderBookOrders> {
        self.inner.get_order_book_orders(query).await
    }

    /// Calls `GET /api/v1/recentTrades`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_recent_trades(
        &self,
        query: &LighterRecentTradesQuery,
    ) -> LighterHttpResult<LighterTrades> {
        self.inner.get_recent_trades(query).await
    }

    /// Calls `GET /api/v1/trades`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_trades(&self, query: &LighterTradesQuery) -> LighterHttpResult<LighterTrades> {
        self.inner.get_trades(query).await
    }

    /// Calls `GET /api/v1/candles`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_candles(
        &self,
        query: &LighterCandlesQuery,
    ) -> LighterHttpResult<LighterCandles> {
        self.inner.get_candles(query).await
    }

    /// Calls `GET /api/v1/fundings`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_fundings(
        &self,
        query: &LighterFundingsQuery,
    ) -> LighterHttpResult<LighterFundings> {
        self.inner.get_fundings(query).await
    }

    /// Fetches the account row for `account_index` via `GET /api/v1/account`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails, the response is invalid, or the
    /// venue returns no account for the index.
    pub async fn get_account_detail(
        &self,
        account_index: i64,
    ) -> LighterHttpResult<LighterAccountDetail> {
        let query = LighterAccountQuery {
            by: LighterAccountLookup::Index,
            value: account_index.to_string(),
        };
        self.inner
            .get_account(&query)
            .await?
            .accounts
            .into_iter()
            .next()
            .ok_or_else(|| {
                LighterHttpError::Parse(format!("no account returned for index {account_index}"))
            })
    }

    /// Calls `GET /api/v1/accountActiveOrders`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_account_active_orders(
        &self,
        query: &LighterAccountActiveOrdersQuery,
    ) -> LighterHttpResult<LighterOrders> {
        self.inner.get_account_active_orders(query).await
    }

    /// Calls `GET /api/v1/accountInactiveOrders`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_account_inactive_orders(
        &self,
        query: &LighterAccountInactiveOrdersQuery,
    ) -> LighterHttpResult<LighterOrders> {
        self.inner.get_account_inactive_orders(query).await
    }

    /// Calls `GET /api/v1/nextNonce` for `(account_index, api_key_index)`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_next_nonce(
        &self,
        account_index: i64,
        api_key_index: u8,
    ) -> LighterHttpResult<LighterNextNonce> {
        let query = LighterNextNonceQuery {
            account_index,
            api_key_index,
        };
        self.inner.get_next_nonce(&query).await
    }

    /// Calls `GET /api/v1/tx` for `tx_hash`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_tx(&self, tx_hash: impl Into<String>) -> LighterHttpResult<LighterTx> {
        self.inner
            .get_tx(&LighterTxQuery {
                by: LighterTxLookup::Hash,
                value: tx_hash.into(),
            })
            .await
    }

    /// Calls `GET /api/v1/getMakerOnlyApiKeys` for `account_index`.
    ///
    /// `auth_token` is the canonical Lighter auth string minted from the
    /// caller's credential.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn get_maker_only_api_keys(
        &self,
        account_index: i64,
        auth_token: impl Into<SecretString>,
    ) -> LighterHttpResult<LighterMakerOnlyApiKeys> {
        let query = Zeroizing::new(LighterMakerOnlyApiKeysQuery {
            authorization: Some(auth_token.into()),
            auth: None,
            account_index,
        });
        self.inner.get_maker_only_api_keys(&query).await
    }

    /// Applies `referral_code` to the L1 address.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn use_referral(
        &self,
        l1_address: &str,
        referral_code: &str,
        auth_token: &SecretString,
    ) -> LighterHttpResult<LighterResultCode> {
        self.inner
            .use_referral(l1_address, referral_code, auth_token.expose_secret())
            .await
    }

    /// Calls `POST /api/v1/sendTx`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn send_tx(
        &self,
        request: &LighterSendTxRequest,
    ) -> LighterHttpResult<LighterSendTxResponse> {
        self.inner.send_tx(request).await
    }

    /// Calls `POST /api/v1/sendTxBatch`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is invalid.
    pub async fn send_tx_batch(
        &self,
        request: &LighterSendTxBatchRequest,
    ) -> LighterHttpResult<LighterSendTxBatchResponse> {
        self.inner.send_tx_batch(request).await
    }

    /// Requests recent trades for an instrument and parses them into [`TradeTick`]s.
    ///
    /// # Errors
    ///
    /// Returns an error if the instrument has not been registered, the request fails, or a trade
    /// cannot be parsed.
    pub async fn request_recent_trades(
        &self,
        instrument: &InstrumentAny,
        limit: u16,
    ) -> LighterHttpResult<Vec<TradeTick>> {
        let market_id = self.market_index(instrument)?;
        let query = LighterRecentTradesQuery { market_id, limit };
        let response = self.inner.get_recent_trades(&query).await?;
        self.parse_trade_ticks(&response.trades, instrument)
    }

    /// Requests historical trades and parses them into [`TradeTick`]s.
    ///
    /// If `query.market_id` is `None`, the value is resolved from the shared market registry.
    ///
    /// # Errors
    ///
    /// Returns an error if the instrument has not been registered, the request fails, or a trade
    /// cannot be parsed.
    pub async fn request_trades(
        &self,
        instrument: &InstrumentAny,
        query: LighterTradesQuery,
    ) -> LighterHttpResult<Vec<TradeTick>> {
        let mut query = Zeroizing::new(query);
        if query.market_id.is_none() {
            query.market_id = Some(self.market_index(instrument)?);
        }
        let response = self.inner.get_trades(&query).await?;
        self.parse_trade_ticks(&response.trades, instrument)
    }

    /// Requests historical candles and parses them into Nautilus [`Bar`]s.
    ///
    /// # Errors
    ///
    /// Returns an error if the instrument has not been registered, the bar
    /// type is unsupported, the request fails, the page cap leaves part of the
    /// requested range uncovered, or a candle cannot be parsed.
    pub async fn request_bars(
        &self,
        instrument: &InstrumentAny,
        bar_type: BarType,
        start: Option<Timestamp>,
        end: Option<Timestamp>,
        limit: Option<u32>,
    ) -> LighterHttpResult<Vec<Bar>> {
        let market_id = self.market_index(instrument)?;
        let resolution = LighterCandleResolution::try_from(&bar_type)?;
        let interval_ms = resolution.interval_millis();
        let now = Timestamp::now();

        if let (Some(start), Some(end)) = (start, end)
            && start >= end
        {
            return Err(LighterHttpError::Parse(format!(
                "invalid bar request range: start={start}, end={end}",
            )));
        }

        let end = end.unwrap_or(now).min(now);

        if let Some(start) = start
            && start >= end
        {
            return Ok(Vec::new());
        }

        let requested_limit = limit.filter(|n| *n > 0).map(|n| n as usize);
        let target_limit = requested_limit.unwrap_or(DEFAULT_BARS_LIMIT);
        let start_was_unspecified = start.is_none();
        let end_ms = end.as_millisecond().max(0);
        let now_ms = now.as_millisecond();

        if end_ms == 0 {
            return Ok(Vec::new());
        }

        let start_ms = start.map_or_else(
            || {
                let lookback_bars = target_limit.saturating_add(1);
                let lookback_bars = i64::try_from(lookback_bars).unwrap_or(i64::MAX);
                let lookback_ms = interval_ms.saturating_mul(lookback_bars);
                end_ms.saturating_sub(lookback_ms)
            },
            |dt| dt.as_millisecond().max(0),
        );

        if start_ms >= end_ms {
            return Ok(Vec::new());
        }

        let mut bars = Vec::new();
        let mut cursor_ms = start_ms;
        let mut pages = 0_usize;
        let page_span_ms = interval_ms.saturating_mul(i64::from(LIGHTER_CANDLES_MAX_LIMIT));

        while cursor_ms < end_ms && pages < MAX_BAR_REQUEST_PAGES {
            if !start_was_unspecified
                && let Some(limit) = requested_limit
                && bars.len() >= limit
            {
                break;
            }

            let window_end_ms = cursor_ms.saturating_add(page_span_ms).min(end_ms);
            if window_end_ms <= cursor_ms {
                break;
            }

            let query = LighterCandlesQuery {
                market_id,
                resolution,
                start_timestamp: cursor_ms,
                end_timestamp: window_end_ms,
                count_back: i64::from(LIGHTER_CANDLES_MAX_LIMIT),
                set_timestamp_to_end: Some(false),
            };
            let response = self.get_candles(&query).await?;
            let mut page = self.parse_bars(&response.candles, instrument, bar_type)?;

            page.sort_by_key(|bar| bar.ts_event);
            for bar in page {
                let bar_start_ms = i64::try_from(bar.ts_event.as_u64() / 1_000_000)
                    .map_err(|e| LighterHttpError::Parse(e.to_string()))?;

                if bar_start_ms < cursor_ms
                    || bar_start_ms >= end_ms
                    || bar_start_ms.saturating_add(interval_ms) > now_ms
                {
                    continue;
                }

                if bars
                    .last()
                    .is_some_and(|last: &Bar| last.ts_event == bar.ts_event)
                {
                    continue;
                }
                bars.push(bar);

                if !start_was_unspecified
                    && let Some(limit) = requested_limit
                    && bars.len() >= limit
                {
                    break;
                }
            }

            cursor_ms = window_end_ms;
            pages += 1;
        }

        let limit_satisfied =
            !start_was_unspecified && requested_limit.is_some_and(|limit| bars.len() >= limit);
        if pages >= MAX_BAR_REQUEST_PAGES && cursor_ms < end_ms && !limit_satisfied {
            return Err(LighterHttpError::HistoryIncomplete {
                data_type: "bar",
                pages,
            });
        }

        if start_was_unspecified && bars.len() > target_limit {
            bars = bars.split_off(bars.len() - target_limit);
        }

        Ok(bars)
    }

    /// Requests historical funding rates and parses them into Nautilus updates.
    ///
    /// Lighter's public `/api/v1/fundings` endpoint returns settled funding
    /// rows at hourly or daily resolution. The adapter requests hourly rows and
    /// returns them in chronological order.
    ///
    /// # Errors
    ///
    /// Returns an error if the instrument is not a perpetual, the instrument
    /// has not been registered, the request range is invalid, the page cap
    /// leaves part of the requested range uncovered, or a row cannot be parsed.
    pub async fn request_funding_rates(
        &self,
        instrument: &InstrumentAny,
        start: Option<Timestamp>,
        end: Option<Timestamp>,
        limit: Option<usize>,
    ) -> LighterHttpResult<Vec<FundingRateUpdate>> {
        if !matches!(instrument, InstrumentAny::CryptoPerpetual(_)) {
            return Err(LighterHttpError::Parse(format!(
                "funding rates are only available for perpetual instruments: {}",
                instrument.id()
            )));
        }

        let market_id = self.market_index(instrument)?;
        let resolution = LighterFundingResolution::OneHour;
        let interval_ms = resolution.interval_millis();
        let now = Timestamp::now();

        if let (Some(start), Some(end)) = (start, end)
            && start >= end
        {
            return Err(LighterHttpError::Parse(format!(
                "invalid funding request range: start={start}, end={end}",
            )));
        }

        let end = end.unwrap_or(now).min(now);

        if let Some(start) = start
            && start >= end
        {
            return Ok(Vec::new());
        }

        let requested_limit = limit.filter(|n| *n > 0);
        let target_limit = requested_limit.unwrap_or(DEFAULT_FUNDING_RATES_LIMIT);
        let start_was_unspecified = start.is_none();
        let end_ms = end.as_millisecond().max(0);

        if end_ms == 0 {
            return Ok(Vec::new());
        }

        let start_ms = start.map_or_else(
            || {
                let lookback_rows = target_limit.saturating_add(1);
                let lookback_rows = i64::try_from(lookback_rows).unwrap_or(i64::MAX);
                let lookback_ms = interval_ms.saturating_mul(lookback_rows);
                end_ms.saturating_sub(lookback_ms)
            },
            |dt| dt.as_millisecond().max(0),
        );

        if start_ms >= end_ms {
            return Ok(Vec::new());
        }

        let ts_init = self.generate_ts_init();
        let interval = Some(resolution.interval_minutes());
        let mut funding_rates = Vec::new();
        let mut cursor_ms = start_ms;
        let mut pages = 0_usize;
        // `cap - 1`, not `cap`: the endpoint excludes `end_timestamp`, so a
        // full-cap span with `count_back = cap` would drop the row at the cursor.
        let page_span_ms =
            interval_ms.saturating_mul(i64::from(LIGHTER_FUNDINGS_MAX_LIMIT.saturating_sub(1)));

        while cursor_ms < end_ms && pages < MAX_FUNDING_REQUEST_PAGES {
            if !start_was_unspecified
                && let Some(limit) = requested_limit
                && funding_rates.len() >= limit
            {
                break;
            }

            let window_end_ms = cursor_ms.saturating_add(page_span_ms).min(end_ms);
            if window_end_ms <= cursor_ms {
                break;
            }

            let query = LighterFundingsQuery {
                market_id,
                resolution,
                start_timestamp: cursor_ms,
                end_timestamp: window_end_ms,
                count_back: i64::from(LIGHTER_FUNDINGS_MAX_LIMIT),
            };
            let response = self.get_fundings(&query).await?;

            let mut page = Vec::with_capacity(response.fundings.len());
            for funding in &response.fundings {
                let update = parse_funding_rate_update(funding, instrument.id(), interval, ts_init)
                    .map_err(LighterHttpError::from)?;
                let timestamp_ms = i64::try_from(update.ts_event.as_u64() / 1_000_000)
                    .map_err(|e| LighterHttpError::Parse(e.to_string()))?;

                if timestamp_ms < cursor_ms || timestamp_ms > end_ms {
                    continue;
                }
                page.push(update);
            }

            page.sort_by_key(|rate| rate.ts_event);
            for update in page {
                if funding_rates
                    .last()
                    .is_some_and(|last: &FundingRateUpdate| last.ts_event == update.ts_event)
                {
                    continue;
                }
                funding_rates.push(update);

                if !start_was_unspecified
                    && let Some(limit) = requested_limit
                    && funding_rates.len() >= limit
                {
                    break;
                }
            }

            cursor_ms = window_end_ms;
            pages += 1;
        }

        let limit_satisfied = !start_was_unspecified
            && requested_limit.is_some_and(|limit| funding_rates.len() >= limit);
        if pages >= MAX_FUNDING_REQUEST_PAGES && cursor_ms < end_ms && !limit_satisfied {
            return Err(LighterHttpError::HistoryIncomplete {
                data_type: "funding rate",
                pages,
            });
        }

        if start_was_unspecified && funding_rates.len() > target_limit {
            funding_rates = funding_rates.split_off(funding_rates.len() - target_limit);
        }

        Ok(funding_rates)
    }

    /// Requests all instruments from Lighter order book metadata.
    ///
    /// Lighter exposes market definitions through `GET /api/v1/orderBookDetails`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or nonempty metadata contains no valid instruments.
    pub async fn request_instruments(&self) -> LighterHttpResult<Vec<InstrumentAny>> {
        self.request_instruments_for_query(&LighterOrderBookDetailsQuery::default())
            .await
    }

    /// Requests all instruments and their market statuses from Lighter order book metadata.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or nonempty metadata contains no valid instruments.
    pub async fn request_instruments_with_status(
        &self,
    ) -> LighterHttpResult<Vec<(InstrumentAny, LighterMarketStatus)>> {
        self.request_instruments_with_status_for_query(&LighterOrderBookDetailsQuery::default())
            .await
    }

    /// Requests a single instrument from Lighter order book metadata.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails, the instrument is not found, or the instrument
    /// cannot be parsed.
    pub async fn request_instrument(
        &self,
        instrument_id: InstrumentId,
    ) -> LighterHttpResult<InstrumentAny> {
        self.request_instrument_with_status(instrument_id)
            .await
            .map(|(instrument, _)| instrument)
    }

    /// Requests a single instrument and its market status from Lighter order book metadata.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails, the instrument is not found, or the instrument
    /// cannot be parsed.
    pub async fn request_instrument_with_status(
        &self,
        instrument_id: InstrumentId,
    ) -> LighterHttpResult<(InstrumentAny, LighterMarketStatus)> {
        let query = LighterOrderBookDetailsQuery {
            market_id: self.market_registry.market_index(&instrument_id),
            filter: None,
        };
        let instruments = self
            .request_instruments_with_status_for_query(&query)
            .await?;

        instruments
            .into_iter()
            .find(|(instrument, _)| instrument.id() == instrument_id)
            .ok_or_else(|| LighterHttpError::Parse(format!("instrument {instrument_id} not found")))
    }

    /// Requests an HTTP order book snapshot and parses it into Nautilus order book deltas.
    ///
    /// # Errors
    ///
    /// Returns an error if the instrument has not been registered, the request fails, or any level
    /// cannot be parsed.
    pub async fn request_order_book_snapshot(
        &self,
        instrument: &InstrumentAny,
        limit: u16,
    ) -> LighterHttpResult<OrderBookDeltas> {
        let query = LighterOrderBookOrdersQuery {
            market_id: self.market_index(instrument)?,
            limit,
        };
        let snapshot = self.inner.get_order_book_orders(&query).await?;
        let ts_init = self.generate_ts_init();

        parse_order_book_snapshot(
            &snapshot,
            instrument.id(),
            instrument.price_precision(),
            instrument.size_precision(),
            ts_init,
            ts_init,
        )
        .map_err(LighterHttpError::from)
    }

    async fn request_instruments_for_query(
        &self,
        query: &LighterOrderBookDetailsQuery,
    ) -> LighterHttpResult<Vec<InstrumentAny>> {
        self.request_instruments_with_status_for_query(query)
            .await
            .map(|instruments| {
                instruments
                    .into_iter()
                    .map(|(instrument, _)| instrument)
                    .collect()
            })
    }

    async fn request_instruments_with_status_for_query(
        &self,
        query: &LighterOrderBookDetailsQuery,
    ) -> LighterHttpResult<Vec<(InstrumentAny, LighterMarketStatus)>> {
        let response = self.inner.get_order_book_details(query).await?;
        let ts_init = self.generate_ts_init();
        parse_order_book_details_instruments_with_status(
            &self.market_registry,
            &response.order_book_details,
            &response.spot_order_book_details,
            ts_init,
        )
        .map_err(LighterHttpError::from)
    }

    fn parse_trade_ticks(
        &self,
        trades: &[LighterTrade],
        instrument: &InstrumentAny,
    ) -> LighterHttpResult<Vec<TradeTick>> {
        let ts_init = self.generate_ts_init();
        trades
            .iter()
            .map(|trade| parse_trade_tick(trade, instrument, ts_init).map_err(Into::into))
            .collect()
    }

    fn parse_bars(
        &self,
        candles: &[LighterCandle],
        instrument: &InstrumentAny,
        bar_type: BarType,
    ) -> LighterHttpResult<Vec<Bar>> {
        let ts_init = self.generate_ts_init();
        candles
            .iter()
            .filter_map(|candle| {
                let has_positive_ohlc = candle.open > Decimal::ZERO
                    && candle.high > Decimal::ZERO
                    && candle.low > Decimal::ZERO
                    && candle.close > Decimal::ZERO;

                if !has_positive_ohlc {
                    log::warn!(
                        "Skipping Lighter candle at timestamp {} with non-positive OHLC values",
                        candle.timestamp,
                    );
                    return None;
                }

                Some(parse_candle_bar(candle, bar_type, instrument, ts_init).map_err(Into::into))
            })
            .collect()
    }

    fn market_index(&self, instrument: &InstrumentAny) -> LighterHttpResult<i64> {
        self.market_registry
            .market_index(&instrument.id())
            .ok_or_else(|| {
                LighterHttpError::Parse(format!(
                    "market index not registered for instrument {}",
                    instrument.id()
                ))
            })
    }

    fn generate_ts_init(&self) -> UnixNanos {
        self.clock.get_time_ns()
    }
}

fn venue_error(code: i32, message: Option<&str>, default_message: &str) -> LighterHttpError {
    LighterHttpError::Venue {
        code: i64::from(code),
        message: message.unwrap_or(default_message).to_string(),
    }
}

#[cfg(test)]
mod tests {
    use nautilus_testkit::http::assert_http_redirect_rejected;
    use rstest::rstest;

    use super::*;

    #[tokio::test]
    async fn test_authenticated_client_rejects_redirects() {
        let client = LighterRawHttpClient::new(LighterEnvironment::Testnet, None, 3, None)
            .unwrap()
            .client;
        assert_http_redirect_rejected(|url| async move {
            client
                .get(url, None, None, Some(3), None)
                .await
                .unwrap()
                .status
                .as_u16()
        })
        .await;
    }

    #[rstest]
    #[case(ENDPOINT_TRADES, "lighter:trades")]
    #[case(ENDPOINT_SEND_TX, "lighter:sendTx")]
    #[case(ENDPOINT_SEND_TX_BATCH, "lighter:sendTxBatch")]
    fn test_rate_limit_keys_uses_rest_bucket_and_route(
        #[case] endpoint: &str,
        #[case] route: &str,
    ) {
        // Transactions are paced by the shared tx limiter, not a separate HTTP
        // key, so every endpoint (reads and tx) keys uniformly on the rest bucket.
        assert_eq!(
            LighterRawHttpClient::rate_limit_keys(endpoint),
            vec![LIGHTER_REST_BUCKET.to_string(), route.to_string()],
        );
    }
}