//! Background fan-out of a Stripe subscription operation across all of a
//! creator's fan subscriptions.
//!
//! Pausing, suspending, or terminating a creator requires one Stripe call per
//! fan subscription. Run inline on a request or webhook handler, that serial
//! fan-out ties the hot path up for the duration of N Stripe round-trips,
//! minutes for a large creator, and on the webhook path long enough to trip
//! Stripe's delivery timeout and trigger a retry storm that re-runs the loop
//! (ultra-fuzz Run 6 S1/S2/R6-Perf-M1). This module moves the loop onto the
//! bounded background queue so the handler returns immediately. The operations
//! are fire-and-forget, errors were only logged inline, and still are.

use std::sync::Arc;

use crate::background::BackgroundTx;
use crate::db::StripeAccountId;
use crate::payments::PaymentProvider;

/// Which Stripe subscription operation to apply to each fan subscription.
#[derive(Clone, Copy, Debug)]
pub enum FanSubOp {
    /// Set (`true`) or clear (`false`) cancel_at_period_end, creator pause/unpause.
    CancelAtPeriodEnd(bool),
    /// Pause collection, admin suspend.
    Pause,
    /// Resume collection, admin unsuspend.
    Resume,
    /// Cancel outright, admin terminate.
    Cancel,
}

impl FanSubOp {
    fn label(self) -> &'static str {
        match self {
            FanSubOp::CancelAtPeriodEnd(true) => "set_cancel_at_period_end",
            FanSubOp::CancelAtPeriodEnd(false) => "clear_cancel_at_period_end",
            FanSubOp::Pause => "pause",
            FanSubOp::Resume => "resume",
            FanSubOp::Cancel => "cancel",
        }
    }

    async fn apply(
        self,
        stripe: &Arc<dyn PaymentProvider>,
        sub_id: &str,
        account_id: &str,
    ) -> crate::error::Result<()> {
        match self {
            FanSubOp::CancelAtPeriodEnd(c) => {
                stripe.set_cancel_at_period_end(sub_id, account_id, c).await
            }
            FanSubOp::Pause => stripe.pause_subscription(sub_id, account_id).await,
            FanSubOp::Resume => stripe.resume_subscription(sub_id, account_id).await,
            FanSubOp::Cancel => stripe.cancel_subscription(sub_id, account_id).await,
        }
    }
}

/// Apply `op` to every subscription in `sub_ids` on the background queue. Returns
/// immediately; the loop runs off the request/webhook hot path. Per-subscription
/// failures are logged and do not abort the rest. No-op for an empty list.
///
/// A failed op is not merely logged: any failures open a WAM ticket (when `wam`
/// is configured) so a dropped pause/suspend/cancel, a fan still charged after a
/// creator pause, or a suspended creator's fans retaining access, is actively
/// surfaced for manual reconciliation rather than lost in the logs (audit Run 13
/// Resilience: fan-out had no dead-letter). The failed subscription IDs are
/// listed in the ticket body.
/// Apply `op` to every subscription in `sub_ids`, awaiting each call inline.
///
/// For contexts with no background queue, the `mnw-admin` CLI, where blocking on
/// the serial Stripe fan-out is fine because there is no request or webhook hot
/// path to protect. Per-subscription failures are logged and do not abort the
/// rest; returns the number that failed so the caller can surface it. No-op (and
/// returns 0) for an empty list.
pub async fn run_fan_sub_fanout(
    stripe: &Arc<dyn PaymentProvider>,
    account_id: &StripeAccountId,
    sub_ids: &[String],
    op: FanSubOp,
) -> usize {
    let mut failed = 0usize;
    for sub_id in sub_ids {
        if let Err(e) = op.apply(stripe, sub_id, account_id.as_str()).await {
            failed += 1;
            tracing::warn!(stripe_sub_id = %sub_id, op = op.label(), error = ?e, "fan subscription op failed (inline)");
        }
    }
    if failed > 0 {
        tracing::warn!(
            total = sub_ids.len(),
            failed,
            op = op.label(),
            "inline fan subscription fan-out completed with failures"
        );
    }
    failed
}

pub fn spawn_fan_sub_fanout(
    bg: &BackgroundTx,
    stripe: Arc<dyn PaymentProvider>,
    account_id: StripeAccountId,
    sub_ids: Vec<String>,
    op: FanSubOp,
    wam: Option<crate::wam_client::WamClient>,
) {
    if sub_ids.is_empty() {
        return;
    }
    bg.spawn("fan-sub stripe fan-out", async move {
        let total = sub_ids.len();
        let mut failed_ids: Vec<&str> = Vec::new();
        for sub_id in &sub_ids {
            if let Err(e) = op.apply(&stripe, sub_id, account_id.as_str()).await {
                failed_ids.push(sub_id);
                tracing::warn!(stripe_sub_id = %sub_id, op = op.label(), error = ?e, "fan subscription op failed");
            }
        }
        if failed_ids.is_empty() {
            tracing::debug!(total, op = op.label(), "fan subscription fan-out completed");
            return;
        }
        let failed = failed_ids.len();
        tracing::warn!(total, failed, op = op.label(), "fan subscription fan-out completed with failures");
        if let Some(wam) = wam {
            let title = format!(
                "Fan subscription fan-out incomplete: {} of {total} '{}' ops failed",
                failed,
                op.label()
            );
            let body = format!(
                "Applying '{}' to a creator's fan subscriptions (account {}) left {failed} of {total} \
                 unreconciled. These subscriptions may still be charging (or retaining access) against \
                 the creator's current state and need manual reconciliation in Stripe.\n\nFailed subscription IDs:\n{}",
                op.label(),
                account_id.as_str(),
                failed_ids.join("\n"),
            );
            wam.create_ticket(&title, Some(&body), "high", "fan-fanout-incomplete", Some(account_id.as_str())).await;
        }
    });
}