headless_lms_server/programs/mailchimp_syncer/

mod.rs

use crate::config::program_config::ProgramConfig;
use crate::prelude::*;
use crate::setup_tracing;
use dotenvy::dotenv;
use headless_lms_models::marketing_consents::MarketingMailingListAccessToken;
use headless_lms_models::marketing_consents::UserEmailSubscription;
use headless_lms_models::marketing_consents::UserMarketingConsentWithDetails;
use headless_lms_utils::http::REQWEST_CLIENT;
use secrecy::ExposeSecret;
use serde_json::json;
use sqlx::{PgConnection, PgPool};
use std::{
    env,
    time::{Duration, Instant},
};
use uuid::Uuid;

mod batch_client;
mod mailchimp_ops;
mod policy_tags;

use batch_client::MAX_MAILCHIMP_BATCH_SIZE;
use mailchimp_ops::{MailchimpExecutor, MailchimpOperation};
use policy_tags::sync_policy_tags_for_users;
use reqwest::Method;

#[derive(Debug, Deserialize)]
struct MailchimpField {
    field_id: String,
    field_name: String,
}

#[derive(Debug)]
struct FieldSchema {
    tag: &'static str,
    name: &'static str,
    default_value: &'static str,
}

const REQUIRED_FIELDS: &[FieldSchema] = &[
    FieldSchema {
        tag: "FNAME",
        name: "First Name",
        default_value: "",
    },
    FieldSchema {
        tag: "LNAME",
        name: "Last Name",
        default_value: "",
    },
    FieldSchema {
        tag: "MARKETING",
        name: "Accepts Marketing",
        default_value: "disallowed",
    },
    FieldSchema {
        tag: "LOCALE",
        name: "Locale",
        default_value: "en",
    },
    FieldSchema {
        tag: "GRADUATED",
        name: "Graduated",
        default_value: "",
    },
    FieldSchema {
        tag: "COURSEID",
        name: "Course ID",
        default_value: "",
    },
    FieldSchema {
        tag: "LANGGRPID",
        name: "Course language Group ID",
        default_value: "",
    },
    FieldSchema {
        tag: "USERID",
        name: "User ID",
        default_value: "",
    },
    FieldSchema {
        tag: "RESEARCH",
        name: "Research consent",
        default_value: "false",
    },
];

/// These fields are excluded from removing all fields that are not in the schema
const FIELDS_EXCLUDED_FROM_REMOVING: &[&str] = &["PHONE", "PACE", "COUNTRY", "MMERGE9"];
const REMOVE_UNSUPPORTED_FIELDS: bool = false;
const PROCESS_UNSUBSCRIBES_INTERVAL_SECS: u64 = 10_800;

const SYNC_INTERVAL_SECS: u64 = 10;
const PRINT_STILL_RUNNING_MESSAGE_TICKS_THRESHOLD: u32 = 60;

const BATCH_POLL_INTERVAL_SECS: u64 = 10;
const BATCH_POLL_TIMEOUT_SECS: u64 = 300;

#[derive(Debug)]
struct SyncUser {
    details: UserMarketingConsentWithDetails,
    payload: serde_json::Value,
}

#[derive(Debug)]
struct EmailSyncResult {
    user_id: Uuid,
    user_mailchimp_id: String,
}

/// The main function that initializes environment variables, config, and sync process.
pub async fn main() -> anyhow::Result<()> {
    initialize_environment()?;

    let config = initialize_configuration().await?;

    let db_pool = initialize_database_pool(&config.database_url).await?;
    let mut conn = db_pool.acquire().await?;

    let mut interval = tokio::time::interval(Duration::from_secs(SYNC_INTERVAL_SECS));
    let mut ticks = 0;

    let access_tokens =
        headless_lms_models::marketing_consents::fetch_all_marketing_mailing_list_access_tokens(
            &mut conn,
        )
        .await?;

    // Iterate through access tokens and ensure Mailchimp schema is set up
    for token in &access_tokens {
        if let Err(e) = ensure_mailchimp_schema(
            &token.mailchimp_mailing_list_id,
            &token.server_prefix,
            &token.access_token,
        )
        .await
        {
            error!(
                "Failed to set up Mailchimp schema for list '{}': {:?}",
                token.mailchimp_mailing_list_id, e
            );
            return Err(e);
        }
    }

    info!("Starting mailchimp syncer (periodic reconciliation loop).");

    let mut last_time_unsubscribes_processed = Instant::now();
    let mut last_time_tags_synced = Instant::now();

    loop {
        interval.tick().await;
        ticks += 1;

        if ticks >= PRINT_STILL_RUNNING_MESSAGE_TICKS_THRESHOLD {
            ticks = 0;
            info!("Still syncing.");
        }
        let mut process_unsubscribes = false;
        if last_time_unsubscribes_processed.elapsed().as_secs()
            >= PROCESS_UNSUBSCRIBES_INTERVAL_SECS
        {
            process_unsubscribes = true;
            last_time_unsubscribes_processed = Instant::now();
        };

        // Check and sync tags for this access token once every hour
        if last_time_tags_synced.elapsed().as_secs() >= 3600 {
            info!("Stage: tag catalog sync (keep local tag metadata aligned with Mailchimp).");
            for token in &access_tokens {
                if let Err(e) = sync_tags_from_mailchimp(
                    &mut conn,
                    &token.mailchimp_mailing_list_id,
                    &token.access_token,
                    &token.server_prefix,
                    token.id,
                    token.course_language_group_id,
                )
                .await
                {
                    error!(
                        "Failed to sync tags for list '{}': {:?}",
                        token.mailchimp_mailing_list_id, e
                    );
                }
            }
            last_time_tags_synced = Instant::now();
        }

        if let Err(e) = sync_contacts(&mut conn, &config, process_unsubscribes).await {
            error!("Error during synchronization: {:?}", e);
            if let Ok(sqlx::Error::Io(..)) = e.downcast::<sqlx::Error>() {
                // this usually happens if the database is reset while running bin/dev etc.
                info!("syncer may have lost its connection to the db, trying to reconnect");
                conn = db_pool.acquire().await?;
            }
        }
    }
}

