videosdk/resources/

batch_calls.rs

//! Bulk outbound phone campaigns.
//!
//! Lifecycle: [`upload`](BatchCallsResource::upload) a recipients file — it is
//! uploaded and parsed in one call — then [`create`](BatchCallsResource::create)
//! to launch or schedule the campaign.

use std::collections::HashMap;
use std::fmt;
use std::path::Path;
use std::sync::Arc;
use std::time::{Duration, Instant};

use futures_util::Stream;
use reqwest::Method;
use serde::de::{Deserializer, Error as _};
use serde::{Deserialize, Serialize};
use serde_json::{json, Map, Value};

use crate::client::{CallOptions, Client};
use crate::common::{string_enum, MessageResponse};
use crate::error::{Error, ErrorKind, Result};
use crate::pagination::{auto_page, paginate, ListParams, Page, PageFetcher};
use crate::query::QueryBuilder;
use crate::resources::escape;

const PATH: &str = "/v2/batch-calls";

/// The recipients file may be at most 5 MiB.
const MAX_UPLOAD_BYTES: usize = 5 * 1024 * 1024;
const DEFAULT_POLL_INTERVAL: Duration = Duration::from_secs(2);
const DEFAULT_POLL_TIMEOUT: Duration = Duration::from_secs(120);

/// Resolves the poll cadence and deadline.
///
/// A zero interval would spin the loop as fast as the API answers, and a zero
/// deadline would give up before the first check. Treat either as unset rather
/// than honouring a value that cannot have been meant.
fn resolve_poll(interval: Option<Duration>, timeout: Option<Duration>) -> (Duration, Duration) {
    (
        interval
            .filter(|interval| !interval.is_zero())
            .unwrap_or(DEFAULT_POLL_INTERVAL),
        timeout
            .filter(|timeout| !timeout.is_zero())
            .unwrap_or(DEFAULT_POLL_TIMEOUT),
    )
}

/// The content types the upload endpoint will sign.
fn upload_content_type(extension: &str) -> Option<&'static str> {
    match extension.to_ascii_lowercase().as_str() {
        "csv" => Some("text/csv"),
        "xls" => Some("application/vnd.ms-excel"),
        "xlsx" => Some("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"),
        _ => None,
    }
}

string_enum! {
    /// When a batch runs.
    BatchCallTiming {
        /// Run as soon as it is created.
        IMMEDIATE => "immediate",
        /// Run at the scheduled time.
        SCHEDULED => "scheduled",
    }
}

string_enum! {
    /// A batch's lifecycle state.
    BatchCallStatus {
        /// The recipients file was uploaded.
        UPLOADED => "uploaded",
        /// The recipients file is being parsed.
        PARSING => "parsing",
        /// The recipients file parsed, and the batch is ready to launch.
        PARSED => "parsed",
        /// The recipients file could not be parsed.
        PARSE_FAILED => "parse_failed",
        /// The batch is saved as a draft.
        DRAFT => "draft",
        /// The batch is scheduled.
        SCHEDULED => "scheduled",
        /// The batch is placing calls.
        RUNNING => "running",
        /// The batch finished.
        COMPLETED => "completed",
        /// The batch is winding down.
        CANCELLING => "cancelling",
        /// The batch was cancelled.
        CANCELLED => "cancelled",
    }
}

string_enum! {
    /// How a batch is cancelled.
    BatchCallCancelMode {
        /// Let in-flight calls finish.
        GRACEFUL => "graceful",
        /// Drop in-flight calls.
        IMMEDIATE => "immediate",
    }
}

string_enum! {
    /// A call status that can trigger a retry.
    BatchCallRetryStatus {
        /// The call failed.
        FAILED => "failed",
        /// The call was not answered.
        NOT_ANSWERED => "not-answered",
        /// The call was rejected.
        REJECTED => "rejected",
        /// The call was missed.
        MISSED => "missed",
        /// The caller hung up first.
        ABANDONED => "abandoned",
    }
}

string_enum! {
    /// A batch-call webhook event.
    BatchCallWebhookEvent {
        /// The batch started.
        BATCHCALL_STARTED => "batchcall-started",
        /// The batch completed.
        BATCHCALL_COMPLETED => "batchcall-completed",
    }
}

