vantage_api_client/rest/

api.rs

use std::sync::Arc;

use ciborium::Value as CborValue;
use indexmap::IndexMap;
use reqwest::header::{AUTHORIZATION, HeaderMap, HeaderName, HeaderValue};
use vantage_api_pool::resilient::{ResilientClient, TransportEvent, TransportObserver};
use vantage_core::{Priority, error};
use vantage_dataset::traits::Result;
use vantage_expressions::Expression;
use vantage_expressions::traits::expressive::ExpressiveEnum;
use vantage_table::pagination::Pagination;
use vantage_types::Record;

use crate::transport::AuthHeader;

/// How the API wraps its row array in the response body.
///
/// Most public APIs use one of these three shapes; the legacy vantage
/// "wrapped under `data`" shape is `Wrapped { array_key: "data" }`.
#[derive(Clone, Debug)]
pub enum ResponseShape {
    /// Body is a bare JSON array of records.
    /// Example: `GET /users` → `[ {…}, {…} ]`. JSONPlaceholder, GitHub, etc.
    BareArray,

    /// Body is a JSON object with the array under a fixed key.
    /// Example: `GET /users` → `{ "data": [ … ] }`.
    Wrapped { array_key: String },

    /// Body is a JSON object with the array under a key matching the
    /// table name. Example (DummyJSON):
    /// `GET /products` → `{ "products": [ … ], "total": …, "skip": …, "limit": … }`.
    WrappedByTableName,
}

impl Default for ResponseShape {
    /// Default matches the legacy 0.1.x shape: `{ "data": [...] }`.
    fn default() -> Self {
        ResponseShape::Wrapped {
            array_key: "data".to_string(),
        }
    }
}

/// Names of the page/limit query parameters the API expects.
///
/// Defaults to `("_page", "_limit")` — the JSON Server convention used
/// by JSONPlaceholder. DummyJSON uses `("skip", "limit")` (in items not
/// pages). Customise via `RestApiBuilder::pagination_params`.
#[derive(Clone, Debug)]
pub struct PaginationParams {
    pub page: String,
    pub limit: String,
    /// If true, the page parameter is sent as a *0-based item offset*
    /// (`skip`) instead of a 1-based page index. DummyJSON-style.
    pub skip_based: bool,
}

impl PaginationParams {
    pub fn page_limit(page: impl Into<String>, limit: impl Into<String>) -> Self {
        Self {
            page: page.into(),
            limit: limit.into(),
            skip_based: false,
        }
    }

    pub fn skip_limit(skip: impl Into<String>, limit: impl Into<String>) -> Self {
        Self {
            page: skip.into(),
            limit: limit.into(),
            skip_based: true,
        }
    }
}

impl Default for PaginationParams {
    fn default() -> Self {
        Self::page_limit("_page", "_limit")
    }
}

/// REST API backend for Vantage — reads data from HTTP JSON endpoints.
///
/// Each table maps to an API endpoint: `{base_url}/{table_name}`.
/// Response shape is configurable via [`RestApi::builder`]; see
/// [`ResponseShape`] for the supported variants.
///
/// Currently read-only — write operations return errors.
/// How a table's conditions are applied to a request.
///
/// URL `{placeholder}` path segments are always filled from matching
/// eq-conditions regardless of strategy; this governs what happens to
/// the *remaining* (non-path) eq-conditions.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum FilterStrategy {
    /// Append remaining eq-conditions as `?field=value` query params
    /// (JSON-Server semantics). The default.
    #[default]
    Query,
    /// Apply remaining eq-conditions as in-memory row filters after the
    /// fetch, never as query params. For APIs whose only server-side
    /// filters are path segments and that reject (or ignore) unknown
    /// query params — e.g. the Mercury control-API, whose CLI likewise
    /// filters version/env client-side after fetching by product path.
    Client,
}

/// How the API takes a sort: one query param naming the column, with a
/// prefix that flips it to descending (`?ordering=-net`, the Django REST
/// Framework and Launch Library convention). Configuring it makes the REST
/// vista orderable, so a paged grid asks the server for its sort instead of
/// re-sorting the rows it happens to hold.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OrderingParams {
    pub param: String,
    pub desc_prefix: String,
}

impl OrderingParams {
    pub fn new(param: impl Into<String>, desc_prefix: impl Into<String>) -> Self {
        Self {
            param: param.into(),
            desc_prefix: desc_prefix.into(),
        }
    }

    /// The value for `param`: the column, prefixed when descending.
    pub fn value(&self, field: &str, dir: vantage_vista::SortDirection) -> String {
        match dir {
            vantage_vista::SortDirection::Ascending => field.to_string(),
            vantage_vista::SortDirection::Descending => format!("{}{field}", self.desc_prefix),
        }
    }
}

/// A sort pushed down to the API: column and direction.
pub(crate) type Order<'a> = Option<(&'a str, vantage_vista::SortDirection)>;

/// One response's rows, see [`RestApi::fetch_page`].
struct Page {
    rows: IndexMap<String, Record<CborValue>>,
    total: Option<i64>,
    server_rows: usize,
    lossy: bool,
}

