spiral-sdk 0.1.7

// Generated by tools/service-contracts/rust.py. Do not edit.
#![doc = "Generated reqwest client with typed errors and streaming bodies."]

use async_trait::async_trait;
use reqwest::{Client, Url};
use spiral_contract::auth::AuthHeader;
use spiral_contract::call::{CallError, ServiceFailure};
use spiral_contract::error::ErrorEnvelope;
use spiral_contract::header::encode_header_request;
use spiral_contract::stream::ByteStream;

use crate::api::scandal::ScandalApi;
use crate::api::scandal::objects::*;
/// HTTP client for the scandal service.
#[derive(Clone)]
pub struct ScandalClient {
    http: Client,
    base: Url,
}

impl ScandalClient {
    /// Create a redirect-disabled client; base may contain a deployment prefix.
    #[expect(
        clippy::disallowed_methods,
        reason = "Generated clients keep their transport independent of the Spiral application runtime."
    )]
    pub fn new(base: &str) -> Result<Self, CallError<ScandalError>> {
        let base = Url::parse(base)
            .map_err(|_| CallError::<ScandalError>::protocol("invalid base URL"))?;
        if !matches!(base.scheme(), "http" | "https")
            || base.host_str().is_none()
            || !base.username().is_empty()
            || base.password().is_some()
            || base.query().is_some()
            || base.fragment().is_some()
        {
            return Err(CallError::<ScandalError>::protocol("invalid base URL"));
        }
        let http = Client::builder()
            .redirect(reqwest::redirect::Policy::none())
            .connect_timeout(std::time::Duration::from_secs(10))
            .build()
            .map_err(CallError::transport)?;
        Ok(Self { http, base })
    }
    /// Supply transport instrumentation or custom connection policy.
    ///
    /// The caller must disable redirects to protect bearer credentials.
    pub fn with_http(mut self, http: Client) -> Self {
        self.http = http;
        self
    }
    /// Discard an upload's stored parts. An abort never removes a stored object, so the attempt's
    /// binding allows it. Aborting an upload that no longer exists succeeds.
    pub async fn abort_blob_upload_http(
        &self,
        auth_header: AuthHeader,
        request: AbortBlobUploadRequest,
    ) -> Result<(), CallError<ScandalError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<ScandalError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("blobs");
            path.push("uploads");
            path.push("abort");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder
            .header("content-type", "application/json")
            .json(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Write Arrow rows as new files of a Table, for a successor of `base` or, without a base, for
    /// the Table's first version. The rows stay invisible until a version that includes them is
    /// prepared, registered and committed. Appends to one Table may run in parallel. With an
    /// `attempt`, the Table's binding is the one the Tables API issued for that attempt, which
    /// creates files only in its directory.
    pub async fn append_rows_http(
        &self,
        auth_header: AuthHeader,
        request: AppendRowsRequest,
        body: reqwest::Body,
    ) -> Result<AppendedFragments, CallError<ScandalError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<ScandalError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("writes");
            path.push("append");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let header = encode_header_request(&request)
            .map_err(|_| CallError::<ScandalError>::protocol("request cannot be encoded"))?;
        let builder = builder
            .header("scandal-request", header)
            .header("content-type", "application/vnd.apache.arrow.stream")
            .body(body);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Write values of new columns for every row one file of a Table version keeps, as column files
    /// attached to that file, which leaves its existing files as they are. The service writes the
    /// rows as they arrive, holding what it has not yet written in the request's memory budget. The
    /// values stay invisible until a version that includes the receipt is prepared on an `edit`
    /// base, registered and committed.
    pub async fn attach_columns_http(
        &self,
        auth_header: AuthHeader,
        request: AttachColumnsRequest,
        body: reqwest::Body,
    ) -> Result<AttachedColumns, CallError<ScandalError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<ScandalError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("writes");
            path.push("columns");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let header = encode_header_request(&request)
            .map_err(|_| CallError::<ScandalError>::protocol("request cannot be encoded"))?;
        let builder = builder
            .header("scandal-request", header)
            .header("content-type", "application/vnd.apache.arrow.stream")
            .body(body);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Return the output schema of named expressions over an input schema, without any data.
    pub async fn bind_projection_http(
        &self,
        auth_header: AuthHeader,
        request: BindProjectionRequest,
    ) -> Result<ArrowSchema, CallError<ScandalError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<ScandalError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("projections");
            path.push("bind");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder
            .header("content-type", "application/json")
            .json(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Join an upload's parts into its blob and return the blob, whose digest merges the parts'
    /// hashes, so no byte is read again. The blob becomes part of the Table only when a committed
    /// row references it.
    pub async fn complete_blob_upload_http(
        &self,
        auth_header: AuthHeader,
        request: CompleteBlobUploadRequest,
    ) -> Result<Blob, CallError<ScandalError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<ScandalError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("blobs");
            path.push("uploads");
            path.push("complete");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder
            .header("content-type", "application/json")
            .json(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Record the deletion of rows of a Table version, addressed by row ID, without rewriting the
    /// rows it keeps. The deletions stay invisible until a version that includes the receipts is
    /// prepared on an `edit` base, registered and committed.
    pub async fn delete_rows_http(
        &self,
        auth_header: AuthHeader,
        request: DeleteRowsRequest,
        body: reqwest::Body,
    ) -> Result<AppendedFragments, CallError<ScandalError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<ScandalError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("writes");
            path.push("delete");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let header = encode_header_request(&request)
            .map_err(|_| CallError::<ScandalError>::protocol("request cannot be encoded"))?;
        let builder = builder
            .header("scandal-request", header)
            .header("content-type", "application/vnd.apache.arrow.stream")
            .body(body);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Evaluate named expressions over the Arrow rows in the body and stream the results back row
    /// for row. Expressions read blobs from the Tables in `tables`, and from other URIs resolved by
    /// the execution host. A failure after the response starts ends the stream without the Arrow
    /// end-of-stream marker.
    pub async fn evaluate_http(
        &self,
        auth_header: AuthHeader,
        request: EvaluateRequest,
        body: reqwest::Body,
    ) -> Result<reqwest::Response, CallError<ScandalError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<ScandalError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("projections");
            path.push("evaluate");
        }
        let builder = self
            .http
            .post(url)
            .header("accept", "application/vnd.apache.arrow.stream");
        let builder = builder.header("authorization", auth_header.as_str());
        let header = encode_header_request(&request)
            .map_err(|_| CallError::<ScandalError>::protocol("request cannot be encoded"))?;
        let builder = builder
            .header("scandal-request", header)
            .header("content-type", "application/vnd.apache.arrow.stream")
            .body(body);
        let response = self
            .send(builder, "application/vnd.apache.arrow.stream")
            .await?;
        Ok(response)
    }
    /// Prepare a Table version from the files that appends wrote, writing only its metadata; the
    /// files stay where the appends put them. Deletes select stable row IDs and write deletion
    /// files through their separate attempt binding. Register the version with the Tables API and
    /// commit it to publish it. The Table binding authorizes publication.
    pub async fn finalize_write_http(
        &self,
        auth_header: AuthHeader,
        request: FinalizeWriteRequest,
    ) -> Result<PreparedWriteReceipt, CallError<ScandalError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<ScandalError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("writes");
            path.push("finalize");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder
            .header("content-type", "application/json")
            .json(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// List the SPQL functions this service can run.
    pub async fn list_functions_http(
        &self,
        auth_header: AuthHeader,
    ) -> Result<FunctionCatalog, CallError<ScandalError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<ScandalError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("functions");
        }
        let builder = self.http.get(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Store the inline payloads of the `spql.Blob` columns in the Arrow rows of the body as blobs
    /// of a Table, packed as the rows of `append_rows` are, and return the rows with each payload
    /// replaced by its address, `spiral://<table-id>/<key>`, its offset, size and BLAKE3 digest.
    /// Payloads share packs in the Table's `blobs/` directory, or in the attempt's; one of 4 MiB or
    /// more is an object of its own. Other values are returned as they are. The response starts
    /// once every pack is stored. The blobs become part of the Table only when a committed row
    /// references them. With an `attempt`, the binding the Tables API issued for that attempt is
    /// enough.
    pub async fn pack_blobs_http(
        &self,
        auth_header: AuthHeader,
        request: PackBlobsRequest,
        body: reqwest::Body,
    ) -> Result<reqwest::Response, CallError<ScandalError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<ScandalError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("blobs");
            path.push("pack");
        }
        let builder = self
            .http
            .post(url)
            .header("accept", "application/vnd.apache.arrow.stream");
        let builder = builder.header("authorization", auth_header.as_str());
        let header = encode_header_request(&request)
            .map_err(|_| CallError::<ScandalError>::protocol("request cannot be encoded"))?;
        let builder = builder
            .header("scandal-request", header)
            .header("content-type", "application/vnd.apache.arrow.stream")
            .body(body);
        let response = self
            .send(builder, "application/vnd.apache.arrow.stream")
            .await?;
        Ok(response)
    }
    /// Write new values for existing rows of a Table version, addressed by row ID, as patch files
    /// beside the data. The rows keep their row IDs and positions. The service writes the rows in
    /// bounded chunks as they arrive, holding each chunk in the request's memory budget. The values
    /// stay invisible until a version that includes the receipts is prepared on an `edit` base,
    /// registered and committed.
    pub async fn patch_rows_http(
        &self,
        auth_header: AuthHeader,
        request: PatchRowsRequest,
        body: reqwest::Body,
    ) -> Result<AppendedFragments, CallError<ScandalError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<ScandalError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("writes");
            path.push("patch");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let header = encode_header_request(&request)
            .map_err(|_| CallError::<ScandalError>::protocol("request cannot be encoded"))?;
        let builder = builder
            .header("scandal-request", header)
            .header("content-type", "application/vnd.apache.arrow.stream")
            .body(body);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Plan how one version of a Table changed since an earlier one, without reading column values:
    /// for each file of the later version, whether the earlier version has it, with the row IDs
    /// patched and deleted since, or it was added since, with the row IDs appended since and those
    /// a rewrite carried over. Planning reads version metadata, the row IDs that patches and
    /// deletes target, and the stored row IDs of files whose row IDs are not ranges. Read the rows
    /// a plan names with a scan's `rowIds`.
    pub async fn plan_changes_http(
        &self,
        auth_header: AuthHeader,
        request: PlanChangesRequest,
    ) -> Result<ChangePlan, CallError<ScandalError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<ScandalError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("changes");
            path.push("plan");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder
            .header("content-type", "application/json")
            .json(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Plan a scan of one Table version: its output schema, how exactly it can enforce each filter
    /// clause, and one split for each file that statistics do not prune, in stored order. Planning
    /// reads only metadata, and refuses an emit it cannot run before any rows are read.
    pub async fn plan_scan_http(
        &self,
        auth_header: AuthHeader,
        request: PlanScanRequest,
    ) -> Result<ScanPlan, CallError<ScandalError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<ScandalError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("scans");
            path.push("plan");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder
            .header("content-type", "application/json")
            .json(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Stream the bytes of a blob that a Table owns, or of a byte range of it. Check a whole read
    /// against the blob's digest and a range against its length; a stream that ends early is
    /// incomplete.
    pub async fn read_blob_http(
        &self,
        auth_header: AuthHeader,
        request: ReadBlobRequest,
    ) -> Result<reqwest::Response, CallError<ScandalError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<ScandalError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("blobs");
            path.push("read");
        }
        let builder = self
            .http
            .post(url)
            .header("accept", "application/octet-stream");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder
            .header("content-type", "application/json")
            .json(&request);
        let response = self.send(builder, "application/octet-stream").await?;
        Ok(response)
    }
    /// Read legacy keyed table rows using metadata resolved by the Tables API and a scoped storage
    /// binding. ScanDAL merges key-space and column-group fragments, WAL patches and deletions
    /// through the legacy native reader. It calls no control-plane service. Returns an Arrow IPC
    /// stream, including its schema when limit is zero. Hosted ScanDAL rejects local file roots
    /// with the same policy as Arbor reads.
    pub async fn read_legacy_table_http(
        &self,
        auth_header: AuthHeader,
        request: ReadLegacyTableRequest,
    ) -> Result<reqwest::Response, CallError<ScandalError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<ScandalError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("legacy-tables");
            path.push("read");
        }
        let builder = self
            .http
            .post(url)
            .header("accept", "application/vnd.apache.arrow.stream");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder
            .header("content-type", "application/json")
            .json(&request);
        let response = self
            .send(builder, "application/vnd.apache.arrow.stream")
            .await?;
        Ok(response)
    }
    /// Read one split of a planned scan as an Arrow IPC stream, in stored order. Every batch ends
    /// with two dictionary columns that arrive even with a batch of no rows: `scandal.next_row`,
    /// the stored row of the split's file to resume from, and `scandal.metrics`, the read's
    /// counters as JSON on the split's last batch when the scan collects them. With an emit, the
    /// batches hold the split's partial results, which are complete only at the end of the stream,
    /// and `scandal.next_row` is the split's end. A failure after the stream starts ends it without
    /// the Arrow end-of-stream marker.
    pub async fn read_scan_http(
        &self,
        auth_header: AuthHeader,
        request: ReadScanRequest,
    ) -> Result<reqwest::Response, CallError<ScandalError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<ScandalError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("scans");
            path.push("read");
        }
        let builder = self
            .http
            .post(url)
            .header("accept", "application/vnd.apache.arrow.stream");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder
            .header("content-type", "application/json")
            .json(&request);
        let response = self
            .send(builder, "application/vnd.apache.arrow.stream")
            .await?;
        Ok(response)
    }
    /// Start a multipart upload of one blob into a write attempt, for a payload too large for one
    /// request. Send its parts to any instance, in any order and each retried on its own, then
    /// complete it for a blob that a row appended in the same write can carry. A binding the Tables
    /// API issued for the attempt is enough.
    pub async fn start_blob_upload_http(
        &self,
        auth_header: AuthHeader,
        request: StartBlobUploadRequest,
    ) -> Result<BlobUpload, CallError<ScandalError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<ScandalError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("blobs");
            path.push("uploads");
            path.push("start");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder
            .header("content-type", "application/json")
            .json(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Report what storage holds for a blob that a Table owns, without reading its bytes.
    pub async fn stat_blob_http(
        &self,
        auth_header: AuthHeader,
        request: StatBlobRequest,
    ) -> Result<BlobStat, CallError<ScandalError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<ScandalError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("blobs");
            path.push("stat");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder
            .header("content-type", "application/json")
            .json(&request);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Read rows of planned splits by position, as an Arrow IPC stream in request order: each
    /// take's rows in the order of its positions, a repeated position's row again, and the takes
    /// one after another. Each split's rows are read in one pass, so positions that share storage
    /// share its reads. Every batch ends with the two dictionary columns of `read_scan`:
    /// `scandal.next_row` holds how many of the request's positions, counted across its takes, the
    /// batches so far delivered, so a client resumes by sending the positions after them;
    /// `scandal.metrics` holds the counters of every split read, on a last batch of no rows, when
    /// the scan collects them. A position past a split's rows fails the request before any output.
    /// A failure after the stream starts ends it without the Arrow end-of-stream marker.
    pub async fn take_rows_http(
        &self,
        auth_header: AuthHeader,
        request: TakeRowsRequest,
    ) -> Result<reqwest::Response, CallError<ScandalError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<ScandalError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("scans");
            path.push("take");
        }
        let builder = self
            .http
            .post(url)
            .header("accept", "application/vnd.apache.arrow.stream");
        let builder = builder.header("authorization", auth_header.as_str());
        let builder = builder
            .header("content-type", "application/json")
            .json(&request);
        let response = self
            .send(builder, "application/vnd.apache.arrow.stream")
            .await?;
        Ok(response)
    }
    /// Store one part of an upload from the request body. Every part but the last holds exactly the
    /// upload's `partBytes`. Sending a part again replaces it. The part counts against the
    /// request's memory budget.
    pub async fn upload_blob_part_http(
        &self,
        auth_header: AuthHeader,
        request: UploadBlobPartRequest,
        body: reqwest::Body,
    ) -> Result<BlobPart, CallError<ScandalError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<ScandalError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("blobs");
            path.push("uploads");
            path.push("part");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let header = encode_header_request(&request)
            .map_err(|_| CallError::<ScandalError>::protocol("request cannot be encoded"))?;
        let builder = builder
            .header("scandal-request", header)
            .header("content-type", "application/octet-stream")
            .body(body);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    /// Store the request body as a new blob of a Table, under its `blobs/` directory, and return
    /// its reference. The service stops reading and fails the request when the body exceeds
    /// `maxBytes`. The blob becomes part of the Table only when a committed row references it. A
    /// publication binding from the Tables API is enough.
    pub async fn write_blob_http(
        &self,
        auth_header: AuthHeader,
        request: WriteBlobRequest,
        body: reqwest::Body,
    ) -> Result<Blob, CallError<ScandalError>> {
        let mut url = self.base.clone();
        {
            let mut path = url.path_segments_mut().map_err(|_| {
                CallError::<ScandalError>::protocol("base URL cannot contain path segments")
            })?;
            path.pop_if_empty();
            path.push("blobs");
            path.push("write");
        }
        let builder = self.http.post(url).header("accept", "application/json");
        let builder = builder.header("authorization", auth_header.as_str());
        let header = encode_header_request(&request)
            .map_err(|_| CallError::<ScandalError>::protocol("request cannot be encoded"))?;
        let builder = builder
            .header("scandal-request", header)
            .header("content-type", "application/octet-stream")
            .body(body);
        let response = self.send(builder, "application/json").await?;
        let status = response.status().as_u16();
        let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
        serde_json::from_slice(&bytes)
            .map_err(|_| CallError::protocol_response(status, "invalid JSON response"))
    }
    async fn send(
        &self,
        builder: reqwest::RequestBuilder,
        expected_media: &str,
    ) -> Result<reqwest::Response, CallError<ScandalError>> {
        let response = builder.send().await.map_err(CallError::transport)?;
        let status = response.status();
        if !status.is_success() {
            let status = status.as_u16();
            let content_type = response
                .headers()
                .get("content-type")
                .and_then(|value| value.to_str().ok())
                .map(str::to_owned);
            let (bytes, truncated) = bounded_error_body::<ScandalError>(
                response,
                64 * 1024,
                status,
                content_type.clone(),
            )
            .await?;
            if truncated {
                return Err(CallError::unknown_remote(status, content_type, bytes, true));
            }
            return match serde_json::from_slice::<ErrorEnvelope>(&bytes) {
                Ok(envelope) => Err(CallError::Service(ServiceFailure::remote(status, envelope))),
                Err(_) => Err(CallError::unknown_remote(
                    status,
                    content_type,
                    bytes,
                    false,
                )),
            };
        }
        if !media_matches(&response, expected_media) {
            return Err(CallError::protocol_response(
                status.as_u16(),
                "unexpected response media type",
            ));
        }
        Ok(response)
    }
}