/// Configures retries for a batch.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct BatchCallRetryConfig {
    /// Whether to retry.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub enabled: Option<bool>,
    /// Which call statuses trigger a retry.
    #[serde(
        default,
        deserialize_with = "crate::common::null_to_default",
        skip_serializing_if = "Vec::is_empty"
    )]
    pub retry_on: Vec<BatchCallRetryStatus>,
    /// The maximum attempts, from 1 to 3.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub max_attempts: Option<u32>,
}

/// Configures a batch's webhook.
#[derive(Debug, Clone, Default, Serialize)]
pub struct BatchCallWebhookConfig {
    /// Where to deliver events.
    pub endpoint: String,
    /// The events to subscribe to.
    #[serde(skip_serializing_if = "Vec::is_empty")]
    pub events: Vec<BatchCallWebhookEvent>,
}

/// A bulk outbound phone campaign.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct BatchCall {
    /// The batch id.
    pub batch_id: String,
    /// The batch's display name.
    pub batch_name: Option<String>,
    /// The batch's description.
    pub description: Option<String>,
    /// The outbound phone number id.
    pub phone_number_id: Option<String>,
    /// The outbound phone number.
    pub phone_number: Option<String>,
    /// The outbound routing rule id.
    pub routing_rule_id: Option<String>,
    /// When the batch runs.
    pub timing: Option<BatchCallTiming>,
    /// The IANA timezone the schedule is interpreted in.
    pub timezone: Option<String>,
    /// The scheduled time, as `HH:MM`.
    pub scheduled_time: Option<String>,
    /// The scheduled date, as `YYYY-MM-DD`.
    pub scheduled_date: Option<String>,
    /// The start of the allowed calling window, as `HH:MM`.
    pub allowed_from: Option<String>,
    /// The end of the allowed calling window, as `HH:MM`.
    pub allowed_until: Option<String>,
    /// The days the batch may call: 0 is Sunday through 6, Saturday.
    #[serde(default, deserialize_with = "crate::common::null_to_default")]
    pub days_of_week: Vec<u32>,
    /// The batch's retry configuration.
    pub retry: Option<BatchCallRetryConfig>,
    /// The batch's status.
    pub status: Option<BatchCallStatus>,
    /// Where the recipients file can be downloaded.
    pub cdn_url: Option<String>,
    /// How many records were uploaded.
    pub records_uploaded: Option<u32>,
    /// Why parsing failed, when it did.
    pub parse_error: Option<String>,
    /// The batch's timestamp (scheduled run time, or creation time for
    /// immediate batches), in epoch seconds.
    pub timestamp: Option<i64>,
    /// Completion percentage, from 0 to 100 (list view only).
    pub progress: Option<f64>,
    /// When the batch was created.
    pub created_at: Option<String>,
    /// When the batch was last updated.
    pub updated_at: Option<String>,
    /// Any fields the server returned that this SDK does not model yet.
    #[serde(flatten)]
    pub extra: Map<String, Value>,
}

/// A single per-number record in a batch.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct BatchCallRecord {
    /// The batch this record belongs to.
    #[serde(default, deserialize_with = "crate::common::null_to_default")]
    pub batch_id: String,
    /// The record id.
    pub record_id: String,
    /// The number to call.
    pub phone_number: Option<String>,
    /// Free-form metadata, merged into the call.
    pub metadata: Option<Map<String, Value>>,
    /// The record's status.
    pub status: Option<String>,
    /// The call placed for this record.
    pub call_id: Option<String>,
    /// Why the call did not connect.
    pub reason: Option<String>,
    /// How many attempts were made.
    pub attempts: Option<u32>,
    /// The recording of this record's call.
    pub recording_id: Option<String>,
    /// The recording object.
    pub recording: Option<Value>,
    /// How long the call lasted, in seconds.
    pub call_duration: Option<f64>,
    /// When the record was created.
    pub created_at: Option<String>,
    /// When the record was last updated.
    pub updated_at: Option<String>,
    /// Any fields the server returned that this SDK does not model yet.
    #[serde(flatten)]
    pub extra: Map<String, Value>,
}

/// A status breakdown for a batch.
#[derive(Debug, Clone, Default)]
pub struct BatchCallStats {
    /// The batch this breakdown describes.
    pub batch_id: String,
    /// The batch's status.
    pub status: Option<BatchCallStatus>,
    /// The total record count.
    pub total: Option<i64>,
    /// The per-status tallies, keyed by display label, e.g. `scheduled`,
    /// `running`, `completed`, `not connected`, `cancelled` — plus `total`.
    pub counts: HashMap<String, i64>,
}