#[derive(Clone, Debug)]
pub struct RestApi {
    base_url: String,
    client: ResilientClient,
    pub(crate) auth_header: AuthHeader,
    response_shape: ResponseShape,
    pagination: PaginationParams,
    /// Paging params were configured explicitly, so windows can be fetched
    /// even with no `total_key` — the end of the set shows as a short page.
    paged: bool,
    /// When true, no `_page`/`_limit` query params are appended and
    /// list endpoints are assumed to return the full result set in
    /// one shot. Caller-side requests for page > 1 short-circuit to
    /// an empty result so a perpetual-grid stops paging after the
    /// first chunk. Useful for FastAPI/Pydantic services that treat
    /// unknown query params as strict filters.
    no_pagination: bool,
    /// How non-path eq-conditions are applied — query params vs.
    /// in-memory post-fetch filtering. See [`FilterStrategy`].
    filter_strategy: FilterStrategy,
    /// Response-envelope key carrying the grand total of matching rows
    /// (e.g. `count`). When set, the shell reports an exact count and
    /// advertises `can_fetch_window` for lazy/scroll loading; when `None`
    /// it falls back to counting fetched rows.
    total_key: Option<String>,
    /// The sort query param, when the API has one. Unset means the API
    /// cannot sort and consumers order fetched rows themselves.
    ordering: Option<OrderingParams>,
    /// Emit `tracing` events for window/count requests.
    debug: bool,
}

impl RestApi {
    /// Create a new REST API pointing at `base_url`. Uses the legacy
    /// default response shape (`{ "data": [...] }`). For other shapes
    /// (bare array, wrapped-by-table-name) use [`RestApi::builder`].
    pub fn new(base_url: impl Into<String>) -> Self {
        RestApi::builder(base_url).build()
    }

    /// Start configuring a [`RestApi`] via the builder.
    pub fn builder(base_url: impl Into<String>) -> RestApiBuilder {
        RestApiBuilder::new(base_url.into())
    }

    /// Set the Authorization header value (e.g. "Bearer `<token>`").
    /// Provided for backwards compatibility — prefer
    /// `RestApi::builder(...).auth(...)`.
    pub fn with_auth(mut self, auth: impl Into<String>) -> Self {
        self.auth_header = AuthHeader::new(auth);
        self
    }

    /// The configured response-envelope total key, if any. When set, the
    /// REST shell can report an exact count and serve `fetch_window`.
    pub fn total_key(&self) -> Option<&str> {
        self.total_key.as_deref()
    }

    /// Whether the shell can serve absolute-offset windows: the API pages
    /// (explicit `pagination_params`, or a `total_key`) and pagination isn't
    /// switched off. Without a total the loader learns the end of the set
    /// from a short page.
    pub fn serves_windows(&self) -> bool {
        !self.no_pagination && (self.paged || self.total_key.is_some())
    }

    /// The sort query param, when configured.
    pub fn ordering(&self) -> Option<&OrderingParams> {
        self.ordering.as_ref()
    }

    /// What the circuit breaker is doing right now. `None` when the
    /// underlying client has no breaker configured.
    pub fn breaker_state(&self) -> Option<crate::BreakerState> {
        self.client.breaker_state()
    }

    /// The resilient client backing this API — the pool, breaker and
    /// observer a caller outside the read path (e.g. an outbox replaying a
    /// queued write) should share rather than build its own.
    pub fn client(&self) -> &ResilientClient {
        &self.client
    }

    /// The configured base URL requests are joined against.
    pub fn base_url(&self) -> &str {
        &self.base_url
    }

    /// Issue an arbitrary HTTP request against `base_url`/`path`, through
    /// the same resilient client and auth header as reads.
    ///
    /// `headers` are applied after the configured `Authorization` header, so
    /// a caller-supplied header of the same name wins. `body`, when given,
    /// is sent as the JSON request body. The call policy follows
    /// [`vantage_core::Priority::current`], same as a table read. On success,
    /// a non-`GET` method reports [`TransportEvent::WritePushed`] to the
    /// observer.
    pub async fn http_request(
        &self,
        method: reqwest::Method,
        path: &str,
        headers: &[(&str, &str)],
        body: Option<&serde_json::Value>,
    ) -> vantage_core::Result<reqwest::Response> {
        let url = join_base_path(&self.base_url, path);

        // `HeaderMap::insert` replaces a same-named entry rather than
        // appending — unlike `RequestBuilder::header` — so a caller header
        // actually overrides the configured auth instead of riding alongside
        // it on the wire.
        let mut header_map = HeaderMap::new();
        if let Some(auth) = self.auth_header.value() {
            header_map.insert(
                AUTHORIZATION,
                HeaderValue::from_str(auth)
                    .expect("configured auth header must be a valid header value"),
            );
        }
        for (name, value) in headers {
            header_map.insert(
                HeaderName::from_bytes(name.as_bytes()).expect("header name must be valid"),
                HeaderValue::from_str(value).expect("header value must be valid"),
            );
        }

        let policy = crate::transport::policy_for(Priority::current());
        let response = self
            .client
            .execute_with(&policy, |http| {
                let req = http
                    .request(method.clone(), &url)
                    .headers(header_map.clone());
                match body {
                    Some(body) => req.json(body),
                    None => req,
                }
            })
            .await
            .map_err(|e| crate::transport::client_error(e, "API request failed", &url))?;

        if method != reqwest::Method::GET {
            self.client.report(TransportEvent::WritePushed);
        }

        Ok(response)
    }

