//! 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`, 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
}

/// 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.
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;
        }
    });
}

#[cfg(test)]
mod tests {
    //! Operation labels. These reach logs and support tickets for creator
    //! subscription changes, so a label that says the opposite of what happened
    //! sends an investigation the wrong way.

    use super::*;

    #[test]
    fn setting_and_clearing_cancel_at_period_end_do_not_share_a_label() {
        assert_eq!(
            FanSubOp::CancelAtPeriodEnd(true).label(),
            "set_cancel_at_period_end"
        );
        assert_eq!(
            FanSubOp::CancelAtPeriodEnd(false).label(),
            "clear_cancel_at_period_end"
        );
    }

    #[test]
    fn every_operation_has_a_distinct_label() {
        let labels = [
            FanSubOp::CancelAtPeriodEnd(true).label(),
            FanSubOp::CancelAtPeriodEnd(false).label(),
            FanSubOp::Pause.label(),
            FanSubOp::Resume.label(),
            FanSubOp::Cancel.label(),
        ];
        let unique: std::collections::HashSet<_> = labels.iter().collect();
        assert_eq!(unique.len(), labels.len(), "labels collide: {labels:?}");
    }

    // ── run_fan_sub_fanout ──
    //
    // The returned count is the whole contract: `mnw-admin` prints it, and the
    // background twin turns the same count into a WAM ticket. A fan-out that
    // reports zero failures when ops failed leaves fans charging against a
    // creator state that no longer matches, with nothing to reconcile from.

    use crate::payments::test_provider::ScriptedProvider;

    /// Two handles on one provider: the fan-out takes the trait object, and the
    /// assertions read the recorded calls off the concrete type.
    fn provider(p: ScriptedProvider) -> (Arc<ScriptedProvider>, Arc<dyn PaymentProvider>) {
        let scripted = Arc::new(p);
        let stripe: Arc<dyn PaymentProvider> = scripted.clone();
        (scripted, stripe)
    }

    fn account() -> StripeAccountId {
        StripeAccountId::new("acct_test1234567890").expect("a well-formed connected account id")
    }

    fn subs(ids: &[&str]) -> Vec<String> {
        ids.iter().map(|s| (*s).to_owned()).collect()
    }

    #[tokio::test]
    async fn applies_the_op_to_every_subscription_in_order() {
        let (scripted, stripe) = provider(ScriptedProvider::healthy());
        let failed = run_fan_sub_fanout(
            &stripe,
            &account(),
            &subs(&["sub_a", "sub_b", "sub_c"]),
            FanSubOp::Pause,
        )
        .await;

        assert_eq!(failed, 0);
        assert_eq!(
            scripted.calls(),
            vec![
                ("pause", "sub_a".to_string()),
                ("pause", "sub_b".to_string()),
                ("pause", "sub_c".to_string()),
            ],
            "every subscription gets the op, and none is skipped"
        );
    }

    #[tokio::test]
    async fn counts_the_failures_rather_than_reporting_success() {
        let (_scripted, stripe) = provider(ScriptedProvider::failing(["sub_b", "sub_d"]));
        let failed = run_fan_sub_fanout(
            &stripe,
            &account(),
            &subs(&["sub_a", "sub_b", "sub_c", "sub_d"]),
            FanSubOp::Cancel,
        )
        .await;

        assert_eq!(
            failed, 2,
            "two ops failed; a count of 0 or 1 is a dropped cancel nobody reconciles"
        );
    }

    #[tokio::test]
    async fn a_failure_does_not_abort_the_rest() {
        let (scripted, stripe) = provider(ScriptedProvider::failing(["sub_a"]));
        let failed = run_fan_sub_fanout(
            &stripe,
            &account(),
            &subs(&["sub_a", "sub_b", "sub_c"]),
            FanSubOp::Resume,
        )
        .await;

        assert_eq!(failed, 1);
        assert_eq!(
            scripted.calls().len(),
            3,
            "the first subscription failing must not strand the two behind it"
        );
    }