impl<'de> Deserialize<'de> for BatchCallStats {
    /// The per-status breakdown arrives as arbitrary sibling keys, so every
    /// remaining numeric field is collected into `counts`.
    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> std::result::Result<Self, D::Error> {
        let raw = Map::deserialize(deserializer)?;
        let mut stats = BatchCallStats {
            batch_id: raw
                .get("batchId")
                .and_then(Value::as_str)
                .unwrap_or_default()
                .to_string(),
            ..Default::default()
        };

        if let Some(status) = raw.get("status") {
            stats.status = Some(serde_json::from_value(status.clone()).map_err(D::Error::custom)?);
        }

        for (key, value) in &raw {
            if key == "batchId" || key == "status" {
                continue;
            }
            if let Some(count) = value.as_i64() {
                stats.counts.insert(key.clone(), count);
                if key == "total" {
                    stats.total = Some(count);
                }
            }
        }
        Ok(stats)
    }
}

/// The presigned target the API returns for a recipients file.
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct BatchPresignedUpload {
    presigned_url: String,
    batch_id: String,
}

/// The result of a bulk record deletion.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DeleteRecordsResult {
    /// The server's confirmation message.
    pub message: String,
    /// How many records were deleted.
    #[serde(default, deserialize_with = "crate::common::null_to_default")]
    pub deleted_count: u32,
}

/* --------------------------------- parameters --------------------------------- */

/// The parameters for [`BatchCallsResource::upload`].
#[derive(Default)]
pub struct UploadBatchCallParams {
    /// The path to a `.csv`, `.xls` or `.xlsx` recipients file, at most 5 MiB.
    /// Required.
    pub file_path: String,
    /// Replaces the file on an existing batch, instead of creating a new one.
    pub batch_id: Option<String>,
    /// Waits until parsing finishes before returning. Defaults to `true`.
    ///
    /// When `false`, upload returns as soon as parsing starts, with the batch
    /// still in [`BatchCallStatus::PARSING`].
    pub wait_for_parse: Option<bool>,
    /// How often to poll while waiting. Defaults to 2 seconds.
    pub poll_interval: Option<Duration>,
    /// How long to wait for parsing. Defaults to 2 minutes.
    pub poll_timeout: Option<Duration>,
    /// Called once parsing has started. An optional progress hook — the parsed
    /// batch is the return value of upload.
    pub on_parsing: Option<Box<dyn FnOnce() + Send>>,
}

impl fmt::Debug for UploadBatchCallParams {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("UploadBatchCallParams")
            .field("file_path", &self.file_path)
            .field("batch_id", &self.batch_id)
            .field("wait_for_parse", &self.wait_for_parse)
            .field("poll_interval", &self.poll_interval)
            .field("poll_timeout", &self.poll_timeout)
            .field("on_parsing", &self.on_parsing.is_some())
            .finish()
    }
}

/// The parameters for [`BatchCallsResource::create`].
#[derive(Debug, Clone, Default, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CreateBatchCallParams {
    /// The display name. Required.
    pub batch_name: String,
    /// The outbound phone number id. Required.
    pub phone_number_id: String,
    /// The outbound routing rule id, attached to the number. Required.
    pub routing_rule_id: String,
    /// The batch's description.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub description: Option<String>,
    /// When the batch runs.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub timing: Option<BatchCallTiming>,
    /// An IANA timezone. Defaults to `Asia/Calcutta`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub timezone: Option<String>,
    /// The scheduled time, as `HH:MM`. Required when scheduled.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub scheduled_time: Option<String>,
    /// The scheduled date, as `YYYY-MM-DD`. Required when scheduled.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub scheduled_date: Option<String>,
    /// The start of the allowed calling window, as `HH:MM`. Scheduled batches only.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub allowed_from: Option<String>,
    /// The end of the allowed calling window, as `HH:MM`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub allowed_until: Option<String>,
    /// The days the batch may call: 0 is Sunday through 6, Saturday.
    #[serde(skip_serializing_if = "Vec::is_empty")]
    pub days_of_week: Vec<u32>,
    /// The retry configuration.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub retry: Option<BatchCallRetryConfig>,
    /// The webhook configuration.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub webhook: Option<BatchCallWebhookConfig>,
    /// Finalizes a previously uploaded and parsed draft.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub batch_id: Option<String>,
    /// Saves as a draft instead of launching.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub save_as_draft: Option<bool>,
}