    /// Build the endpoint path for `table_name`, substituting any
    /// `{placeholder}` segments from matching eq-conditions.
    ///
    /// Returns the absolute URL up to (but excluding) the query string,
    /// alongside the indices of conditions consumed by the substitution
    /// — those are dropped from the query string by `build_query_string`.
    ///
    /// Tables that don't use templates (no `{}` in the name) pass
    /// through unchanged and consume no conditions.
    fn endpoint_url(
        &self,
        table_name: &str,
        conditions: &[&Expression<CborValue>],
    ) -> Result<(String, Vec<usize>)> {
        let mut consumed = Vec::new();
        let mut path = String::with_capacity(table_name.len());
        let mut rest = table_name;
        while let Some(open) = rest.find('{') {
            path.push_str(&rest[..open]);
            let after = &rest[open + 1..];
            let close = after.find('}').ok_or_else(|| {
                error!(
                    "Unclosed `{` in table name URI template",
                    table_name = table_name
                )
            })?;
            let placeholder = &after[..close];
            let (idx, value) = conditions
                .iter()
                .enumerate()
                .find_map(|(i, cond)| {
                    if consumed.contains(&i) {
                        return None;
                    }
                    let (field, value) = crate::condition_to_query_param(cond)?;
                    (field == placeholder).then_some((i, value))
                })
                .ok_or_else(|| {
                    error!(
                        "No eq-condition provided for URI placeholder",
                        placeholder = placeholder,
                        table_name = table_name
                    )
                })?;
            consumed.push(idx);
            path.push_str(&urlencode(&value));
            rest = &after[close + 1..];
        }
        path.push_str(rest);
        Ok((format!("{}/{}", self.base_url, path), consumed))
    }

    /// Decide which conditions go in the query string and which are applied to
    /// the rows after they arrive.
    ///
    /// Under [`FilterStrategy::Client`] non-path eq-conditions are *not* sent —
    /// the API rejects or ignores unknown params — so they come back as
    /// client-side filters and every condition is marked consumed to keep it out
    /// of the URL. Otherwise nothing is filtered locally and only the path
    /// placeholders are consumed.
    ///
    /// Shared by the real fetch and by [`preview_request`](Self::preview_request)
    /// so a previewed URL cannot claim a filter the fetch would have applied in
    /// memory, or vice versa.
    fn split_filters(
        &self,
        conds: &[&Expression<CborValue>],
        consumed: Vec<usize>,
    ) -> (Vec<usize>, Vec<(String, String)>) {
        if self.filter_strategy == FilterStrategy::Client {
            let filters = conds
                .iter()
                .enumerate()
                .filter(|(i, _)| !consumed.contains(i))
                .filter_map(|(_, c)| crate::condition_to_query_param(c))
                .collect();
            ((0..conds.len()).collect(), filters)
        } else {
            (consumed, Vec::new())
        }
    }

    /// Build the combined query-string from pagination + conditions.
    /// `consumed` lists condition indices already baked into the URI
    /// path; those don't appear in the query string. Conditions that
    /// don't peel cleanly into eq pairs are skipped — same "best effort"
    /// stance as before.
    fn build_query_string(
        &self,
        window: Option<(i64, i64)>,
        conditions: &[&Expression<CborValue>],
        consumed: &[usize],
        order: Order<'_>,
    ) -> String {
        let mut params: Vec<(String, String)> = Vec::new();

        // Pagination first — matches the order users see in the URL bar.
        // When `no_pagination` is set the API doesn't accept page/limit
        // query params (and may treat them as strict filters that
        // return empty), so we leave them off.
        //
        // `window` is a half-open `[offset, offset+limit)` band. Skip-based
        // APIs take the offset verbatim; page-based APIs are addressed by
        // 1-based page, derived from the offset (the loader may hand
        // non-page-aligned windows, so it rounds down to the containing page).
        if !self.no_pagination
            && let Some((offset, limit)) = window
        {
            let offset = offset.max(0);
            let limit = limit.max(1);
            let page_value = if self.pagination.skip_based {
                offset.to_string()
            } else {
                (offset / limit + 1).to_string()
            };
            params.push((self.pagination.page.clone(), page_value));
            params.push((self.pagination.limit.clone(), limit.to_string()));
        }

        // A sort reaches the query only when the API declared how it takes
        // one; the vista never offers `add_order` otherwise.
        if let (Some(spec), Some((field, dir))) = (&self.ordering, order) {
            params.push((spec.param.clone(), spec.value(field, dir)));
        }

        // Conditions: each `eq` becomes `?field=value`. Multiple
        // conditions AND together (JSON Server semantics).
        for (i, cond) in conditions.iter().enumerate() {
            if consumed.contains(&i) {
                continue;
            }
            if let Some((field, value)) = crate::condition_to_query_param(cond) {
                params.push((field, value));
            }
        }

        if params.is_empty() {
            return String::new();
        }
        let mut s = String::from("?");
        for (i, (k, v)) in params.iter().enumerate() {
            if i > 0 {
                s.push('&');
            }
            // Minimal URL encoding — we encode `&` and `=` and spaces
            // because those break the query format. Anything else
            // passes through; the JSON Server convention is permissive.
            s.push_str(&urlencode(k));
            s.push('=');
            s.push_str(&urlencode(v));
        }
        s
    }