    #[tokio::test]
    async fn every_subscription_failing_is_counted_as_every_subscription() {
        let (_scripted, stripe) = provider(ScriptedProvider::failing(["sub_a", "sub_b"]));
        let failed = run_fan_sub_fanout(
            &stripe,
            &account(),
            &subs(&["sub_a", "sub_b"]),
            FanSubOp::Pause,
        )
        .await;

        assert_eq!(failed, 2, "a saturating or decrementing counter shows here");
    }

    #[tokio::test]
    async fn an_empty_list_is_no_calls_and_no_failures() {
        let (scripted, stripe) = provider(ScriptedProvider::healthy());
        let failed = run_fan_sub_fanout(&stripe, &account(), &[], FanSubOp::Cancel).await;

        assert_eq!(failed, 0);
        assert!(scripted.calls().is_empty());
    }

    #[tokio::test]
    async fn cancel_at_period_end_carries_the_flag_it_was_given() {
        for (cancel, expected) in [
            (true, "set_cancel_at_period_end"),
            (false, "clear_cancel_at_period_end"),
        ] {
            let (scripted, stripe) = provider(ScriptedProvider::healthy());
            run_fan_sub_fanout(
                &stripe,
                &account(),
                &subs(&["sub_a"]),
                FanSubOp::CancelAtPeriodEnd(cancel),
            )
            .await;

            assert_eq!(
                scripted.calls(),
                vec![(expected, "sub_a".to_string())],
                "CancelAtPeriodEnd({cancel}) must reach Stripe as {expected}"
            );
        }
    }

    // ── spawn_fan_sub_fanout ──
    //
    // The background twin returns before doing anything, so the only thing a
    // test can hold it to is that the work arrives. That is also the whole risk:
    // a fan-out that silently never runs looks identical to one that succeeded,
    // on a path whose failure is fans still being charged.

    /// Wait for `want` calls to land, or give up. The queue is drained by a
    /// separate task, so there is nothing to await on directly.
    async fn settle(scripted: &Arc<ScriptedProvider>, want: usize) -> Vec<(&'static str, String)> {
        for _ in 0..200 {
            let calls = scripted.calls();
            if calls.len() >= want {
                return calls;
            }
            tokio::time::sleep(std::time::Duration::from_millis(5)).await;
        }
        scripted.calls()
    }

    #[tokio::test]
    async fn the_queued_fan_out_actually_reaches_stripe() {
        let (scripted, stripe) = provider(ScriptedProvider::healthy());
        let bg = crate::background::spawn_pool_detached();

        spawn_fan_sub_fanout(
            &bg,
            stripe,
            account(),
            subs(&["sub_a", "sub_b"]),
            FanSubOp::Cancel,
            None,
        );

        assert_eq!(
            settle(&scripted, 2).await,
            vec![
                ("cancel", "sub_a".to_string()),
                ("cancel", "sub_b".to_string()),
            ],
            "the ops must arrive; a fan-out that returns and does nothing is invisible"
        );
    }

    #[tokio::test]
    async fn a_failing_op_does_not_strand_the_queued_remainder() {
        let (scripted, stripe) = provider(ScriptedProvider::failing(["sub_a"]));
        let bg = crate::background::spawn_pool_detached();

        spawn_fan_sub_fanout(
            &bg,
            stripe,
            account(),
            subs(&["sub_a", "sub_b", "sub_c"]),
            FanSubOp::Pause,
            None,
        );

        assert_eq!(settle(&scripted, 3).await.len(), 3);
    }

    #[tokio::test]
    async fn an_empty_queued_fan_out_asks_stripe_for_nothing() {
        let (scripted, stripe) = provider(ScriptedProvider::healthy());
        let bg = crate::background::spawn_pool_detached();

        spawn_fan_sub_fanout(&bg, stripe, account(), Vec::new(), FanSubOp::Resume, None);

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        assert!(scripted.calls().is_empty());
    }
}