/// Initializes environment variables, logging, and tracing setup.
fn initialize_environment() -> anyhow::Result<()> {
    // TODO: Audit that the environment access only happens in single-threaded code.
    unsafe { env::set_var("RUST_LOG", "info,actix_web=info,sqlx=warn") };
    dotenv().ok();
    setup_tracing()?;
    Ok(())
}

/// Structure to hold the configuration settings, such as the database URL.
struct SyncerConfig {
    database_url: String,
}

/// Initializes and returns configuration settings (database URL).
async fn initialize_configuration() -> anyhow::Result<SyncerConfig> {
    let database_url = ProgramConfig::database_url_with_default();

    Ok(SyncerConfig { database_url })
}

/// Initializes the PostgreSQL connection pool from the provided database URL.
async fn initialize_database_pool(database_url: &str) -> anyhow::Result<PgPool> {
    PgPool::connect(database_url).await.map_err(|e| {
        anyhow::anyhow!(
            "Failed to connect to the database at {}: {:?}",
            database_url,
            e
        )
    })
}

/// Ensures the Mailchimp schema is up to date, adding required fields and removing any extra ones.
async fn ensure_mailchimp_schema(
    list_id: &str,
    server_prefix: &str,
    access_token: &DbSecret,
) -> anyhow::Result<()> {
    let existing_fields =
        fetch_current_mailchimp_fields(list_id, server_prefix, access_token).await?;

    if REMOVE_UNSUPPORTED_FIELDS {
        // Remove extra fields not in REQUIRED_FIELDS or FIELDS_EXCLUDED_FROM_REMOVING
        for field in existing_fields.iter() {
            if !REQUIRED_FIELDS
                .iter()
                .any(|r| r.tag == field.field_name.as_str())
                && !FIELDS_EXCLUDED_FROM_REMOVING.contains(&field.field_name.as_str())
            {
                match remove_field_from_mailchimp(
                    list_id,
                    &field.field_id,
                    server_prefix,
                    access_token,
                )
                .await
                {
                    Err(e) => {
                        warn!("Could not remove field '{}': {}", field.field_name, e);
                    }
                    _ => {
                        info!("Removed field '{}'", field.field_name);
                    }
                }
            }
        }
    }

    // Add any required fields that are missing
    for required_field in REQUIRED_FIELDS.iter() {
        if !existing_fields
            .iter()
            .any(|f| f.field_name == required_field.tag)
        {
            match add_field_to_mailchimp(list_id, required_field, server_prefix, access_token).await
            {
                Err(e) => {
                    warn!(
                        "Failed to add required field '{}': {}",
                        required_field.name, e
                    );
                }
                _ => {
                    info!(
                        "Successfully added required field '{}'",
                        required_field.name
                    );
                }
            }
        } else {
            info!(
                "Field '{}' already exists, skipping addition.",
                required_field.name
            );
        }
    }

    Ok(())
}

/// Fetches the current merge fields from the Mailchimp list schema.
async fn fetch_current_mailchimp_fields(
    list_id: &str,
    server_prefix: &str,
    access_token: &DbSecret,
) -> Result<Vec<MailchimpField>, anyhow::Error> {
    let url = format!(
        "https://{}.api.mailchimp.com/3.0/lists/{}/merge-fields",
        server_prefix, list_id
    );

    let response = REQWEST_CLIENT
        .get(&url)
        .header(
            "Authorization",
            format!("apikey {}", access_token.expose_secret()),
        )
        .send()
        .await?;

    if response.status().is_success() {
        let json = response.json::<serde_json::Value>().await?;

        let fields: Vec<MailchimpField> = json["merge_fields"]
            .as_array()
            .unwrap_or(&vec![])
            .iter()
            .filter_map(|field| {
                let field_id = field["merge_id"].as_u64();
                let field_name = field["tag"].as_str();

                if let (Some(field_id), Some(field_name)) = (field_id, field_name) {
                    Some(MailchimpField {
                        field_id: field_id.to_string(),
                        field_name: field_name.to_string(),
                    })
                } else {
                    None
                }
            })
            .collect();

        Ok(fields)
    } else {
        let error_text = response.text().await?;
        error!("Error fetching merge fields: {}", error_text);
        Err(anyhow::anyhow!("Failed to fetch current Mailchimp fields."))
    }
}