/// The editable fields of a batch: draft, scheduled or parsed.
#[derive(Debug, Clone, Default, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct UpdateBatchCallParams {
    /// The display name.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub batch_name: Option<String>,
    /// The outbound phone number id.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub phone_number_id: Option<String>,
    /// The outbound routing rule id.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub routing_rule_id: Option<String>,
    /// The batch's description.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub description: Option<String>,
    /// When the batch runs.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub timing: Option<BatchCallTiming>,
    /// An IANA timezone.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub timezone: Option<String>,
    /// The scheduled time, as `HH:MM`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub scheduled_time: Option<String>,
    /// The scheduled date, as `YYYY-MM-DD`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub scheduled_date: Option<String>,
    /// The start of the allowed calling window.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub allowed_from: Option<String>,
    /// The end of the allowed calling window.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub allowed_until: Option<String>,
    /// The days the batch may call.
    #[serde(skip_serializing_if = "Vec::is_empty")]
    pub days_of_week: Vec<u32>,
    /// The retry configuration.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub retry: Option<BatchCallRetryConfig>,
    /// Saves as a draft instead of launching.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub save_as_draft: Option<bool>,
}

/// The query parameters for [`BatchCallsResource::list`].
#[derive(Debug, Clone, Default)]
pub struct ListBatchCallsParams {
    /// The 1-based page number.
    pub page: Option<u32>,
    /// Items per page.
    pub per_page: Option<u32>,
    /// An opaque cursor from a previous page.
    pub cursor: Option<String>,
    /// Filters by batch status.
    pub status: Option<BatchCallStatus>,
    /// An ISO 8601 lower bound.
    pub start_date: Option<String>,
    /// An ISO 8601 upper bound.
    pub end_date: Option<String>,
    /// The minimum number of uploaded records.
    pub min_value: Option<u32>,
    /// The maximum number of uploaded records.
    pub max_value: Option<u32>,
}

/// The query parameters for [`BatchCallRecordsResource::list`].
#[derive(Debug, Clone, Default)]
pub struct ListBatchCallRecordsParams {
    /// The 1-based page number.
    pub page: Option<u32>,
    /// Items per page.
    pub per_page: Option<u32>,
    /// An opaque cursor from a previous page.
    pub cursor: Option<String>,
    /// A comma-separated set of display labels, e.g. `scheduled,not connected`.
    pub status: Option<String>,
    /// Matches over record id, phone number, status and reason.
    pub search: Option<String>,
}

/// The editable fields of a pending record.
#[derive(Debug, Clone, Default, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct UpdateBatchCallRecordParams {
    /// The number to call.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub phone_number: Option<String>,
    /// Free-form metadata, merged into the call.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub metadata: Option<Map<String, Value>>,
}

/// The query parameters for [`BatchCallRecordsResource::export`].
#[derive(Debug, Clone, Default)]
pub struct ExportBatchCallRecordsParams {
    /// A comma-separated set of display labels.
    pub status: Option<String>,
    /// Matches over record id, phone number, status and reason.
    pub search: Option<String>,
}

macro_rules! pagination {
    ($name:ident) => {
        impl $name {
            fn pagination(&self) -> ListParams {
                ListParams {
                    page: self.page,
                    per_page: self.per_page,
                    cursor: self.cursor.clone(),
                }
            }
        }
    };
}

pagination!(ListBatchCallsParams);
pagination!(ListBatchCallRecordsParams);

/* --------------------------------- resources ----------------------------------- */

/// Bulk outbound phone campaigns. Reached via [`Client::batch_call`].
#[derive(Debug, Clone, Copy)]
pub struct BatchCallsResource<'a> {
    client: &'a Client,
}

impl<'a> BatchCallsResource<'a> {
    pub(crate) fn new(client: &'a Client) -> Self {
        Self { client }
    }

