use super::{CreatorTier, DbCreatorSubscription, PgPool, Result, UserId};

/// Create or reactivate a creator tier subscription record.
///
/// Uses ON CONFLICT DO UPDATE on the user_id unique index to handle
/// both duplicate webhooks and re-subscription after cancellation.
/// Returns `None` if the row already existed with the same stripe_subscription_id
/// (duplicate webhook), `Some` if this was a fresh insert or a re-subscription
/// with a different subscription ID.
#[tracing::instrument(skip_all)]
pub async fn create_creator_subscription<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    user_id: UserId,
    stripe_subscription_id: &str,
    stripe_customer_id: &str,
    tier: CreatorTier,
) -> Result<Option<DbCreatorSubscription>> {
    // Use WHERE clause on the DO UPDATE to only update if the subscription_id
    // is different (new subscription) or status is not already active.
    // When the WHERE fails, DO UPDATE becomes a no-op and RETURNING yields no row.
    let sub = sqlx::query_as::<_, DbCreatorSubscription>(
        r"
        INSERT INTO creator_subscriptions (user_id, stripe_subscription_id, stripe_customer_id, tier)
        VALUES ($1, $2, $3, $4)
        ON CONFLICT (user_id) DO UPDATE
        SET stripe_subscription_id = EXCLUDED.stripe_subscription_id,
            stripe_customer_id = EXCLUDED.stripe_customer_id,
            tier = EXCLUDED.tier,
            status = 'active',
            canceled_at = NULL,
            grace_enforced_at = NULL
        WHERE creator_subscriptions.stripe_subscription_id != EXCLUDED.stripe_subscription_id
           OR creator_subscriptions.status != 'active'
        RETURNING *
        ",
    )
    .bind(user_id)
    .bind(stripe_subscription_id)
    .bind(stripe_customer_id)
    .bind(tier)
    .fetch_optional(executor)
    .await?;

    Ok(sub)
}

/// Look up a creator subscription by its Stripe subscription ID.
#[tracing::instrument(skip_all)]
pub async fn get_creator_sub_by_stripe_id(
    pool: &PgPool,
    stripe_subscription_id: &str,
) -> Result<Option<DbCreatorSubscription>> {
    let sub = sqlx::query_as::<_, DbCreatorSubscription>(
        "SELECT * FROM creator_subscriptions WHERE stripe_subscription_id = $1",
    )
    .bind(stripe_subscription_id)
    .fetch_optional(pool)
    .await?;

    Ok(sub)
}

/// Get a user's creator subscription (any status).
#[tracing::instrument(skip_all)]
pub async fn get_creator_sub_by_user(
    pool: &PgPool,
    user_id: UserId,
) -> Result<Option<DbCreatorSubscription>> {
    let sub = sqlx::query_as::<_, DbCreatorSubscription>(
        "SELECT * FROM creator_subscriptions WHERE user_id = $1",
    )
    .bind(user_id)
    .fetch_optional(pool)
    .await?;

    Ok(sub)
}

/// Get the active creator tier for a user (None if no active subscription).
#[tracing::instrument(skip_all)]
pub async fn get_active_creator_tier(
    pool: &PgPool,
    user_id: UserId,
) -> Result<Option<CreatorTier>> {
    let tier = sqlx::query_scalar::<_, String>(
        "SELECT tier FROM creator_subscriptions WHERE user_id = $1 AND status = 'active'",
    )
    .bind(user_id)
    .fetch_optional(pool)
    .await?;

    match tier {
        None => Ok(None),
        // A present-but-unparseable tier means the DB holds a tier string the
        // enum doesn't know (enum drift). Silently mapping that to `None` would
        // strip a paying creator's entitlements with no signal, surface it
        // loudly instead of swallowing it (Run 21). The `enum_drift` test and the
        // DB CHECK constraint make this unreachable in practice.
        Some(t) => match t.parse::<CreatorTier>() {
            Ok(parsed) => Ok(Some(parsed)),
            Err(_) => {
                tracing::error!(
                    user_id = %user_id, tier = %t,
                    "active creator subscription has an unrecognized tier string (enum drift)"
                );
                Err(crate::error::AppError::Internal(anyhow::anyhow!(
                    "unrecognized creator tier '{t}' for user {user_id}"
                )))
            }
        },
    }
}

// Apply a Stripe-driven status and/or period update in one guarded statement.
// `canceled` is terminal; reactivation runs through `create_creator_subscription`'s
// `ON CONFLICT (user_id) DO UPDATE` at checkout, never through this path. Replaces
// the old split status/period setters (the period half lacked the guard). See
// `crate::db::subscription_writer`.
crate::db::subscription_writer::define_stripe_subscription_writer!(
    apply_stripe_update,
    "creator_subscriptions",
    DbCreatorSubscription
);