    /// Render the request a read would issue, without issuing it.
    ///
    /// Shares [`endpoint_url`](Self::endpoint_url) and
    /// [`build_query_string`](Self::build_query_string) with the real fetch
    /// path, so a previewed URL cannot drift from the one that gets sent.
    ///
    /// One difference is deliberate: `fetch_raw_body` first *awaits* any
    /// deferred condition (a foreign key whose value arrives with the parent
    /// row), and awaiting is what a preview must not do. Those are counted
    /// under `deferred_conditions` and left out of the URL instead.
    pub(crate) fn preview_request<'a>(
        &self,
        table_name: &str,
        window: Option<(i64, i64)>,
        order: Order<'_>,
        conditions: impl IntoIterator<Item = &'a Expression<CborValue>>,
    ) -> serde_json::Value {
        let conds: Vec<&Expression<CborValue>> = conditions.into_iter().collect();

        // Conditions the query-param lowering cannot peel into an eq pair —
        // deferred foreign keys, and anything else not shaped `field = value`.
        let unresolved = conds
            .iter()
            .filter(|c| crate::condition_to_query_param(c).is_none())
            .count();

        let (endpoint, consumed) = match self.endpoint_url(table_name, &conds) {
            Ok(pair) => pair,
            // A URI template placeholder went unfilled. If some condition is
            // still unresolved, the real fetch would have awaited it *before*
            // building the path, so this is a preview limitation and the
            // template is the honest answer. With nothing outstanding, the
            // fetch would fail here too — report that.
            Err(_) if unresolved > 0 => {
                return serde_json::json!({
                    "driver": "rest-api",
                    "method": "GET",
                    "url": format!("{}/{}", self.base_url, table_name),
                    "unresolved_conditions": unresolved,
                    "note": "path placeholders are filled from conditions resolved \
                             at fetch time; the template is shown unfilled",
                });
            }
            Err(e) => {
                return serde_json::json!({
                    "driver": "rest-api",
                    "base_url": self.base_url,
                    "error": e.to_string(),
                });
            }
        };

        let (query_consumed, client_filters) = self.split_filters(&conds, consumed);
        let query = self.build_query_string(window, &conds, &query_consumed, order);

        serde_json::json!({
            "driver": "rest-api",
            "method": "GET",
            "url": join_query(&endpoint, &query),
            "auth_header": self.auth_header.masked(),
            // Under `FilterStrategy::Client` these never reach the server: the
            // rows come back unfiltered and are narrowed in memory.
            "client_side_filters": client_filters
                .into_iter()
                .map(|(k, v)| format!("{k}={v}"))
                .collect::<Vec<_>>(),
            "unresolved_conditions": unresolved,
        })
    }

    /// Fetch data from the API endpoint and return parsed records.
    ///
    /// `id_field` selects which JSON field is treated as the record ID;
    /// if `None`, row indices are used. The page-based `pagination` is
    /// lowered to a `[offset, offset+limit)` window; `conditions` are
    /// pushed into the URL query string — eq-conditions become
    /// `?field=value`. Conditions that can't be peeled into a simple
    /// eq are silently skipped (caller-side filtering still applies if
    /// needed).
    pub(crate) async fn fetch_records<'a>(
        &self,
        table_name: &str,
        id_field: Option<&str>,
        pagination: Option<&Pagination>,
        conditions: impl IntoIterator<Item = &'a Expression<CborValue>>,
    ) -> Result<IndexMap<String, Record<CborValue>>> {
        let window = pagination.map(|p| (p.skip(), p.limit()));
        self.fetch_windowed(table_name, id_field, window, None, conditions)
            .await
            .map(|(records, _total)| records)
    }

    /// Fetch a single half-open row window `[offset, offset+limit)` — the
    /// primitive a paged, lazily-loaded grid drives on scroll (offset is
    /// an absolute row index, not a page number).
    /// Fetch one half-open row window, plus the envelope's `total_key` when
    /// the response carries one.
    ///
    /// The total comes out of the **same response as the rows**. Every paged
    /// endpoint reports it on every reply, so a caller wanting both a window
    /// and a grand total takes them together here rather than pairing a
    /// window fetch with [`Self::fetch_total`] — that pairing costs a second
    /// round trip for a number already in hand.
    pub(crate) async fn fetch_window_records_counted<'a>(
        &self,
        table_name: &str,
        id_field: Option<&str>,
        offset: i64,
        limit: i64,
        order: Order<'_>,
        conditions: impl IntoIterator<Item = &'a Expression<CborValue>>,
    ) -> Result<(IndexMap<String, Record<CborValue>>, Option<i64>)> {
        self.fetch_windowed(
            table_name,
            id_field,
            Some((offset, limit)),
            order,
            conditions,
        )
        .await
    }

    /// Read the grand total of matching rows from the response envelope's
    /// configured `total_key` (e.g. `count`). Returns `None` when no
    /// `total_key` is set — the caller then falls back to counting fetched
    /// rows. Issues a cheap `limit=1` request so the body carries the count
    /// without paying for the rows.
    pub(crate) async fn fetch_total<'a>(
        &self,
        table_name: &str,
        conditions: impl IntoIterator<Item = &'a Expression<CborValue>>,
    ) -> Result<Option<i64>> {
        let Some(total_key) = self.total_key.clone() else {
            return Ok(None);
        };
        let (body, _client_filters) = self
            .fetch_raw_body(table_name, Some((0, 1)), None, conditions)
            .await?;
        let total = body
            .get(total_key.as_str())
            .and_then(|v| v.as_i64())
            .ok_or_else(|| {
                error!(
                    "total_key missing or not an integer in API response",
                    total_key = total_key.as_str()
                )
            })?;
        if self.debug {
            tracing::debug!(target: "vantage_api_client::rest", total, "REST count");
        }
        Ok(Some(total))
    }

    /// Resolve conditions, build the windowed request URL, GET it (with the
    /// auth header if configured), and return the parsed JSON body together
    /// with any client-side filters that still need applying (under
    /// [`FilterStrategy::Client`]).
    async fn fetch_raw_body<'a>(
        &self,
        table_name: &str,
        window: Option<(i64, i64)>,
        order: Order<'_>,
        conditions: impl IntoIterator<Item = &'a Expression<CborValue>>,
    ) -> Result<(serde_json::Value, Vec<(String, String)>)> {
        // Conditions may carry `DeferredFn` values — typically from
        // `related_in_condition` for `with_one`-style traversals where the FK
        // lives in a parent record we haven't fetched yet. Resolve them once,
        // up front, so the rest of the pipeline sees only sync scalars.
        let raw: Vec<&Expression<CborValue>> = conditions.into_iter().collect();
        let mut resolved: Vec<Expression<CborValue>> = Vec::with_capacity(raw.len());
        for cond in raw {
            resolved.push(resolve_deferreds(cond.clone()).await?);
        }
        let conds: Vec<&Expression<CborValue>> = resolved.iter().collect();
        let (endpoint, consumed) = self.endpoint_url(table_name, &conds)?;

        let (query_consumed, client_filters) = self.split_filters(&conds, consumed);
        let query = self.build_query_string(window, &conds, &query_consumed, order);
        let url = join_query(&endpoint, &query);

        // The `(0, 1)` window is the count probe (reads only the envelope's
        // total); log it at debug so it doesn't drown the real data fetches.
        if self.debug {
            if window == Some((0, 1)) {
                tracing::debug!(target: "vantage_api_client::rest", table = table_name, url = %url, "REST GET (count probe)");
            } else {
                tracing::info!(target: "vantage_api_client::rest", table = table_name, url = %url, "REST GET");
            }
        }

        // Time every round trip, unconditionally — a remote API is the one part
        // of a read the process cannot bound, and a slow page is far more often
        // one slow GET than anything local. Reported regardless of `debug` so
        // the cost is attributable from a default log; a request over a second
        // is worth an operator's attention, hence `info` at that point.
        let started = std::time::Instant::now();
        let policy = crate::transport::policy_for(Priority::current());
        let response = self
            .client
            .execute_with(&policy, |http| {
                let req = http.get(&url);
                match self.auth_header.value() {
                    Some(auth) => req.header(AUTHORIZATION, auth),
                    None => req,
                }
            })
            .await
            .map_err(|e| {
                let ms = started.elapsed().as_millis() as u64;
                // A transient failure (5xx, network, open breaker) is visible
                // in the datasource's health and will be tried again by a
                // later call; only an answer no retry can change is worth a
                // warning.
                if e.is_final() {
                    tracing::warn!(
                        target: "vantage_api_client::rest",
                        table = table_name,
                        url = %url,
                        ms,
                        attempts = e.attempts,
                        "REST GET failed",
                    );
                } else {
                    tracing::debug!(
                        target: "vantage_api_client::rest",
                        table = table_name,
                        url = %url,
                        ms,
                        attempts = e.attempts,
                        kind = e.kind_name(),
                        "REST GET gave up on a transient failure",
                    );
                }
                crate::transport::client_error(e, "API request failed", &url)
            })?;

        let body: serde_json::Value = response
            .json()
            .await
            .map_err(|e| error!("Failed to parse API response as JSON", detail = e))?;

        let ms = started.elapsed().as_millis() as u64;
        let probe = window == Some((0, 1));
        if ms >= 1000 {
            tracing::info!(
                target: "vantage_api_client::rest",
                table = table_name,
                url = %url,
                ms,
                count_probe = probe,
                "slow REST GET",
            );
        } else {
            tracing::debug!(
                target: "vantage_api_client::rest",
                table = table_name,
                url = %url,
                ms,
                count_probe = probe,
                "REST GET done",
            );
        }

        Ok((body, client_filters))
    }

    /// Also reports the envelope total when `total_key` is configured and the
    /// body carries it. Unlike [`Self::fetch_total`] this never errors on a
    /// missing total: the rows are the point here, and a caller that needs a
    /// definitive count can still ask for one.
    ///
    /// A window comes back short only when the server ran out: callers take a
    /// short window as the end of the set.
    async fn fetch_windowed<'a>(
        &self,
        table_name: &str,
        id_field: Option<&str>,
        window: Option<(i64, i64)>,
        order: Order<'_>,
        conditions: impl IntoIterator<Item = &'a Expression<CborValue>>,
    ) -> Result<(IndexMap<String, Record<CborValue>>, Option<i64>)> {
        // Non-paginating endpoints return the whole list on the first
        // window; a later window would just re-deliver the same rows and the
        // perpetual grid would never mark itself exhausted. Short-circuit any
        // window past the start to empty so the grid sees the chunk shrink
        // and stops asking for more.
        if self.no_pagination && window.is_some_and(|(offset, _)| offset > 0) {
            return Ok((IndexMap::new(), None));
        }
        let conditions: Vec<&Expression<CborValue>> = conditions.into_iter().collect();
        let Some((offset, limit)) = window.filter(|_| !self.no_pagination) else {
            let page = self
                .fetch_page(table_name, id_field, window, order, &conditions)
                .await?;
            return Ok((page.rows, page.total));
        };
        let (offset, limit) = (offset.max(0), limit.max(1));

        // Page-based APIs are addressed by whole pages: start at the page
        // holding `offset`.
        let start = if self.pagination.skip_based {
            offset
        } else {
            offset - offset % limit
        };
        let first = self
            .fetch_page(
                table_name,
                id_field,
                Some((start, limit)),
                order,
                &conditions,
            )
            .await?;
        if start == offset && !first.lossy {
            return Ok((first.rows, first.total));
        }

        // Assemble the window page by page until it is full or the server
        // runs out. When rows were dropped (client filters, repeated ids) a
        // row's place in the result is only known by walking from the start.
        let (mut server_offset, mut skip) = if first.lossy {
            (0, offset)
        } else {
            (start, offset - start)
        };
        let mut page = (server_offset == start).then_some(first);
        let mut rows = IndexMap::new();
        let mut total = None;
        let mut lossy = false;
        loop {
            let current = match page.take() {
                Some(p) => p,
                None => {
                    self.fetch_page(
                        table_name,
                        id_field,
                        Some((server_offset, limit)),
                        order,
                        &conditions,
                    )
                    .await?
                }
            };
            total = total.or(current.total);
            lossy |= current.lossy;
            for (id, row) in current.rows {
                if skip > 0 {
                    skip -= 1;
                } else if (rows.len() as i64) < limit {
                    rows.insert(id, row);
                }
            }
            if rows.len() as i64 >= limit || (current.server_rows as i64) < limit {
                break;
            }
            server_offset += limit;
        }
        Ok((rows, if lossy { None } else { total }))
    }

    /// One request: the page's rows keyed by id, the envelope total, and how
    /// many rows the server sent — more than `rows` holds when client filters
    /// or repeated ids dropped some (`lossy`).
    async fn fetch_page(
        &self,
        table_name: &str,
        id_field: Option<&str>,
        window: Option<(i64, i64)>,
        order: Order<'_>,
        conditions: &[&Expression<CborValue>],
    ) -> Result<Page> {
        let (body, client_filters) = self
            .fetch_raw_body(table_name, window, order, conditions.iter().copied())
            .await?;
        let total = self
            .total_key
            .as_deref()
            .and_then(|key| body.get(key))
            .and_then(|v| v.as_i64());
        let data = self.extract_array(&body, table_name)?;

        let mut records = IndexMap::new();
        for (row_idx, item) in data.iter().enumerate() {
            let obj = item
                .as_object()
                .ok_or_else(|| error!("API data item is not an object", index = row_idx))?;

            // Extract ID from the configured id_field, or use row index
            let id = id_field
                .and_then(|field| obj.get(field))
                .and_then(|v| match v {
                    serde_json::Value::String(s) => Some(s.clone()),
                    serde_json::Value::Number(n) => Some(n.to_string()),
                    _ => None,
                })
                .unwrap_or_else(|| row_idx.to_string());

            // The HTTP body parses as JSON for free; convert to CBOR
            // at this single boundary so the rest of the pipeline
            // (Table, Vista) sees the universal carrier. json_to_cbor
            // is total — JSON is a strict subset of CBOR.
            let mut record: Record<CborValue> = Record::new();
            for (k, v) in obj {
                record.insert(k.clone(), vantage_types::json_to_cbor(v.clone()));
            }

            records.insert(id, record);
        }

        // Client-side filtering (FilterStrategy::Client): drop rows that
        // don't match the non-path eq-conditions. A condition whose field
        // is absent from a row is treated as a pass (it was a path/request
        // param, not a record field) — mirroring the AWS connector and the
        // Mercury CLI's own post-fetch `_filter_deployments`.
        if !client_filters.is_empty() {
            records.retain(|_id, record| {
                client_filters
                    .iter()
                    .all(|(field, want)| match record.get(field) {
                        Some(v) => crate::cbor_to_query_string(v).as_deref() == Some(want.as_str()),
                        None => true,
                    })
            });
        }

        // Counts the rows this call hands back, so a client-side filter shows
        // up as fewer rows pulled than the server sent.
        self.client
            .report(TransportEvent::RowsPulled { n: records.len() });

        // The envelope counted what the SERVER matched, before any rows were
        // dropped above — reporting that total now would size a grid to rows it
        // will never be given. No total is better than a wrong one.
        let total = if client_filters.is_empty() {
            total
        } else {
            None
        };
        Ok(Page {
            lossy: records.len() < data.len(),
            server_rows: data.len(),
            rows: records,
            total,
        })
    }
}