    /// A batch's per-number records.
    pub fn records(&self) -> BatchCallRecordsResource<'a> {
        BatchCallRecordsResource {
            client: self.client,
        }
    }

    /// Uploads a recipients file, starts parsing, and — unless `wait_for_parse`
    /// is `false` — waits until the batch is parsed and ready to launch.
    ///
    /// Pass the returned batch's `batch_id` to
    /// [`create`](BatchCallsResource::create).
    ///
    /// Under the hood this asks for a presigned URL, `PUT`s the file straight to
    /// storage without an `Authorization` header — sending one would invalidate
    /// the presigned signature — then triggers parsing and polls.
    ///
    /// Drop the returned future to cancel.
    pub async fn upload(&self, params: UploadBatchCallParams) -> Result<BatchCall> {
        let path = Path::new(&params.file_path);
        let file_name = path
            .file_name()
            .map(|name| name.to_string_lossy().into_owned())
            .ok_or_else(|| {
                Error::validation(format!(
                    "invalid batch-call file path {:?}",
                    params.file_path
                ))
            })?;

        let extension = path
            .extension()
            .map(|e| e.to_string_lossy())
            .unwrap_or_default();
        let content_type = upload_content_type(&extension).ok_or_else(|| {
            Error::validation(format!(
                "unsupported batch-call file {file_name:?}; allowed types: .csv, .xls, .xlsx"
            ))
        })?;

        let data = tokio::fs::read(&params.file_path).await.map_err(|e| {
            Error::validation(format!(
                "could not read batch-call file {:?}: {e}",
                params.file_path
            ))
        })?;
        if data.len() > MAX_UPLOAD_BYTES {
            return Err(Error::validation(format!(
                "batch-call file exceeds the {} MB limit",
                MAX_UPLOAD_BYTES / 1024 / 1024
            )));
        }

        let mut upload_body = Map::new();
        upload_body.insert("fileName".into(), json!(file_name));
        upload_body.insert("fileSize".into(), json!(data.len()));
        if let Some(batch_id) = &params.batch_id {
            upload_body.insert("batchId".into(), json!(batch_id));
        }

        let upload_path = format!("{PATH}/upload");
        let presigned: BatchPresignedUpload = self
            .client
            .json(Method::POST, &upload_path, CallOptions::json(&upload_body)?)
            .await?;

        self.client
            .put_binary(&presigned.presigned_url, &data, content_type)
            .await?;

        let parse_path = format!("{PATH}/parse");
        let parse_body = json!({ "batchId": presigned.batch_id });
        self.client
            .none(Method::POST, &parse_path, CallOptions::json(&parse_body)?)
            .await?;

        if let Some(on_parsing) = params.on_parsing {
            on_parsing();
        }

        if params.wait_for_parse == Some(false) {
            return self.get(&presigned.batch_id).await;
        }
        let (interval, timeout) = resolve_poll(params.poll_interval, params.poll_timeout);
        self.poll_until_parsed(&presigned.batch_id, interval, timeout)
            .await
    }

    /// Polls until the batch is parsed, parsing fails, or the timeout elapses.
    async fn poll_until_parsed(
        &self,
        batch_id: &str,
        interval: Duration,
        timeout: Duration,
    ) -> Result<BatchCall> {
        let deadline = Instant::now() + timeout;
        loop {
            let batch = self.get(batch_id).await?;
            match batch.status.as_ref().map(BatchCallStatus::as_str) {
                Some("parsed") => return Ok(batch),
                Some("parse_failed") => {
                    let mut message = "parsing the batch-call file failed".to_string();
                    if let Some(parse_error) = &batch.parse_error {
                        message.push_str(": ");
                        message.push_str(parse_error);
                    }
                    return Err(Error::operation(
                        ErrorKind::Generic,
                        "batchcall_parse_failed",
                        message,
                    ));
                }
                _ => {}
            }

            if Instant::now() >= deadline {
                let status = batch
                    .status
                    .as_ref()
                    .map(BatchCallStatus::as_str)
                    .unwrap_or("unknown");
                return Err(Error::operation(
                    ErrorKind::Timeout,
                    "batchcall_parse_timeout",
                    format!(
                        "timed out after {timeout:?} waiting for batch {batch_id:?} to finish \
                         parsing (last status: {status})"
                    ),
                ));
            }
            tokio::time::sleep(interval).await;
        }
    }

    /// Launches or schedules a batch from a parsed `batch_id`, or saves a draft.
    pub async fn create(&self, params: CreateBatchCallParams) -> Result<BatchCall> {
        self.client
            .json(Method::POST, PATH, CallOptions::json(&params)?)
            .await
    }

    /// Lists batches, one page at a time.
    pub async fn list(&self, params: ListBatchCallsParams) -> Result<Page<BatchCall>> {
        paginate(
            self.fetcher(&params),
            &params.pagination(),
            "batchCalls",
            None,
        )
        .await
    }

    /// Lists batches, transparently fetching every page.
    pub fn list_stream(
        &self,
        params: ListBatchCallsParams,
    ) -> impl Stream<Item = Result<BatchCall>> + Send {
        auto_page(
            self.fetcher(&params),
            params.pagination(),
            "batchCalls",
            None,
        )
    }

    /// Fetches a batch by id.
    pub async fn get(&self, batch_id: &str) -> Result<BatchCall> {
        let path = format!("{PATH}/{}", escape(batch_id));
        self.client
            .json(Method::GET, &path, CallOptions::new())
            .await
    }

    /// Updates an editable batch: draft, scheduled or parsed.
    pub async fn update(&self, batch_id: &str, params: UpdateBatchCallParams) -> Result<BatchCall> {
        let path = format!("{PATH}/{}", escape(batch_id));
        self.client
            .json(Method::PUT, &path, CallOptions::json(&params)?)
            .await
    }

    /// Soft-deletes a batch and its records.
    pub async fn delete(&self, batch_id: &str) -> Result<MessageResponse> {
        let path = format!("{PATH}/{}", escape(batch_id));
        self.client
            .json(Method::DELETE, &path, CallOptions::new())
            .await
    }

    /// Cancels a running or scheduled batch.
    pub async fn cancel(&self, batch_id: &str, mode: BatchCallCancelMode) -> Result<BatchCall> {
        let path = format!("{PATH}/{}/cancel", escape(batch_id));
        let body = json!({ "mode": mode });
        self.client
            .json(Method::POST, &path, CallOptions::json(&body)?)
            .await
    }

    /// Returns a status breakdown for a batch.
    pub async fn stats(&self, batch_id: &str) -> Result<BatchCallStats> {
        let path = format!("{PATH}/{}/stats", escape(batch_id));
        self.client
            .json(Method::GET, &path, CallOptions::new())
            .await
    }

    fn fetcher(&self, params: &ListBatchCallsParams) -> PageFetcher {
        let client = self.client.clone();
        let params = params.clone();
        Arc::new(move |page, per_page| {
            let client = client.clone();
            let params = params.clone();
            Box::pin(async move {
                let query = QueryBuilder::new()
                    .opt("page", page)
                    .opt("perPage", per_page)
                    .opt_str(
                        "status",
                        params.status.as_ref().map(BatchCallStatus::as_str),
                    )
                    .opt_str("startDate", params.start_date.as_deref())
                    .opt_str("endDate", params.end_date.as_deref())
                    .opt("minValue", params.min_value)
                    .opt("maxValue", params.max_value)
                    .into_pairs();
                client
                    .json::<Value>(Method::GET, PATH, CallOptions::new().query(query))
                    .await
            })
        })
    }
}