/// Adds a new merge field to the Mailchimp list.
async fn add_field_to_mailchimp(
    list_id: &str,
    field_schema: &FieldSchema,
    server_prefix: &str,
    access_token: &DbSecret,
) -> anyhow::Result<()> {
    let url = format!(
        "https://{}.api.mailchimp.com/3.0/lists/{}/merge-fields",
        server_prefix, list_id
    );

    let body = json!({
        "tag": field_schema.tag,
        "name": field_schema.name,
        "type": "text",
        "default_value": field_schema.default_value,
    });

    let response = REQWEST_CLIENT
        .post(&url)
        .header(
            "Authorization",
            format!("apikey {}", access_token.expose_secret()),
        )
        .json(&body)
        .send()
        .await?;

    if response.status().is_success() {
        Ok(())
    } else {
        let status = response.status();
        let error_text = response
            .text()
            .await
            .unwrap_or_else(|_| "No additional error info.".to_string());
        Err(anyhow::anyhow!(
            "Failed to add field to Mailchimp. Status: {}. Error: {}",
            status,
            error_text
        ))
    }
}

/// Removes a merge field from the Mailchimp list by with a field ID.
async fn remove_field_from_mailchimp(
    list_id: &str,
    field_id: &str,
    server_prefix: &str,
    access_token: &DbSecret,
) -> anyhow::Result<()> {
    let url = format!(
        "https://{}.api.mailchimp.com/3.0/lists/{}/merge-fields/{}",
        server_prefix, list_id, field_id
    );

    let response = REQWEST_CLIENT
        .delete(&url)
        .header(
            "Authorization",
            format!("apikey {}", access_token.expose_secret()),
        )
        .send()
        .await?;

    if response.status().is_success() {
        Ok(())
    } else {
        let status = response.status();
        let error_text = response
            .text()
            .await
            .unwrap_or_else(|_| "No additional error info.".to_string());
        Err(anyhow::anyhow!(
            "Failed to remove field from Mailchimp. Status: {}. Error: {}",
            status,
            error_text
        ))
    }
}

/// Fetch tags from mailchimp and sync the changes to the database
pub async fn sync_tags_from_mailchimp(
    conn: &mut PgConnection,
    list_id: &str,
    access_token: &DbSecret,
    server_prefix: &str,
    marketing_mailing_list_access_token_id: Uuid,
    course_language_group_id: Uuid,
) -> anyhow::Result<()> {
    let url = format!(
        "https://{}.api.mailchimp.com/3.0/lists/{}/tag-search",
        server_prefix, list_id
    );

    let response = REQWEST_CLIENT
        .get(&url)
        .header(
            "Authorization",
            format!("apikey {}", access_token.expose_secret()),
        )
        .send()
        .await?;

    // Extract tags from the response
    let response_json = response.json::<serde_json::Value>().await?;

    let mailchimp_tags = match response_json.get("tags") {
        Some(tags) if tags.is_array() => tags
            .as_array()
            .ok_or_else(|| anyhow::anyhow!("tags field is not an array despite is_array() check"))?
            .iter()
            .filter_map(|tag| {
                let name = tag.get("name")?.as_str()?.to_string();

                let id = match tag.get("id") {
                    Some(serde_json::Value::Number(num)) => num.to_string(),
                    Some(serde_json::Value::String(str_id)) => str_id.clone(),
                    _ => return None,
                };

                Some((id, name))
            })
            .collect::<Vec<(String, String)>>(),
        _ => {
            warn!("No tags found for list '{}', skipping sync.", list_id);
            return Ok(());
        }
    };

    // Fetch the current tags from the database
    let db_tags = headless_lms_models::marketing_consents::fetch_tags_with_course_language_group_id_and_marketing_mailing_list_access_token_id(
            conn,
            course_language_group_id,
            marketing_mailing_list_access_token_id,
        )
        .await?;

    // Check if any tags from Mailchimp are renamed
    for (tag_id, tag_name) in &mailchimp_tags {
        if let Some(db_tag) = db_tags.iter().find(|db_tag| {
            db_tag.get("id").and_then(|v| v.as_str()).map(|v| v.trim()) == Some(tag_id.trim())
        }) {
            let db_tag_name = db_tag
                .get("tag_name")
                .and_then(|v| v.as_str())
                .unwrap_or_default();
            if db_tag_name != tag_name {
                headless_lms_models::marketing_consents::upsert_tag(
                    conn,
                    course_language_group_id,
                    marketing_mailing_list_access_token_id,
                    tag_id.clone(),
                    tag_name.clone(),
                )
                .await?;
            }
        }
    }

    // Check if any tags in the database have been removed via Mailchimp
    for db_tag in db_tags.iter() {
        let db_tag_id = db_tag
            .get("id")
            .and_then(|v| v.as_str())
            .map(|v| v.trim().to_string())
            .unwrap_or_default();

        // Check if this tag exists in the Mailchimp tags
        if !mailchimp_tags
            .iter()
            .any(|(tag_id, _)| tag_id.trim() == db_tag_id.trim())
        {
            if db_tag_id.is_empty() {
                warn!("Skipping tag deletion due to missing ID: {:?}", db_tag);
                continue;
            }
            headless_lms_models::marketing_consents::delete_tag(
                conn,
                db_tag_id.clone(),
                course_language_group_id,
            )
            .await?;
        }
    }

    Ok(())
}

