headless_lms_server/programs/mailchimp_syncer/
mailchimp_ops.rs

1use headless_lms_models::marketing_consents::MarketingMailingListAccessToken;
2use headless_lms_utils::http::REQWEST_CLIENT;
3use reqwest::Method;
4use secrecy::ExposeSecret;
5use serde_json::Value;
6use std::collections::HashMap;
7use std::time::Duration;
8
9use super::BATCH_RESULT_DOWNLOAD_TIMEOUT_SECS;
10use super::batch_client::{
11    BatchOperation, parse_batch_results_from_tar_gz, parse_operation_response_value,
12    poll_batch_until_finished_with_response_url, submit_batch,
13};
14
15const DEFAULT_BATCH_THRESHOLD: usize = 5;
16
17#[derive(Debug, Clone)]
18pub struct MailchimpOperation {
19    pub method: Method,
20    pub path: String,
21    pub body: Option<Value>,
22    pub operation_id: Option<String>,
23}
24
25#[derive(Debug, Clone)]
26pub struct MailchimpOpResult {
27    pub operation_id: Option<String>,
28    pub status_code: u16,
29    pub response_raw: Value,
30    pub response_json: Option<Value>,
31    pub error: Option<String>,
32}
33
34impl MailchimpOpResult {
35    pub fn is_success(&self) -> bool {
36        (200..=299).contains(&self.status_code) && self.error.is_none()
37    }
38}
39
40pub struct MailchimpExecutor {
41    batch_threshold: usize,
42    timeout: Duration,
43    poll_interval: Duration,
44}
45
46impl MailchimpExecutor {
47    /// Executes Mailchimp operations via direct calls or the /batches API.
48    pub fn new(timeout: Duration, poll_interval: Duration) -> Self {
49        Self {
50            batch_threshold: DEFAULT_BATCH_THRESHOLD,
51            timeout,
52            poll_interval,
53        }
54    }
55
56    pub async fn execute(
57        &self,
58        token: &MarketingMailingListAccessToken,
59        ops: Vec<MailchimpOperation>,
60    ) -> anyhow::Result<Vec<MailchimpOpResult>> {
61        if ops.is_empty() {
62            return Ok(vec![]);
63        }
64        if ops.len() < self.batch_threshold {
65            Ok(execute_direct(token, ops).await)
66        } else {
67            execute_batch(token, ops, self.timeout, self.poll_interval).await
68        }
69    }
70}
71
72async fn execute_direct(
73    token: &MarketingMailingListAccessToken,
74    ops: Vec<MailchimpOperation>,
75) -> Vec<MailchimpOpResult> {
76    let mut results = Vec::with_capacity(ops.len());
77    for op in ops {
78        let url = format!(
79            "https://{}.api.mailchimp.com/3.0{}",
80            token.server_prefix, op.path
81        );
82        let mut request = REQWEST_CLIENT.request(op.method.clone(), &url).header(
83            "Authorization",
84            format!("apikey {}", token.access_token.expose_secret()),
85        );
86        if let Some(body) = op.body.clone() {
87            request = request.json(&body);
88        }
89        let response = match request.send().await {
90            Ok(resp) => resp,
91            Err(err) => {
92                results.push(MailchimpOpResult {
93                    operation_id: op.operation_id.clone(),
94                    status_code: 0,
95                    response_raw: Value::Null,
96                    response_json: None,
97                    error: Some(format!("Transport error: {}", err)),
98                });
99                continue;
100            }
101        };
102
103        let status_code = response.status().as_u16();
104        let body_bytes = match response.bytes().await {
105            Ok(bytes) => bytes,
106            Err(err) => {
107                results.push(MailchimpOpResult {
108                    operation_id: op.operation_id.clone(),
109                    status_code,
110                    response_raw: Value::Null,
111                    response_json: None,
112                    error: Some(format!("Body read error: {}", err)),
113                });
114                continue;
115            }
116        };
117
118        let response_raw = match serde_json::from_slice::<Value>(&body_bytes) {
119            Ok(json) => json,
120            Err(_) => Value::String(String::from_utf8_lossy(&body_bytes).to_string()),
121        };
122        let response_json = parse_operation_response_value(&response_raw);
123        results.push(MailchimpOpResult {
124            operation_id: op.operation_id.clone(),
125            status_code,
126            response_raw,
127            response_json,
128            error: None,
129        });
130    }
131    results
132}
133
134async fn execute_batch(
135    token: &MarketingMailingListAccessToken,
136    ops: Vec<MailchimpOperation>,
137    timeout: Duration,
138    poll_interval: Duration,
139) -> anyhow::Result<Vec<MailchimpOpResult>> {
140    if ops.is_empty() {
141        return Ok(vec![]);
142    }
143
144    let batch_ops: Vec<BatchOperation> = ops
145        .iter()
146        .map(|op| BatchOperation {
147            method: op.method.as_str().to_string(),
148            path: op.path.clone(),
149            body: op.body.as_ref().map(|b| b.to_string()).unwrap_or_default(),
150            operation_id: op.operation_id.clone(),
151        })
152        .collect();
153
154    let batch_ids = submit_batch(&token.server_prefix, &token.access_token, batch_ops).await?;
155
156    let mut results = Vec::new();
157    for batch_id in &batch_ids {
158        let poll_result = poll_batch_until_finished_with_response_url(
159            &token.server_prefix,
160            &token.access_token,
161            batch_id,
162            timeout,
163            poll_interval,
164        )
165        .await?;
166        if poll_result.total_operations > 0 {
167            info!(
168                "Mailchimp batch {} finished with {} errored operations out of {}",
169                batch_id, poll_result.errored_operations, poll_result.total_operations
170            );
171        }
172        let response_url = match poll_result.response_body_url {
173            Some(url) => url,
174            None => {
175                return Err(anyhow::anyhow!(
176                    "Mailchimp batch {} finished without response_body_url",
177                    batch_id
178                ));
179            }
180        };
181        let response = REQWEST_CLIENT
182            .get(&response_url)
183            .timeout(Duration::from_secs(BATCH_RESULT_DOWNLOAD_TIMEOUT_SECS))
184            .send()
185            .await?;
186        if !response.status().is_success() {
187            let error_text = response.text().await.unwrap_or_default();
188            return Err(anyhow::anyhow!(
189                "Failed to fetch Mailchimp batch results: {}",
190                error_text
191            ));
192        }
193        let bytes = response.bytes().await?;
194        let batch_results = parse_batch_results_from_tar_gz(&bytes)?;
195        for result in batch_results {
196            let response_json = parse_operation_response_value(&result.response);
197            results.push(MailchimpOpResult {
198                operation_id: result.operation_id.clone(),
199                status_code: result.status_code,
200                response_raw: result.response,
201                response_json,
202                error: None,
203            });
204        }
205    }
206
207    Ok(order_results(&ops, results))
208}
209
210fn order_results(
211    ops: &[MailchimpOperation],
212    results: Vec<MailchimpOpResult>,
213) -> Vec<MailchimpOpResult> {
214    let mut ordered = Vec::new();
215    let mut results_by_id: HashMap<String, Vec<MailchimpOpResult>> = HashMap::new();
216    let mut unmatched = Vec::new();
217
218    for result in results {
219        if let Some(ref op_id) = result.operation_id {
220            results_by_id.entry(op_id.clone()).or_default().push(result);
221        } else {
222            unmatched.push(result);
223        }
224    }
225
226    for op in ops {
227        if let Some(ref op_id) = op.operation_id
228            && let Some(mut entries) = results_by_id.remove(op_id)
229        {
230            ordered.append(&mut entries);
231        }
232    }
233
234    for (_, mut entries) in results_by_id {
235        ordered.append(&mut entries);
236    }
237
238    ordered.extend(unmatched);
239    ordered
240}
241
242#[cfg(test)]
243mod tests {
244    use super::*;
245
246    #[test]
247    fn order_results_respects_input_order() {
248        let ops = vec![
249            MailchimpOperation {
250                method: Method::GET,
251                path: "/a".to_string(),
252                body: None,
253                operation_id: Some("u1".to_string()),
254            },
255            MailchimpOperation {
256                method: Method::GET,
257                path: "/b".to_string(),
258                body: None,
259                operation_id: Some("u2".to_string()),
260            },
261        ];
262        let results = vec![
263            MailchimpOpResult {
264                operation_id: Some("u2".to_string()),
265                status_code: 200,
266                response_raw: Value::Null,
267                response_json: None,
268                error: None,
269            },
270            MailchimpOpResult {
271                operation_id: Some("u1".to_string()),
272                status_code: 200,
273                response_raw: Value::Null,
274                response_json: None,
275                error: None,
276            },
277        ];
278        let ordered = order_results(&ops, results);
279        assert_eq!(ordered[0].operation_id.as_deref(), Some("u1"));
280        assert_eq!(ordered[1].operation_id.as_deref(), Some("u2"));
281    }
282}
headless_lms_server/programs/mailchimp_syncer/mailchimp_ops.rs

headless_lms_server/programs/mailchimp_syncer/
mailchimp_ops.rs