fn urlencode(s: &str) -> String {
    urlencoding::encode(s).into_owned()
}

/// Join a base URL and a path with exactly one slash between them,
/// regardless of whether either side already carries one.
fn join_base_path(base: &str, path: &str) -> String {
    format!(
        "{}/{}",
        base.trim_end_matches('/'),
        path.trim_start_matches('/')
    )
}

/// Append a `build_query_string` result (always opening with `?`, or empty)
/// to an endpoint URL. The table path may itself carry a query string (e.g.
/// `launches/?mode=detailed`), in which case the appended params must join
/// with `&` — otherwise the URL gets two `?` and the API rejects it.
fn join_query(endpoint: &str, query: &str) -> String {
    match query.strip_prefix('?') {
        Some(rest) if endpoint.contains('?') => format!("{endpoint}&{rest}"),
        _ => format!("{endpoint}{query}"),
    }
}

/// Walk an `Expression`'s parameter tree and force any `Deferred`
/// branches to their resolved form. Used at the `fetch_records`
/// boundary so the URL builder only sees sync scalars.
///
/// Recursion lives on the heap (boxed) because the future's body
/// contains another `async` call of the same shape — Rust can't size
/// a directly-recursive `async fn` without indirection.
fn resolve_deferreds(
    mut expr: Expression<CborValue>,
) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<Expression<CborValue>>> + Send>> {
    Box::pin(async move {
        for param in expr.parameters.iter_mut() {
            match param {
                ExpressiveEnum::Deferred(deferred) => {
                    *param = deferred.call().await?;
                }
                ExpressiveEnum::Nested(inner) => {
                    let resolved = resolve_deferreds(inner.clone()).await?;
                    *inner = resolved;
                }
                ExpressiveEnum::Scalar(_) => {}
            }
        }
        Ok(expr)
    })
}