/// Synchronizes the user contacts with Mailchimp.
/// Added a boolean flag to determine whether to process unsubscribes.
async fn sync_contacts(
    conn: &mut PgConnection,
    _config: &SyncerConfig,
    process_unsubscribes: bool,
) -> anyhow::Result<()> {
    let access_tokens =
        headless_lms_models::marketing_consents::fetch_all_marketing_mailing_list_access_tokens(
            conn,
        )
        .await?;

    let mut successfully_synced_user_ids = Vec::new();

    // Iterate through tokens and fetch and send user details to Mailchimp
    for token in access_tokens {
        let course_language_group_slug =
            match headless_lms_models::course_language_groups::get_slug_by_id(
                conn,
                token.course_language_group_id,
            )
            .await
            {
                Ok(Some(s)) => Some(s),
                Ok(None) => None,
                Err(e) => {
                    error!(
                        course_language_group_id = %token.course_language_group_id,
                        "Failed to get course language group slug: {:?}",
                        e
                    );
                    return Err(e.into());
                }
            };

        // Fetch all users from Mailchimp and sync possible changes locally
        if process_unsubscribes {
            info!(
                "Stage: unsubscribe sync (apply Mailchimp compliance/unsubscribe changes locally)."
            );
            let mailchimp_data = fetch_unsubscribed_users_from_mailchimp_in_chunks(
                &token.mailchimp_mailing_list_id,
                &token.server_prefix,
                &token.access_token,
                1000,
            )
            .await?;

            info!(
                "Processing Mailchimp data for list: {}",
                token.mailchimp_mailing_list_id
            );

            process_unsubscribed_users_from_mailchimp(conn, mailchimp_data).await?;
        }

        // Fetch unsynced emails and update them in Mailchimp
        let users_with_unsynced_emails =
            headless_lms_models::marketing_consents::fetch_all_unsynced_updated_emails(
                conn,
                token.course_language_group_id,
            )
            .await?;

        info!(
            "Stage: email updates (ensure member identifiers stay correct). Found {} unsynced user email(s) for course language group: {}",
            users_with_unsynced_emails.len(),
            token.course_language_group_id
        );

        if !users_with_unsynced_emails.is_empty() {
            let email_sync_results = update_emails_in_mailchimp(
                users_with_unsynced_emails,
                &token.mailchimp_mailing_list_id,
                &token.server_prefix,
                &token.access_token,
            )
            .await?;

            let email_synced_user_ids: Vec<Uuid> =
                email_sync_results.iter().map(|r| r.user_id).collect();
            successfully_synced_user_ids.extend(email_synced_user_ids.clone());

            if !email_sync_results.is_empty() {
                let mailchimp_id_by_user: std::collections::HashMap<Uuid, String> =
                    email_sync_results
                        .iter()
                        .map(|r| (r.user_id, r.user_mailchimp_id.clone()))
                        .collect();
                let user_details =
                    headless_lms_models::marketing_consents::fetch_user_marketing_consents_with_details_by_user_ids(
                        conn,
                        token.course_language_group_id,
                        &email_synced_user_ids,
                    )
                    .await?;
                match sync_policy_tags_for_users(
                    &token,
                    &user_details,
                    &mailchimp_id_by_user,
                    Duration::from_secs(BATCH_POLL_TIMEOUT_SECS),
                    Duration::from_secs(BATCH_POLL_INTERVAL_SECS),
                )
                .await
                {
                    Ok(results) => {
                        let (ok, failed): (Vec<_>, Vec<_>) =
                            results.into_iter().partition(|r| r.success);
                        if !failed.is_empty() {
                            let sample: Vec<String> = failed
                                .iter()
                                .take(5)
                                .map(|r| {
                                    format!(
                                        "{}: {}",
                                        r.user_id,
                                        r.error.as_deref().unwrap_or("unknown error")
                                    )
                                })
                                .collect();
                            warn!(
                                "Policy tag sync after email updates for list '{}' had {} failure(s) out of {}. Sample: {:?}",
                                token.mailchimp_mailing_list_id,
                                failed.len(),
                                ok.len() + failed.len(),
                                sample
                            );
                        }
                    }
                    Err(e) => {
                        error!(
                            "Failed to sync policy tags after email updates for list '{}': {:?}",
                            token.mailchimp_mailing_list_id, e
                        );
                    }
                }
            }
        }

        let tag_objects = headless_lms_models::marketing_consents::fetch_tags_with_course_language_group_id_and_marketing_mailing_list_access_token_id(conn, token.course_language_group_id, token.id).await?;

        // Fetch unsynced user consents and update them in Mailchimp
        let unsynced_users_details =
            headless_lms_models::marketing_consents::fetch_all_unsynced_user_marketing_consents_by_course_language_group_id(
                conn,
                token.course_language_group_id,
            )
            .await?;

        info!(
            "Stage: member upsert (merge fields + consent). Found {} unsynced user consent(s) for course language group: {}",
            unsynced_users_details.len(),
            token.course_language_group_id
        );

        if !unsynced_users_details.is_empty() {
            let consent_synced_user_ids =
                send_users_to_mailchimp(conn, &token, &unsynced_users_details, tag_objects).await?;

            if let Some(ref slug) = course_language_group_slug
                && !consent_synced_user_ids.is_empty()
            {
                let mailchimp_id_mapping =
                    headless_lms_models::marketing_consents::fetch_user_mailchimp_id_mapping(
                        conn,
                        token.course_language_group_id,
                        &consent_synced_user_ids,
                    )
                    .await?;
                if let Err(e) = sync_completed_tag_for_members(
                    &unsynced_users_details,
                    &consent_synced_user_ids,
                    &mailchimp_id_mapping,
                    slug,
                    &token,
                )
                .await
                {
                    error!(
                        "Failed to sync completed tag for list '{}': {:?}",
                        token.mailchimp_mailing_list_id, e
                    );
                }
            }

            // Store the successfully synced user IDs from syncing user consents
            successfully_synced_user_ids.extend(consent_synced_user_ids);
        }
    }

    // If there are any successfully synced users, update the database to mark them as synced
    if !successfully_synced_user_ids.is_empty() {
        match headless_lms_models::marketing_consents::update_synced_to_mailchimp_at_to_all_synced_users(
        conn,
        &successfully_synced_user_ids,
    )
    .await
    {
        Ok(_) => {
            info!(
                "Stage: mark synced (avoid repeat work). Successfully updated synced status for {} users.",
                successfully_synced_user_ids.len()
            );
        }
        Err(e) => {
            error!(
                "Failed to update synced status for {} users: {:?}",
                successfully_synced_user_ids.len(),
                e
            );
        }
    }
    }

    Ok(())
}