/// A batch's per-number records. Reached via [`BatchCallsResource::records`].
#[derive(Debug, Clone, Copy)]
pub struct BatchCallRecordsResource<'a> {
    client: &'a Client,
}

impl<'a> BatchCallRecordsResource<'a> {
    /// Lists a batch's records, one page at a time.
    pub async fn list(
        &self,
        batch_id: &str,
        params: ListBatchCallRecordsParams,
    ) -> Result<Page<BatchCallRecord>> {
        let fetcher = self.fetcher(batch_id, &params);
        paginate(fetcher, &params.pagination(), "records", None).await
    }

    /// Lists a batch's records, transparently fetching every page.
    pub fn list_stream(
        &self,
        batch_id: &str,
        params: ListBatchCallRecordsParams,
    ) -> impl Stream<Item = Result<BatchCallRecord>> + Send {
        let fetcher = self.fetcher(batch_id, &params);
        auto_page(fetcher, params.pagination(), "records", None)
    }

    /// Edits a single pending record before it runs.
    pub async fn update(
        &self,
        batch_id: &str,
        record_id: &str,
        params: UpdateBatchCallRecordParams,
    ) -> Result<BatchCallRecord> {
        let path = format!("{PATH}/{}/records/{}", escape(batch_id), escape(record_id));
        self.client
            .json(Method::PUT, &path, CallOptions::json(&params)?)
            .await
    }