impl RestApi {
    /// Pull the row array out of the response body, according to the
    /// configured `ResponseShape`. A bare-array API answers a single-record
    /// endpoint (`activities/{id}`) with one object: that reads as one row.
    fn extract_array<'a>(
        &self,
        body: &'a serde_json::Value,
        table_name: &str,
    ) -> Result<&'a [serde_json::Value]> {
        match &self.response_shape {
            ResponseShape::BareArray => match body {
                serde_json::Value::Array(rows) => Ok(rows),
                serde_json::Value::Object(_) => Ok(std::slice::from_ref(body)),
                _ => Err(error!(
                    "Expected response body to be a JSON array or object (BareArray shape)"
                )),
            },
            ResponseShape::Wrapped { array_key } => body[array_key]
                .as_array()
                .map(Vec::as_slice)
                .ok_or_else(|| {
                    error!(
                        "Response missing array under wrapper key",
                        array_key = array_key
                    )
                }),
            ResponseShape::WrappedByTableName => body[table_name]
                .as_array()
                .map(Vec::as_slice)
                .ok_or_else(|| {
                    error!(
                        "Response missing array under table-name key",
                        table_name = table_name
                    )
                }),
        }
    }
}

/// Builder for [`RestApi`]. Lets callers pick a [`ResponseShape`] and
/// override the pagination parameter names.
///
/// ```no_run
/// use vantage_api_client::{RestApi, ResponseShape, PaginationParams};
///
/// // JSONPlaceholder: bare arrays, JSON-Server pagination conventions.
/// let api = RestApi::builder("https://jsonplaceholder.typicode.com")
///     .response_shape(ResponseShape::BareArray)
///     .build();
///
/// // DummyJSON: wrapped-by-table-name, skip-based pagination.
/// let api = RestApi::builder("https://dummyjson.com")
///     .response_shape(ResponseShape::WrappedByTableName)
///     .pagination_params(PaginationParams::skip_limit("skip", "limit"))
///     .build();
/// ```
#[derive(Clone, Debug)]
pub struct RestApiBuilder {
    base_url: String,
    auth_header: AuthHeader,
    response_shape: ResponseShape,
    pagination: PaginationParams,
    /// `pagination_params` was called: the API is known to page.
    paged: bool,
    no_pagination: bool,
    filter_strategy: FilterStrategy,
    total_key: Option<String>,
    ordering: Option<OrderingParams>,
    debug: bool,
    transport: crate::transport::ClientConfig,
}