/// Sends a batch of users to Mailchimp for synchronization.
pub async fn send_users_to_mailchimp(
    conn: &mut PgConnection,
    token: &MarketingMailingListAccessToken,
    users_details: &[UserMarketingConsentWithDetails],
    tag_objects: Vec<serde_json::Value>,
) -> anyhow::Result<Vec<Uuid>> {
    let mut users_to_sync = vec![];
    let mut sent_user_ids = Vec::new();
    let mut successfully_synced_user_ids = Vec::new();
    let mut user_id_contact_id_pairs = Vec::new();

    // Prepare each user's data for Mailchimp
    for user in users_details {
        // Check user has given permission to send data to mailchimp
        if let Some(ref subscription) = user.email_subscription_in_mailchimp
            && subscription == "subscribed"
        {
            sent_user_ids.push(user.user_id);
            let user_details = json!({
                "email_address": user.email,
                "status": user.email_subscription_in_mailchimp,
                "merge_fields": {
                    "FNAME": user.first_name.clone().unwrap_or("".to_string()),
                    "LNAME": user.last_name.clone().unwrap_or("".to_string()),
                    "MARKETING": if user.consent { "allowed" } else { "disallowed" },
                    "LOCALE": user.locale,
                    "GRADUATED": user.completed_course_at.map(|cca| cca.to_rfc3339()).unwrap_or("".to_string()),
                    "USERID": user.user_id,
                    "COURSEID": user.course_id,
                    "LANGGRPID": user.course_language_group_id,
                    "RESEARCH" : if user.research_consent.unwrap_or(false) { "allowed" } else { "disallowed" },
                    "COUNTRY" : user.country.clone().unwrap_or("".to_string()),
                },
               "tags": tag_objects.iter().map(|tag| tag["name"].clone()).collect::<Vec<_>>()
            });
            users_to_sync.push(SyncUser {
                details: user.clone(),
                payload: user_details,
            });
        }
    }

    if users_to_sync.is_empty() {
        info!("No new users to sync.");
        return Ok(vec![]);
    }

    let url = format!(
        "https://{}.api.mailchimp.com/3.0/lists/{}",
        token.server_prefix, token.mailchimp_mailing_list_id
    );

    let total_chunks = users_to_sync.len().div_ceil(MAX_MAILCHIMP_BATCH_SIZE);
    info!(
        "Syncing {} members to list '{}' in {} chunk(s)",
        users_to_sync.len(),
        token.mailchimp_mailing_list_id,
        total_chunks
    );

    for (chunk_index, chunk) in users_to_sync.chunks(MAX_MAILCHIMP_BATCH_SIZE).enumerate() {
        info!(
            "Syncing users chunk {}/{} ({} members)",
            chunk_index + 1,
            total_chunks,
            chunk.len()
        );
        let chunk_members: Vec<serde_json::Value> =
            chunk.iter().map(|user| user.payload.clone()).collect();
        let batch_request = json!({
            "members": chunk_members,
            "update_existing": true
        });

        let response = REQWEST_CLIENT
            .post(&url)
            .header("Content-Type", "application/json")
            .header(
                "Authorization",
                format!("apikey {}", token.access_token.expose_secret()),
            )
            .json(&batch_request)
            .send()
            .await?;

        if !response.status().is_success() {
            let status = response.status();
            let error_text = response.text().await?;
            return Err(anyhow::anyhow!(
                "Error syncing users to Mailchimp. Status: {}. Error: {}",
                status,
                error_text
            ));
        }

        let response_data: serde_json::Value = response.json().await?;
        let mut chunk_contact_count = 0;
        let mut chunk_mailchimp_id_by_user: std::collections::HashMap<Uuid, String> =
            std::collections::HashMap::new();
        for user in chunk {
            if let Some(ref mailchimp_id) = user.details.user_mailchimp_id {
                chunk_mailchimp_id_by_user
                    .entry(user.details.user_id)
                    .or_insert_with(|| mailchimp_id.clone());
            }
        }
        for key in &["new_members", "updated_members"] {
            if let Some(members) = response_data[key].as_array() {
                for member in members {
                    if let Some(user_id) = member["merge_fields"]["USERID"].as_str() {
                        if let Ok(uuid) = uuid::Uuid::parse_str(user_id) {
                            successfully_synced_user_ids.push(uuid);
                        }
                        if let Some(contact_id) = member["contact_id"].as_str() {
                            user_id_contact_id_pairs
                                .push((user_id.to_string(), contact_id.to_string()));
                            if let Ok(uuid) = uuid::Uuid::parse_str(user_id) {
                                chunk_mailchimp_id_by_user.insert(uuid, contact_id.to_string());
                            }
                            chunk_contact_count += 1;
                        }
                    }
                }
            }
        }
        if let Some(errors) = response_data["errors"].as_array()
            && !errors.is_empty()
        {
            let sample: Vec<String> = errors
                .iter()
                .take(5)
                .map(|e| {
                    let email = e
                        .get("email_address")
                        .and_then(|v| v.as_str())
                        .unwrap_or("?");
                    let msg = e.get("error").and_then(|v| v.as_str()).unwrap_or_else(|| {
                        e.get("message").and_then(|v| v.as_str()).unwrap_or("?")
                    });
                    format!("{}: {}", email, msg)
                })
                .collect();
            warn!(
                "Mailchimp batch subscribe chunk {}/{} returned {} error(s) (e.g. unsubscribed/rejected). Sample: {:?}",
                chunk_index + 1,
                total_chunks,
                errors.len(),
                sample
            );
        }
        info!(
            "Chunk {}/{}: {} contact_id(s) from new_members/updated_members",
            chunk_index + 1,
            total_chunks,
            chunk_contact_count
        );

        let chunk_users: Vec<UserMarketingConsentWithDetails> =
            chunk.iter().map(|user| user.details.clone()).collect();
        match sync_policy_tags_for_users(
            token,
            &chunk_users,
            &chunk_mailchimp_id_by_user,
            Duration::from_secs(BATCH_POLL_TIMEOUT_SECS),
            Duration::from_secs(BATCH_POLL_INTERVAL_SECS),
        )
        .await
        {
            Ok(results) => {
                let (ok, failed): (Vec<_>, Vec<_>) = results.into_iter().partition(|r| r.success);
                if !failed.is_empty() {
                    let sample: Vec<String> = failed
                        .iter()
                        .take(5)
                        .map(|r| {
                            format!(
                                "{}: {}",
                                r.user_id,
                                r.error.as_deref().unwrap_or("unknown error")
                            )
                        })
                        .collect();
                    warn!(
                        "Policy tag sync for list '{}' chunk {}/{} had {} failure(s) out of {}. Sample: {:?}",
                        token.mailchimp_mailing_list_id,
                        chunk_index + 1,
                        total_chunks,
                        failed.len(),
                        ok.len() + failed.len(),
                        sample
                    );
                }
            }
            Err(e) => {
                error!(
                    "Failed to sync policy tags for list '{}' chunk {}/{}: {:?}",
                    token.mailchimp_mailing_list_id,
                    chunk_index + 1,
                    total_chunks,
                    e
                );
            }
        }
    }

    let got_contact_id_set: std::collections::HashSet<Uuid> = user_id_contact_id_pairs
        .iter()
        .filter_map(|(uid, _)| Uuid::parse_str(uid).ok())
        .collect();
    let no_contact_id_user_ids: Vec<Uuid> = sent_user_ids
        .iter()
        .filter(|id| !got_contact_id_set.contains(id))
        .copied()
        .collect();

    if !no_contact_id_user_ids.is_empty() {
        let sample_len = no_contact_id_user_ids.len().min(10);
        warn!(
            "Mailchimp did not return contact_id for {} member(s) (likely unsubscribed, removed, or rejected). Marking synced_to_mailchimp_at to stop retry. First {} user_id(s): {:?}",
            no_contact_id_user_ids.len(),
            sample_len,
            &no_contact_id_user_ids[..sample_len]
        );
        if let Err(e) = headless_lms_models::marketing_consents::update_synced_to_mailchimp_at_to_all_synced_users(
            conn,
            &no_contact_id_user_ids,
        )
        .await
        {
            error!(
                "Failed to update synced_to_mailchimp_at for no-contact_id users: {:?}",
                e
            );
        }
    }

    info!(
        "Batch subscribe list '{}': sent {} member(s), got {} contact_id(s){}",
        token.mailchimp_mailing_list_id,
        sent_user_ids.len(),
        user_id_contact_id_pairs.len(),
        if no_contact_id_user_ids.is_empty() {
            String::new()
        } else {
            format!(
                ", {} without contact_id (marked synced to stop retry)",
                no_contact_id_user_ids.len()
            )
        }
    );

    if !user_id_contact_id_pairs.is_empty() {
        headless_lms_models::marketing_consents::update_user_mailchimp_id_at_to_all_synced_users(
            conn,
            user_id_contact_id_pairs,
        )
        .await?;
    }

    Ok(successfully_synced_user_ids)
}

