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";
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);
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),
)
}
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! {
BatchCallTiming {
IMMEDIATE => "immediate",
SCHEDULED => "scheduled",
}
}
string_enum! {
BatchCallStatus {
UPLOADED => "uploaded",
PARSING => "parsing",
PARSED => "parsed",
PARSE_FAILED => "parse_failed",
DRAFT => "draft",
SCHEDULED => "scheduled",
RUNNING => "running",
COMPLETED => "completed",
CANCELLING => "cancelling",
CANCELLED => "cancelled",
}
}
string_enum! {
BatchCallCancelMode {
GRACEFUL => "graceful",
IMMEDIATE => "immediate",
}
}
string_enum! {
BatchCallRetryStatus {
FAILED => "failed",
NOT_ANSWERED => "not-answered",
REJECTED => "rejected",
MISSED => "missed",
ABANDONED => "abandoned",
}
}
string_enum! {
BatchCallWebhookEvent {
BATCHCALL_STARTED => "batchcall-started",
BATCHCALL_COMPLETED => "batchcall-completed",
}
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct BatchCallRetryConfig {
#[serde(skip_serializing_if = "Option::is_none")]
pub enabled: Option<bool>,
#[serde(
default,
deserialize_with = "crate::common::null_to_default",
skip_serializing_if = "Vec::is_empty"
)]
pub retry_on: Vec<BatchCallRetryStatus>,
#[serde(skip_serializing_if = "Option::is_none")]
pub max_attempts: Option<u32>,
}
#[derive(Debug, Clone, Default, Serialize)]
pub struct BatchCallWebhookConfig {
pub endpoint: String,
#[serde(skip_serializing_if = "Vec::is_empty")]
pub events: Vec<BatchCallWebhookEvent>,
}
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct BatchCall {
pub batch_id: String,
pub batch_name: Option<String>,
pub description: Option<String>,
pub phone_number_id: Option<String>,
pub phone_number: Option<String>,
pub routing_rule_id: Option<String>,
pub timing: Option<BatchCallTiming>,
pub timezone: Option<String>,
pub scheduled_time: Option<String>,
pub scheduled_date: Option<String>,
pub allowed_from: Option<String>,
pub allowed_until: Option<String>,
#[serde(default, deserialize_with = "crate::common::null_to_default")]
pub days_of_week: Vec<u32>,
pub retry: Option<BatchCallRetryConfig>,
pub status: Option<BatchCallStatus>,
pub cdn_url: Option<String>,
pub records_uploaded: Option<u32>,
pub parse_error: Option<String>,
pub timestamp: Option<i64>,
pub progress: Option<f64>,
pub created_at: Option<String>,
pub updated_at: Option<String>,
#[serde(flatten)]
pub extra: Map<String, Value>,
}
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct BatchCallRecord {
#[serde(default, deserialize_with = "crate::common::null_to_default")]
pub batch_id: String,
pub record_id: String,
pub phone_number: Option<String>,
pub metadata: Option<Map<String, Value>>,
pub status: Option<String>,
pub call_id: Option<String>,
pub reason: Option<String>,
pub attempts: Option<u32>,
pub recording_id: Option<String>,
pub recording: Option<Value>,
pub call_duration: Option<f64>,
pub created_at: Option<String>,
pub updated_at: Option<String>,
#[serde(flatten)]
pub extra: Map<String, Value>,
}
#[derive(Debug, Clone, Default)]
pub struct BatchCallStats {
pub batch_id: String,
pub status: Option<BatchCallStatus>,
pub total: Option<i64>,
pub counts: HashMap<String, i64>,
}
impl<'de> Deserialize<'de> for BatchCallStats {
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)
}
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct BatchPresignedUpload {
presigned_url: String,
batch_id: String,
}
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DeleteRecordsResult {
pub message: String,
#[serde(default, deserialize_with = "crate::common::null_to_default")]
pub deleted_count: u32,
}
#[derive(Default)]
pub struct UploadBatchCallParams {
pub file_path: String,
pub batch_id: Option<String>,
pub wait_for_parse: Option<bool>,
pub poll_interval: Option<Duration>,
pub poll_timeout: Option<Duration>,
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()
}
}
#[derive(Debug, Clone, Default, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CreateBatchCallParams {
pub batch_name: String,
pub phone_number_id: String,
pub routing_rule_id: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub description: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub timing: Option<BatchCallTiming>,
#[serde(skip_serializing_if = "Option::is_none")]
pub timezone: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub scheduled_time: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub scheduled_date: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub allowed_from: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub allowed_until: Option<String>,
#[serde(skip_serializing_if = "Vec::is_empty")]
pub days_of_week: Vec<u32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub retry: Option<BatchCallRetryConfig>,
#[serde(skip_serializing_if = "Option::is_none")]
pub webhook: Option<BatchCallWebhookConfig>,
#[serde(skip_serializing_if = "Option::is_none")]
pub batch_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub save_as_draft: Option<bool>,
}
#[derive(Debug, Clone, Default, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct UpdateBatchCallParams {
#[serde(skip_serializing_if = "Option::is_none")]
pub batch_name: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub phone_number_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub routing_rule_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub description: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub timing: Option<BatchCallTiming>,
#[serde(skip_serializing_if = "Option::is_none")]
pub timezone: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub scheduled_time: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub scheduled_date: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub allowed_from: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub allowed_until: Option<String>,
#[serde(skip_serializing_if = "Vec::is_empty")]
pub days_of_week: Vec<u32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub retry: Option<BatchCallRetryConfig>,
#[serde(skip_serializing_if = "Option::is_none")]
pub save_as_draft: Option<bool>,
}
#[derive(Debug, Clone, Default)]
pub struct ListBatchCallsParams {
pub page: Option<u32>,
pub per_page: Option<u32>,
pub cursor: Option<String>,
pub status: Option<BatchCallStatus>,
pub start_date: Option<String>,
pub end_date: Option<String>,
pub min_value: Option<u32>,
pub max_value: Option<u32>,
}
#[derive(Debug, Clone, Default)]
pub struct ListBatchCallRecordsParams {
pub page: Option<u32>,
pub per_page: Option<u32>,
pub cursor: Option<String>,
pub status: Option<String>,
pub search: Option<String>,
}
#[derive(Debug, Clone, Default, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct UpdateBatchCallRecordParams {
#[serde(skip_serializing_if = "Option::is_none")]
pub phone_number: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub metadata: Option<Map<String, Value>>,
}
#[derive(Debug, Clone, Default)]
pub struct ExportBatchCallRecordsParams {
pub status: Option<String>,
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);
#[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 }
}
pub fn records(&self) -> BatchCallRecordsResource<'a> {
BatchCallRecordsResource {
client: self.client,
}
}
pub async fn upload(&self, params: UploadBatchCallParams) -> Result<BatchCall> {
let path = Path::new(¶ms.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(¶ms.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) = ¶ms.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
}
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;
}
}
pub async fn create(&self, params: CreateBatchCallParams) -> Result<BatchCall> {
self.client
.json(Method::POST, PATH, CallOptions::json(¶ms)?)
.await
}
pub async fn list(&self, params: ListBatchCallsParams) -> Result<Page<BatchCall>> {
paginate(
self.fetcher(¶ms),
¶ms.pagination(),
"batchCalls",
None,
)
.await
}
pub fn list_stream(
&self,
params: ListBatchCallsParams,
) -> impl Stream<Item = Result<BatchCall>> + Send {
auto_page(
self.fetcher(¶ms),
params.pagination(),
"batchCalls",
None,
)
}
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
}
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(¶ms)?)
.await
}
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
}
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
}
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
})
})
}
}
#[derive(Debug, Clone, Copy)]
pub struct BatchCallRecordsResource<'a> {
client: &'a Client,
}
impl<'a> BatchCallRecordsResource<'a> {
pub async fn list(
&self,
batch_id: &str,
params: ListBatchCallRecordsParams,
) -> Result<Page<BatchCallRecord>> {
let fetcher = self.fetcher(batch_id, ¶ms);
paginate(fetcher, ¶ms.pagination(), "records", None).await
}
pub fn list_stream(
&self,
batch_id: &str,
params: ListBatchCallRecordsParams,
) -> impl Stream<Item = Result<BatchCallRecord>> + Send {
let fetcher = self.fetcher(batch_id, ¶ms);
auto_page(fetcher, params.pagination(), "records", None)
}
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(¶ms)?)
.await
}
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
}
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 {
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() {
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);
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);
}
}