fn media_matches(response: &reqwest::Response, expected: &str) -> bool {
    response
        .headers()
        .get("content-type")
        .and_then(|value| value.to_str().ok())
        .and_then(|value| value.split(';').next())
        .is_some_and(|value| value.trim().eq_ignore_ascii_case(expected))
}

async fn bounded_body(
    mut response: reqwest::Response,
    limit: usize,
) -> Result<Vec<u8>, CallError<ScandalError>> {
    let status = response.status().as_u16();
    let mut result = Vec::new();
    while let Some(chunk) = response
        .chunk()
        .await
        .map_err(|error| CallError::transport_response(status, error))?
    {
        if chunk.len() > limit.saturating_sub(result.len()) {
            return Err(CallError::protocol_response(
                status,
                "response exceeds JSON size limit",
            ));
        }
        result.extend_from_slice(&chunk);
    }
    Ok(result)
}

async fn bounded_error_body<E>(
    mut response: reqwest::Response,
    limit: usize,
    status: u16,
    content_type: Option<String>,
) -> Result<(Vec<u8>, bool), CallError<E>> {
    let mut result = Vec::new();
    loop {
        let chunk = response.chunk().await.map_err(|error| {
            CallError::unreadable_remote(status, content_type.clone(), result.clone(), error)
        })?;
        let Some(chunk) = chunk else {
            return Ok((result, false));
        };
        let remaining = limit.saturating_sub(result.len());
        if chunk.len() > remaining {
            result.extend_from_slice(&chunk[..remaining]);
            return Ok((result, true));
        }
        result.extend_from_slice(&chunk);
    }
}
#[async_trait]
impl ScandalApi for ScandalClient {
    async fn abort_blob_upload(
        &self,
        auth_header: AuthHeader,
        request: AbortBlobUploadRequest,
    ) -> Result<(), CallError<ScandalError>> {
        let response = Self::abort_blob_upload_http(self, auth_header, request).await?;
        Ok(response)
    }
    async fn append_rows(
        &self,
        auth_header: AuthHeader,
        request: AppendRowsRequest,
        body: ByteStream,
    ) -> Result<AppendedFragments, CallError<ScandalError>> {
        let response =
            Self::append_rows_http(self, auth_header, request, reqwest::Body::wrap_stream(body))
                .await?;
        Ok(response)
    }
    async fn attach_columns(
        &self,
        auth_header: AuthHeader,
        request: AttachColumnsRequest,
        body: ByteStream,
    ) -> Result<AttachedColumns, CallError<ScandalError>> {
        let response =
            Self::attach_columns_http(self, auth_header, request, reqwest::Body::wrap_stream(body))
                .await?;
        Ok(response)
    }
    async fn bind_projection(
        &self,
        auth_header: AuthHeader,
        request: BindProjectionRequest,
    ) -> Result<ArrowSchema, CallError<ScandalError>> {
        let response = Self::bind_projection_http(self, auth_header, request).await?;
        Ok(response)
    }
    async fn complete_blob_upload(
        &self,
        auth_header: AuthHeader,
        request: CompleteBlobUploadRequest,
    ) -> Result<Blob, CallError<ScandalError>> {
        let response = Self::complete_blob_upload_http(self, auth_header, request).await?;
        Ok(response)
    }
    async fn delete_rows(
        &self,
        auth_header: AuthHeader,
        request: DeleteRowsRequest,
        body: ByteStream,
    ) -> Result<AppendedFragments, CallError<ScandalError>> {
        let response =
            Self::delete_rows_http(self, auth_header, request, reqwest::Body::wrap_stream(body))
                .await?;
        Ok(response)
    }
    async fn evaluate(
        &self,
        auth_header: AuthHeader,
        request: EvaluateRequest,
        body: ByteStream,
    ) -> Result<ByteStream, CallError<ScandalError>> {
        let response =
            Self::evaluate_http(self, auth_header, request, reqwest::Body::wrap_stream(body))
                .await?;
        Ok(ByteStream::new(response.bytes_stream()))
    }
    async fn finalize_write(
        &self,
        auth_header: AuthHeader,
        request: FinalizeWriteRequest,
    ) -> Result<PreparedWriteReceipt, CallError<ScandalError>> {
        let response = Self::finalize_write_http(self, auth_header, request).await?;
        Ok(response)
    }
    async fn list_functions(
        &self,
        auth_header: AuthHeader,
    ) -> Result<FunctionCatalog, CallError<ScandalError>> {
        let response = Self::list_functions_http(self, auth_header).await?;
        Ok(response)
    }
    async fn pack_blobs(
        &self,
        auth_header: AuthHeader,
        request: PackBlobsRequest,
        body: ByteStream,
    ) -> Result<ByteStream, CallError<ScandalError>> {
        let response =
            Self::pack_blobs_http(self, auth_header, request, reqwest::Body::wrap_stream(body))
                .await?;
        Ok(ByteStream::new(response.bytes_stream()))
    }
    async fn patch_rows(
        &self,
        auth_header: AuthHeader,
        request: PatchRowsRequest,
        body: ByteStream,
    ) -> Result<AppendedFragments, CallError<ScandalError>> {
        let response =
            Self::patch_rows_http(self, auth_header, request, reqwest::Body::wrap_stream(body))
                .await?;
        Ok(response)
    }
    async fn plan_changes(
        &self,
        auth_header: AuthHeader,
        request: PlanChangesRequest,
    ) -> Result<ChangePlan, CallError<ScandalError>> {
        let response = Self::plan_changes_http(self, auth_header, request).await?;
        Ok(response)
    }
    async fn plan_scan(
        &self,
        auth_header: AuthHeader,
        request: PlanScanRequest,
    ) -> Result<ScanPlan, CallError<ScandalError>> {
        let response = Self::plan_scan_http(self, auth_header, request).await?;
        Ok(response)
    }
    async fn read_blob(
        &self,
        auth_header: AuthHeader,
        request: ReadBlobRequest,
    ) -> Result<ByteStream, CallError<ScandalError>> {
        let response = Self::read_blob_http(self, auth_header, request).await?;
        Ok(ByteStream::new(response.bytes_stream()))
    }
    async fn read_legacy_table(
        &self,
        auth_header: AuthHeader,
        request: ReadLegacyTableRequest,
    ) -> Result<ByteStream, CallError<ScandalError>> {
        let response = Self::read_legacy_table_http(self, auth_header, request).await?;
        Ok(ByteStream::new(response.bytes_stream()))
    }
    async fn read_scan(
        &self,
        auth_header: AuthHeader,
        request: ReadScanRequest,
    ) -> Result<ByteStream, CallError<ScandalError>> {
        let response = Self::read_scan_http(self, auth_header, request).await?;
        Ok(ByteStream::new(response.bytes_stream()))
    }
    async fn start_blob_upload(
        &self,
        auth_header: AuthHeader,
        request: StartBlobUploadRequest,
    ) -> Result<BlobUpload, CallError<ScandalError>> {
        let response = Self::start_blob_upload_http(self, auth_header, request).await?;
        Ok(response)
    }
    async fn stat_blob(
        &self,
        auth_header: AuthHeader,
        request: StatBlobRequest,
    ) -> Result<BlobStat, CallError<ScandalError>> {
        let response = Self::stat_blob_http(self, auth_header, request).await?;
        Ok(response)
    }
    async fn take_rows(
        &self,
        auth_header: AuthHeader,
        request: TakeRowsRequest,
    ) -> Result<ByteStream, CallError<ScandalError>> {
        let response = Self::take_rows_http(self, auth_header, request).await?;
        Ok(ByteStream::new(response.bytes_stream()))
    }
    async fn upload_blob_part(
        &self,
        auth_header: AuthHeader,
        request: UploadBlobPartRequest,
        body: ByteStream,
    ) -> Result<BlobPart, CallError<ScandalError>> {
        let response = Self::upload_blob_part_http(
            self,
            auth_header,
            request,
            reqwest::Body::wrap_stream(body),
        )
        .await?;
        Ok(response)
    }
    async fn write_blob(
        &self,
        auth_header: AuthHeader,
        request: WriteBlobRequest,
        body: ByteStream,
    ) -> Result<Blob, CallError<ScandalError>> {
        let response =
            Self::write_blob_http(self, auth_header, request, reqwest::Body::wrap_stream(body))
                .await?;
        Ok(response)
    }
}