/// Sets or removes the "{slug}-completed" tag on Mailchimp members based on course completion. Skips users without user_mailchimp_id.
async fn sync_completed_tag_for_members(
    users_details: &[UserMarketingConsentWithDetails],
    successfully_synced_user_ids: &[Uuid],
    mailchimp_id_by_user: &std::collections::HashMap<Uuid, String>,
    slug: &str,
    token: &MarketingMailingListAccessToken,
) -> anyhow::Result<()> {
    let tag_name = format!("{}-completed", slug);
    let success_set: std::collections::HashSet<_> = successfully_synced_user_ids.iter().collect();

    let mut operations = Vec::new();
    for user in users_details {
        if !success_set.contains(&user.user_id) {
            continue;
        }
        let Some(user_mailchimp_id) = mailchimp_id_by_user.get(&user.user_id) else {
            continue;
        };
        let status = if user.completed_course_at.is_some() {
            "active"
        } else {
            "inactive"
        };
        operations.push(MailchimpOperation {
            method: Method::POST,
            path: format!(
                "/lists/{}/members/{}/tags",
                token.mailchimp_mailing_list_id, user_mailchimp_id
            ),
            body: Some(json!({
                "tags": [
                    { "name": tag_name, "status": status }
                ]
            })),
            operation_id: Some(user.user_id.to_string()),
        });
    }

    if operations.is_empty() {
        return Ok(());
    }

    let ops_count = operations.len();
    info!(
        "Stage: completion tags (mark course completion). Preparing {} operation(s) for list '{}'",
        ops_count, token.mailchimp_mailing_list_id
    );

    let timeout = Duration::from_secs(BATCH_POLL_TIMEOUT_SECS);
    let poll_interval = Duration::from_secs(BATCH_POLL_INTERVAL_SECS);
    let executor = MailchimpExecutor::new(timeout, poll_interval);

    let start_time = Instant::now();
    let results = executor.execute(token, operations).await?;
    let failures: Vec<_> = results.iter().filter(|r| !r.is_success()).collect();
    if !failures.is_empty() {
        let sample: Vec<String> = failures
            .iter()
            .take(5)
            .map(|r| {
                format!(
                    "{}: {}",
                    r.operation_id.as_deref().unwrap_or("?"),
                    r.error.as_deref().unwrap_or("unknown error")
                )
            })
            .collect();
        warn!(
            "Completion tag sync for list '{}' had {} failure(s) out of {}. Sample: {:?}",
            token.mailchimp_mailing_list_id,
            failures.len(),
            results.len(),
            sample
        );
        return Err(anyhow::anyhow!("completion tag sync failed"));
    }

    info!(
        "Completed sync of {} completion tags for list '{}' in {:.2}s",
        ops_count,
        token.mailchimp_mailing_list_id,
        start_time.elapsed().as_secs_f64()
    );

    Ok(())
}