impl RestApiBuilder {
    fn new(base_url: String) -> Self {
        Self {
            base_url,
            auth_header: AuthHeader::default(),
            response_shape: ResponseShape::default(),
            pagination: PaginationParams::default(),
            paged: false,
            no_pagination: false,
            filter_strategy: FilterStrategy::default(),
            total_key: None,
            ordering: None,
            debug: false,
            transport: crate::transport::ClientConfig::default(),
        }
    }

    /// Concurrent requests to this API at most (default 4).
    pub fn max_parallel(mut self, n: usize) -> Self {
        self.transport.max_parallel = n.max(1);
        self
    }

    /// Requests per second to this API at most.
    pub fn rate_limit(mut self, per_second: f64) -> Self {
        self.transport.rate_limit = Some(per_second);
        self
    }

    /// Report every attempt, retry and breaker transition under `key`
    /// (the datasource name).
    pub fn observer(
        mut self,
        key: impl Into<Arc<str>>,
        observer: Arc<dyn TransportObserver>,
    ) -> Self {
        self.transport.observer = Some((key.into(), observer));
        self
    }

    /// A pre-configured `reqwest::Client` (timeouts, proxies, TLS).
    pub fn http_client(mut self, client: reqwest::Client) -> Self {
        self.transport.http = Some(client);
        self
    }

    /// Set the Authorization header value (e.g. "Bearer `<token>`").
    pub fn auth(mut self, auth: impl Into<String>) -> Self {
        self.auth_header = AuthHeader::new(auth);
        self.transport.auth_refresher = None;
        self
    }

    /// Get the bearer token from `refresher` instead of a fixed header: it
    /// is asked on the first request, and again when the API answers `401`.
    /// A request waits for it, so a refresher may run an interactive
    /// sign-in. Replaces [`auth`](Self::auth).
    pub fn auth_refresher(mut self, refresher: crate::AuthRefresher) -> Self {
        self.auth_header = AuthHeader::default();
        self.transport.auth_refresher = Some(refresher);
        self
    }

    /// Choose how the API wraps its row array. Defaults to
    /// `Wrapped { array_key: "data" }` for backwards compat.
    pub fn response_shape(mut self, shape: ResponseShape) -> Self {
        self.response_shape = shape;
        self
    }

    /// Override the page/limit query parameter names. Default is
    /// `("_page", "_limit")` (JSON Server convention).
    pub fn pagination_params(mut self, pagination: PaginationParams) -> Self {
        self.pagination = pagination;
        self.paged = true;
        self
    }

    /// Disable pagination entirely — no `_page`/`_limit` query
    /// params are appended, and a request for page > 1 is short-
    /// circuited to an empty result. Use this for APIs that don't
    /// paginate (return the full list every call) or that treat
    /// unknown query params as strict filters.
    pub fn no_pagination(mut self) -> Self {
        self.no_pagination = true;
        self
    }