/// Cancel a creator subscription (set status + canceled_at).
#[tracing::instrument(skip_all)]
pub async fn cancel_creator_sub(
    pool: &PgPool,
    stripe_subscription_id: &str,
) -> Result<Option<DbCreatorSubscription>> {
    let sub = sqlx::query_as::<_, DbCreatorSubscription>(
        r"
        UPDATE creator_subscriptions
        SET status = 'canceled', canceled_at = COALESCE(canceled_at, NOW())
        WHERE stripe_subscription_id = $1
        RETURNING *
        ",
    )
    .bind(stripe_subscription_id)
    .fetch_optional(pool)
    .await?;

    Ok(sub)
}

/// Sync the users.creator_tier column from the subscription status.
/// Called after checkout/update/cancel to keep the denormalized column in sync.
#[tracing::instrument(skip_all)]
pub async fn sync_user_creator_tier(pool: &PgPool, user_id: UserId) -> Result<()> {
    sqlx::query(
        r"
        UPDATE users SET creator_tier = (
            SELECT tier FROM creator_subscriptions
            WHERE user_id = $1 AND status = 'active'
        )
        WHERE id = $1
        ",
    )
    .bind(user_id)
    .execute(pool)
    .await?;

    Ok(())
}

/// Get user IDs of creators with canceled subscriptions 30+ days ago
/// whose items have not yet been hidden.
#[tracing::instrument(skip_all)]
pub async fn get_expired_grace_creators(pool: &PgPool) -> Result<Vec<UserId>> {
    // Bounded batch per call. The scheduler enforces these inline on the tick
    // (two DB round-trips per creator), so an unbounded result set would let a
    // backlog stall the tick. `grace_enforced_at` is set as each creator is
    // processed, so successive ticks drain the rest; ORDER BY oldest-first keeps
    // it deterministic and starvation-free.
    let ids: Vec<UserId> = sqlx::query_scalar(
        r"
        SELECT user_id FROM creator_subscriptions
        WHERE status = 'canceled'
          AND canceled_at IS NOT NULL
          AND canceled_at < NOW() - INTERVAL '30 days'
          AND grace_enforced_at IS NULL
        ORDER BY canceled_at ASC
        LIMIT 200
        ",
    )
    .fetch_all(pool)
    .await?;

    Ok(ids)
}

/// Stamp `grace_enforced_at` for all given creators in one statement, paired with
/// `items::hide_all_items_for_users` on the post-grace sweep (Perf-S4, Run 9).
/// No-op on an empty slice.
#[tracing::instrument(skip_all)]
pub async fn mark_grace_enforced_batch(pool: &PgPool, user_ids: &[UserId]) -> Result<()> {
    if user_ids.is_empty() {
        return Ok(());
    }
    sqlx::query(
        "UPDATE creator_subscriptions SET grace_enforced_at = NOW() WHERE user_id = ANY($1)",
    )
    .bind(user_ids)
    .execute(pool)
    .await?;

    Ok(())
}

/// Count fully-paying creators, `status = 'active'` only.
///
/// Excludes trialing (free trial), past_due (payment failed but not yet
/// canceled), canceled-in-grace (winding down), and incomplete states.
/// This is the number that goes on the runway disclosure as "paying
/// creators today": revenue-bearing seats, no fudge.
#[tracing::instrument(skip_all)]
pub async fn count_active_paying(pool: &PgPool) -> Result<i64> {
    let count: (i64,) =
        sqlx::query_as("SELECT COUNT(*) FROM creator_subscriptions WHERE status = 'active'")
            .fetch_one(pool)
            .await?;
    Ok(count.0)
}

/// Count creators in a trial or 30-day cancellation grace period.
///
/// These are not revenue-bearing today but represent the near-term
/// pipeline: trialing seats may convert, grace seats may resubscribe
/// before enforcement. Disclosed as a secondary number on the runway
/// surface so the headline `count_active_paying` stays strict.
#[tracing::instrument(skip_all)]
pub async fn count_trialing_or_grace(pool: &PgPool) -> Result<i64> {
    let count: (i64,) = sqlx::query_as(
        r"
        SELECT COUNT(*) FROM creator_subscriptions
        WHERE status = 'trialing'
           OR (
                status = 'canceled'
                AND canceled_at IS NOT NULL
                AND canceled_at > NOW() - INTERVAL '30 days'
                AND grace_enforced_at IS NULL
              )
        ",
    )
    .fetch_one(pool)
    .await?;
    Ok(count.0)
}

/// Check whether a user is in the 30-day cancellation grace period.
///
/// Returns `true` if the subscription is canceled but within 30 days of cancellation
/// and enforcement has not yet been applied.
#[tracing::instrument(skip_all)]
pub async fn is_in_grace_period(pool: &PgPool, user_id: UserId) -> Result<bool> {
    let in_grace: bool = sqlx::query_scalar(
        r"
        SELECT EXISTS(
            SELECT 1 FROM creator_subscriptions
            WHERE user_id = $1
              AND status = 'canceled'
              AND canceled_at IS NOT NULL
              AND canceled_at > NOW() - INTERVAL '30 days'
              AND grace_enforced_at IS NULL
        )
        ",
    )
    .bind(user_id)
    .fetch_one(pool)
    .await?;

    Ok(in_grace)
}