/// Updates the email addresses of multiple users in a Mailchimp mailing list.
async fn update_emails_in_mailchimp(
    users: Vec<UserEmailSubscription>,
    list_id: &str,
    server_prefix: &str,
    access_token: &DbSecret,
) -> anyhow::Result<Vec<EmailSyncResult>> {
    let mut successfully_synced_users = Vec::new();
    let mut failed_user_ids = Vec::new();

    for user in users {
        if let Some(ref user_mailchimp_id) = user.user_mailchimp_id {
            if let Some(ref status) = user.email_subscription_in_mailchimp
                && status != "subscribed"
            {
                continue; // Skip this user if they are not subscribed because Mailchimp only updates emails that are subscribed
            }

            let url = format!(
                "https://{}.api.mailchimp.com/3.0/lists/{}/members/{}",
                server_prefix, list_id, user_mailchimp_id
            );

            // Prepare the body for the PUT request
            let body = serde_json::json!({
                "email_address": &user.email,
                "status": &user.email_subscription_in_mailchimp,
            });

            // Update the email
            let update_response = REQWEST_CLIENT
                .put(&url)
                .header(
                    "Authorization",
                    format!("apikey {}", access_token.expose_secret()),
                )
                .json(&body)
                .send()
                .await?;

            if update_response.status().is_success() {
                successfully_synced_users.push(EmailSyncResult {
                    user_id: user.user_id,
                    user_mailchimp_id: user_mailchimp_id.clone(),
                });
            } else {
                failed_user_ids.push(user.user_id);
            }
        } else {
            continue;
        }
    }

    if !failed_user_ids.is_empty() {
        info!("Failed to update the following users:");
        for user_id in &failed_user_ids {
            error!("User ID: {}", user_id);
        }
    }

    Ok(successfully_synced_users)
}