    /// Bulk soft-deletes records.
    pub async fn delete(
        &self,
        batch_id: &str,
        record_ids: &[String],
    ) -> Result<DeleteRecordsResult> {
        let path = format!("{PATH}/{}/records", escape(batch_id));
        let body = json!({ "recordIds": record_ids });
        self.client
            .json(Method::DELETE, &path, CallOptions::json(&body)?)
            .await
    }

    /// Exports a batch's records as CSV text.
    pub async fn export(
        &self,
        batch_id: &str,
        params: ExportBatchCallRecordsParams,
    ) -> Result<String> {
        let path = format!("{PATH}/{}/records/export", escape(batch_id));
        let query = QueryBuilder::new()
            .opt_str("status", params.status.as_deref())
            .opt_str("search", params.search.as_deref())
            .into_pairs();
        self.client
            .text(Method::GET, &path, CallOptions::new().query(query))
            .await
    }

    fn fetcher(&self, batch_id: &str, params: &ListBatchCallRecordsParams) -> PageFetcher {
        let client = self.client.clone();
        let params = params.clone();
        let path = format!("{PATH}/{}/records", escape(batch_id));
        Arc::new(move |page, per_page| {
            let client = client.clone();
            let params = params.clone();
            let path = path.clone();
            Box::pin(async move {
                // This endpoint calls its search parameter `query`.
                let query = QueryBuilder::new()
                    .opt("page", page)
                    .opt("perPage", per_page)
                    .opt_str("status", params.status.as_deref())
                    .opt_str("query", params.search.as_deref())
                    .into_pairs();
                client
                    .json::<Value>(Method::GET, &path, CallOptions::new().query(query))
                    .await
            })
        })
    }
}

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

    #[test]
    fn stats_collect_every_numeric_sibling_into_counts() {
        let stats: BatchCallStats = serde_json::from_value(json!({
            "batchId": "b-1",
            "status": "running",
            "total": 10,
            "scheduled": 3,
            "completed": 5,
            "not connected": 2,
            "someString": "ignored",
        }))
        .unwrap();

        assert_eq!(stats.batch_id, "b-1");
        assert_eq!(stats.status.unwrap().as_str(), "running");
        assert_eq!(stats.total, Some(10));
        assert_eq!(stats.counts["scheduled"], 3);
        assert_eq!(stats.counts["not connected"], 2);
        assert_eq!(stats.counts["total"], 10, "total is also a count");
        assert!(!stats.counts.contains_key("someString"));
        assert!(!stats.counts.contains_key("status"));
    }

    #[test]
    fn a_zero_poll_interval_or_timeout_falls_back_to_the_default() {
        // Honouring a zero interval would busy-loop the API.
        let (interval, timeout) = resolve_poll(Some(Duration::ZERO), Some(Duration::ZERO));
        assert_eq!(interval, DEFAULT_POLL_INTERVAL);
        assert_eq!(timeout, DEFAULT_POLL_TIMEOUT);

        let (interval, timeout) = resolve_poll(None, None);
        assert_eq!(interval, DEFAULT_POLL_INTERVAL);
        assert_eq!(timeout, DEFAULT_POLL_TIMEOUT);

        // A real value is honoured, however small.
        let (interval, timeout) = resolve_poll(
            Some(Duration::from_millis(5)),
            Some(Duration::from_millis(50)),
        );
        assert_eq!(interval, Duration::from_millis(5));
        assert_eq!(timeout, Duration::from_millis(50));
    }

    #[test]
    fn upload_content_types_cover_the_allowed_extensions() {
        assert_eq!(upload_content_type("csv"), Some("text/csv"));
        assert_eq!(upload_content_type("CSV"), Some("text/csv"));
        assert_eq!(upload_content_type("xls"), Some("application/vnd.ms-excel"));
        assert!(upload_content_type("xlsx").is_some());
        assert_eq!(upload_content_type("txt"), None);
        assert_eq!(upload_content_type(""), None);
    }
}