    /// Choose how non-path eq-conditions are applied. Default is
    /// [`FilterStrategy::Query`]; use [`FilterStrategy::Client`] for
    /// APIs that only filter via path segments and reject/ignore unknown
    /// query params (the conditions are then applied in memory).
    pub fn filter_strategy(mut self, strategy: FilterStrategy) -> Self {
        self.filter_strategy = strategy;
        self
    }

    /// Name the response-envelope key carrying the grand total of matching
    /// rows (e.g. `count`). Setting it lets the shell report an exact count
    /// and advertise `can_fetch_window` for lazy/scroll loading.
    pub fn total_key(mut self, key: impl Into<String>) -> Self {
        self.total_key = Some(key.into());
        self
    }

    /// The API's sort query param (see [`OrderingParams`]). With it set the
    /// vista reports `can_order` and every column is orderable.
    pub fn ordering(mut self, ordering: OrderingParams) -> Self {
        self.ordering = Some(ordering);
        self
    }

    /// Emit `tracing` events for window/count requests.
    pub fn debug(mut self, debug: bool) -> Self {
        self.debug = debug;
        self
    }

    pub fn build(self) -> RestApi {
        RestApi {
            base_url: self.base_url,
            client: crate::transport::build_client(self.transport),
            auth_header: self.auth_header,
            response_shape: self.response_shape,
            pagination: self.pagination,
            paged: self.paged,
            no_pagination: self.no_pagination,
            filter_strategy: self.filter_strategy,
            total_key: self.total_key,
            ordering: self.ordering,
            debug: self.debug,
        }
    }
}

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

    /// `build_query_string` with no conditions, exercising only the
    /// window → pagination-param mapping.
    fn qs(api: &RestApi, window: Option<(i64, i64)>) -> String {
        api.build_query_string(window, &[], &[], None)
    }

    #[test]
    fn debug_masks_auth_header() {
        let api = RestApi::builder("http://x")
            .auth("Bearer secret-token")
            .build();
        let text = format!("{api:?}");
        assert!(!text.contains("secret-token"), "{text}");
        assert!(text.contains("<set>"), "{text}");
    }

    #[test]
    fn skip_based_window_uses_offset_verbatim() {
        let api = RestApi::builder("http://x")
            .pagination_params(PaginationParams::skip_limit("skip", "limit"))
            .build();
        assert_eq!(qs(&api, Some((20, 10))), "?skip=20&limit=10");
    }

    #[test]
    fn page_based_window_derives_one_based_page() {
        let api = RestApi::builder("http://x").build(); // default _page/_limit
        // offset 20 / limit 10 → page 3 (1-based).
        assert_eq!(qs(&api, Some((20, 10))), "?_page=3&_limit=10");
    }

    #[test]
    fn no_window_emits_no_pagination_params() {
        let api = RestApi::builder("http://x").build();
        assert_eq!(qs(&api, None), "");
    }

    #[test]
    fn no_pagination_suppresses_window_params() {
        let api = RestApi::builder("http://x").no_pagination().build();
        assert_eq!(qs(&api, Some((20, 10))), "");
    }

    #[test]
    fn query_string_joins_plain_endpoint_with_question_mark() {
        assert_eq!(
            join_query("http://x/launches/", "?_page=1&_limit=10"),
            "http://x/launches/?_page=1&_limit=10"
        );
    }

    #[test]
    fn query_string_joins_templated_endpoint_with_ampersand() {
        // Endpoint already carries `?mode=detailed`; pagination must append
        // with `&`, not a second `?`.
        assert_eq!(
            join_query("http://x/launches/?mode=detailed", "?offset=0&limit=1"),
            "http://x/launches/?mode=detailed&offset=0&limit=1"
        );
    }

    #[test]
    fn empty_query_string_leaves_endpoint_untouched() {
        assert_eq!(
            join_query("http://x/launches/?mode=detailed", ""),
            "http://x/launches/?mode=detailed"
        );
    }

    /// Live regression for the double-`?` bug: a real fetch against the
    /// Launch Library 2 dev API using a table path that already carries a
    /// query string (`launches/?mode=detailed`). Before the `join_query`
    /// fix the request URL was `…/launches/?mode=detailed?offset=0&limit=1`
    /// and the server answered 500. Network-gated, so `#[ignore]`d:
    /// `cargo test -p vantage-api-client -- --ignored query_string`.
    #[tokio::test]
    #[ignore = "hits the live Launch Library 2 dev API"]
    async fn live_templated_table_path_fetches_rows() {
        let api = RestApi::builder("https://lldev.thespacedevs.com/2.3.0")
            .pagination_params(PaginationParams::skip_limit("offset", "limit"))
            .response_shape(ResponseShape::Wrapped {
                array_key: "results".into(),
            })
            .total_key("count")
            .build();

        let total = api
            .fetch_total("launches/?mode=detailed", [])
            .await
            .expect("fetch_total");
        assert!(total.is_some_and(|n| n > 0), "expected a positive count");

        let (rows, window_total) = api
            .fetch_window_records_counted("launches/?mode=detailed", Some("id"), 0, 3, None, [])
            .await
            .expect("fetch_window_records_counted");
        assert_eq!(rows.len(), 3, "expected the requested 3-row window");
        // The whole point of the counted window: the same response that
        // carried the rows also carried the count, so `fetch_total`'s extra
        // round trip buys nothing a caller couldn't already have.
        assert_eq!(
            window_total, total,
            "the window's envelope total should match the dedicated count",
        );
    }
}