/// Fetches data from Mailchimp in chunks.
async fn fetch_unsubscribed_users_from_mailchimp_in_chunks(
    list_id: &str,
    server_prefix: &str,
    access_token: &DbSecret,
    chunk_size: usize,
) -> anyhow::Result<Vec<(String, String, String, String)>> {
    let mut all_data = Vec::new();
    let mut offset = 0;

    loop {
        let url = format!(
            "https://{}.api.mailchimp.com/3.0/lists/{}/members?offset={}&count={}&fields=members.merge_fields,members.status,members.last_changed&status=unsubscribed,non-subscribed",
            server_prefix, list_id, offset, chunk_size
        );

        let response = REQWEST_CLIENT
            .get(&url)
            .header(
                "Authorization",
                format!("apikey {}", access_token.expose_secret()),
            )
            .send()
            .await?
            .json::<serde_json::Value>()
            .await?;

        let empty_vec = vec![];
        let members = response["members"].as_array().unwrap_or(&empty_vec);
        if members.is_empty() {
            break;
        }

        for member in members {
            // Process the member, but only if necessary fields are present and valid
            if let (Some(status), Some(last_changed), Some(merge_fields)) = (
                member["status"].as_str(),
                member["last_changed"].as_str(),
                member["merge_fields"].as_object(),
            ) {
                // Ensure both USERID and LANGGRPID are present and valid
                if let (Some(user_id), Some(language_group_id)) = (
                    merge_fields.get("USERID").and_then(|v| v.as_str()),
                    merge_fields.get("LANGGRPID").and_then(|v| v.as_str()),
                ) {
                    // Avoid adding data if any field is missing or empty
                    if !user_id.is_empty() && !language_group_id.is_empty() {
                        all_data.push((
                            user_id.to_string(),
                            last_changed.to_string(),
                            language_group_id.to_string(),
                            status.to_string(),
                        ));
                    }
                }
            }
        }

        // Check the pagination info from the response
        let total_items = response["total_items"].as_u64().unwrap_or(0) as usize;
        if offset + chunk_size >= total_items {
            break;
        }

        offset += chunk_size;
    }

    Ok(all_data)
}

const BATCH_SIZE: usize = 1000;

async fn process_unsubscribed_users_from_mailchimp(
    conn: &mut PgConnection,
    mailchimp_data: Vec<(String, String, String, String)>,
) -> anyhow::Result<()> {
    let total_records = mailchimp_data.len();
    let total_chunks = total_records.div_ceil(BATCH_SIZE);

    for (chunk_num, chunk) in mailchimp_data.chunks(BATCH_SIZE).enumerate() {
        if chunk.is_empty() {
            continue;
        }

        if let Err(e) = headless_lms_models::marketing_consents::update_unsubscribed_users_from_mailchimp_in_bulk(
            conn,
            chunk.to_vec(),
        )
        .await
        {
            error!(
                "Error while processing chunk {}/{}: {}",
                chunk_num + 1,
                total_chunks,
                e
            );
        }
    }

    Ok(())
}