//! Subscription queries: tier CRUD, subscription lifecycle, and access control.

use sqlx::PgPool;

use super::models::{
    DbSubscription, DbSubscriptionTier, DbUserSubscriptionRow, SubscriberExportRow,
    SubscriptionExportRow,
};
use super::{ItemId, PriceCents, ProjectId, SubscriptionId, SubscriptionTierId, UserId};
use crate::error::Result;

// ── Tier CRUD ──

/// Create a new subscription tier for a project.
#[tracing::instrument(skip_all)]
pub(crate) async fn create_subscription_tier(
    pool: &PgPool,
    project_id: ProjectId,
    name: &str,
    description: Option<&str>,
    price_cents: PriceCents,
) -> Result<DbSubscriptionTier> {
    let tier = sqlx::query_as!(
        DbSubscriptionTier,
        r#"
        INSERT INTO subscription_tiers (project_id, name, description, price_cents)
        VALUES ($1, $2, $3, $4)
        RETURNING id AS "id: SubscriptionTierId", project_id AS "project_id: ProjectId",
                  name, description, price_cents, stripe_product_id, stripe_price_id,
                  sort_order, is_active,
                  created_at AS "created_at: chrono::DateTime<chrono::Utc>",
                  updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
                  item_id AS "item_id: ItemId"
        "#,
        project_id as ProjectId,
        name,
        description,
        price_cents.as_i32(),
    )
    .fetch_one(pool)
    .await?;

    Ok(tier)
}

#[tracing::instrument(skip_all)]
pub(crate) async fn get_subscription_tier_by_id(
    pool: &PgPool,
    id: SubscriptionTierId,
) -> Result<Option<DbSubscriptionTier>> {
    let tier = sqlx::query_as!(
        DbSubscriptionTier,
        r#"
        SELECT id AS "id: SubscriptionTierId", project_id AS "project_id: ProjectId",
               name, description, price_cents, stripe_product_id, stripe_price_id,
               sort_order, is_active,
               created_at AS "created_at: chrono::DateTime<chrono::Utc>",
               updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
               item_id AS "item_id: ItemId"
        FROM subscription_tiers WHERE id = $1
        "#,
        id as SubscriptionTierId,
    )
    .fetch_optional(pool)
    .await?;

    Ok(tier)
}

/// Get all active tiers for a project, ordered by sort_order.
#[tracing::instrument(skip_all)]
pub(crate) async fn get_active_tiers_by_project(
    pool: &PgPool,
    project_id: ProjectId,
) -> Result<Vec<DbSubscriptionTier>> {
    let tiers = sqlx::query_as!(
        DbSubscriptionTier,
        r#"
        SELECT id AS "id: SubscriptionTierId", project_id AS "project_id: ProjectId",
               name, description, price_cents, stripe_product_id, stripe_price_id,
               sort_order, is_active,
               created_at AS "created_at: chrono::DateTime<chrono::Utc>",
               updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
               item_id AS "item_id: ItemId"
        FROM subscription_tiers WHERE project_id = $1 AND is_active = true ORDER BY sort_order, created_at
        "#,
        project_id as ProjectId,
    )
    .fetch_all(pool)
    .await?;

    Ok(tiers)
}

/// Get all tiers for a project (active and inactive), for dashboard management.
#[tracing::instrument(skip_all)]
pub(crate) async fn get_all_tiers_by_project(
    pool: &PgPool,
    project_id: ProjectId,
) -> Result<Vec<DbSubscriptionTier>> {
    let tiers = sqlx::query_as!(
        DbSubscriptionTier,
        r#"
        SELECT id AS "id: SubscriptionTierId", project_id AS "project_id: ProjectId",
               name, description, price_cents, stripe_product_id, stripe_price_id,
               sort_order, is_active,
               created_at AS "created_at: chrono::DateTime<chrono::Utc>",
               updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
               item_id AS "item_id: ItemId"
        FROM subscription_tiers WHERE project_id = $1 ORDER BY sort_order, created_at
        "#,
        project_id as ProjectId,
    )
    .fetch_all(pool)
    .await?;

    Ok(tiers)
}

/// Update a subscription tier's name, description, and active status.
#[tracing::instrument(skip_all)]
pub(crate) async fn update_subscription_tier(
    pool: &PgPool,
    id: SubscriptionTierId,
    name: &str,
    description: Option<&str>,
    is_active: bool,
) -> Result<DbSubscriptionTier> {
    let tier = sqlx::query_as!(
        DbSubscriptionTier,
        r#"
        UPDATE subscription_tiers
        SET name = $2, description = $3, is_active = $4
        WHERE id = $1
        RETURNING id AS "id: SubscriptionTierId", project_id AS "project_id: ProjectId",
                  name, description, price_cents, stripe_product_id, stripe_price_id,
                  sort_order, is_active,
                  created_at AS "created_at: chrono::DateTime<chrono::Utc>",
                  updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
                  item_id AS "item_id: ItemId"
        "#,
        id as SubscriptionTierId,
        name,
        description,
        is_active,
    )
    .fetch_one(pool)
    .await?;

    Ok(tier)
}

/// Store Stripe product and price IDs on a tier after creating them on connected account.
#[tracing::instrument(skip_all)]
pub(crate) async fn update_tier_stripe_ids(
    pool: &PgPool,
    tier_id: SubscriptionTierId,
    product_id: &str,
    price_id: &str,
) -> Result<()> {
    sqlx::query!(
        r#"
        UPDATE subscription_tiers
        SET stripe_product_id = $2, stripe_price_id = $3
        WHERE id = $1
        "#,
        tier_id as SubscriptionTierId,
        product_id,
        price_id,
    )
    .execute(pool)
    .await?;

    Ok(())
}

/// Delete a subscription tier. Soft-deletes (sets is_active=false) if any
/// subscriptions reference it; hard-deletes otherwise.
///
/// Uses a transaction with FOR UPDATE to prevent a TOCTOU race where a
/// subscription could be created between the existence check and the delete.
#[tracing::instrument(skip_all)]
pub(crate) async fn delete_subscription_tier(pool: &PgPool, id: SubscriptionTierId) -> Result<()> {
    let mut tx = pool.begin().await?;

    // Lock the tier row to serialize against concurrent subscription creation
    sqlx::query!(
        "SELECT id FROM subscription_tiers WHERE id = $1 FOR UPDATE",
        id as SubscriptionTierId
    )
    .fetch_optional(&mut *tx)
    .await?
    .ok_or(sqlx::Error::RowNotFound)?;

    let has_subscriptions: bool = sqlx::query_scalar!(
        r#"SELECT EXISTS(SELECT 1 FROM subscriptions WHERE tier_id = $1) AS "exists!""#,
        id as SubscriptionTierId,
    )
    .fetch_one(&mut *tx)
    .await?;

    if has_subscriptions {
        sqlx::query!(
            "UPDATE subscription_tiers SET is_active = false WHERE id = $1",
            id as SubscriptionTierId
        )
        .execute(&mut *tx)
        .await?;
    } else {
        sqlx::query!(
            "DELETE FROM subscription_tiers WHERE id = $1",
            id as SubscriptionTierId
        )
        .execute(&mut *tx)
        .await?;
    }

    tx.commit().await?;
    Ok(())
}

// ── Subscription lifecycle ──

/// Create a new subscription record after successful checkout.
///
/// Returns `None` if the subscription already exists (duplicate webhook
/// or concurrent active subscription for the same user+project).
/// The partial UNIQUE index on `(subscriber_id, project_id) WHERE status = 'active'`
/// prevents multiple *active* subscriptions at the DB level.
///
/// Single-live-row invariant: before inserting, any lingering non-active-but-live
/// row (`past_due`/`trialing`/`incomplete`) for the same `(subscriber, project)` is
/// canceled, so a resubscribe over a stale row can't leave two live rows. The access
/// gate (`SubscriptionGate::PREDICATE`) already ignores those statuses, so this is
/// data hygiene, not an access fix. The cleanup is co-located here because this is the
/// only INSERT path into `subscriptions`; widening the partial-unique predicate instead
/// would collide with `idx_subscriptions_unique` (the `(subscriber, project,
/// stripe_subscription_id)` triple) and break the `ON CONFLICT DO NOTHING`
/// duplicate-webhook idempotency this path relies on. Takes `&mut PgConnection` (not a
/// one-shot executor) so the cleanup and the insert run on the caller's transaction.
#[tracing::instrument(skip_all)]
pub(crate) async fn create_subscription(
    conn: &mut sqlx::PgConnection,
    subscriber_id: UserId,
    tier_id: SubscriptionTierId,
    project_id: ProjectId,
    stripe_subscription_id: &str,
    stripe_customer_id: &str,
) -> Result<Option<DbSubscription>> {
    sqlx::query!(
        r#"
        UPDATE subscriptions
        SET status = 'canceled', canceled_at = NOW(), updated_at = NOW()
        WHERE subscriber_id = $1 AND project_id = $2 AND item_id IS NULL
          AND status IN ('past_due', 'trialing', 'incomplete')
        "#,
        subscriber_id as UserId,
        project_id as ProjectId,
    )
    .execute(&mut *conn)
    .await?;

    let sub = sqlx::query_as!(
        DbSubscription,
        r#"
        INSERT INTO subscriptions (subscriber_id, tier_id, project_id, stripe_subscription_id, stripe_customer_id)
        VALUES ($1, $2, $3, $4, $5)
        ON CONFLICT DO NOTHING
        RETURNING id AS "id: SubscriptionId", subscriber_id AS "subscriber_id: UserId",
                  tier_id AS "tier_id: SubscriptionTierId", project_id AS "project_id: ProjectId",
                  stripe_subscription_id, stripe_customer_id,
                  status AS "status: super::SubscriptionStatus",
                  current_period_start AS "current_period_start: chrono::DateTime<chrono::Utc>",
                  current_period_end AS "current_period_end: chrono::DateTime<chrono::Utc>",
                  canceled_at AS "canceled_at: chrono::DateTime<chrono::Utc>",
                  created_at AS "created_at: chrono::DateTime<chrono::Utc>",
                  updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
                  item_id AS "item_id: ItemId",
                  paused_at AS "paused_at: chrono::DateTime<chrono::Utc>"
        "#,
        subscriber_id as UserId,
        tier_id as SubscriptionTierId,
        project_id as ProjectId,
        stripe_subscription_id,
        stripe_customer_id,
    )
    .fetch_optional(&mut *conn)
    .await?;

    Ok(sub)
}

/// Look up a subscription by its Stripe subscription ID.
#[tracing::instrument(skip_all)]
pub(crate) async fn get_subscription_by_stripe_id(
    pool: &PgPool,
    stripe_sub_id: &str,
) -> Result<Option<DbSubscription>> {
    let sub = sqlx::query_as!(
        DbSubscription,
        r#"
        SELECT id AS "id: SubscriptionId", subscriber_id AS "subscriber_id: UserId",
               tier_id AS "tier_id: SubscriptionTierId", project_id AS "project_id: ProjectId",
               stripe_subscription_id, stripe_customer_id,
               status AS "status: super::SubscriptionStatus",
               current_period_start AS "current_period_start: chrono::DateTime<chrono::Utc>",
               current_period_end AS "current_period_end: chrono::DateTime<chrono::Utc>",
               canceled_at AS "canceled_at: chrono::DateTime<chrono::Utc>",
               created_at AS "created_at: chrono::DateTime<chrono::Utc>",
               updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
               item_id AS "item_id: ItemId",
               paused_at AS "paused_at: chrono::DateTime<chrono::Utc>"
        FROM subscriptions WHERE stripe_subscription_id = $1
        "#,
        stripe_sub_id,
    )
    .fetch_optional(pool)
    .await?;

    Ok(sub)
}

// Apply a Stripe-driven status and/or period update in one guarded statement.
// `canceled` is terminal, an out-of-order `updated`(active) or `invoice.paid`
// landing after a `deleted` cannot revive the row (status) nor refresh its
// period, because both columns are written here under the single guard. The
// old split `update_subscription_status` + `update_subscription_period` (whose
// period half lacked the guard) are replaced by this; reactivation only ever
// happens at checkout via `create_subscription`, never through this path.
crate::db::subscription_writer::define_stripe_subscription_writer!(
    apply_stripe_update,
    "subscriptions",
    DbSubscription
);

/// Mark a subscription as canceled.
#[tracing::instrument(skip_all)]
pub(crate) async fn cancel_subscription(
    pool: &PgPool,
    stripe_sub_id: &str,
) -> Result<Option<DbSubscription>> {
    let sub = sqlx::query_as!(
        DbSubscription,
        r#"
        UPDATE subscriptions
        SET status = 'canceled', canceled_at = COALESCE(canceled_at, NOW())
        WHERE stripe_subscription_id = $1
        RETURNING id AS "id: SubscriptionId", subscriber_id AS "subscriber_id: UserId",
                  tier_id AS "tier_id: SubscriptionTierId", project_id AS "project_id: ProjectId",
                  stripe_subscription_id, stripe_customer_id,
                  status AS "status: super::SubscriptionStatus",
                  current_period_start AS "current_period_start: chrono::DateTime<chrono::Utc>",
                  current_period_end AS "current_period_end: chrono::DateTime<chrono::Utc>",
                  canceled_at AS "canceled_at: chrono::DateTime<chrono::Utc>",
                  created_at AS "created_at: chrono::DateTime<chrono::Utc>",
                  updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
                  item_id AS "item_id: ItemId",
                  paused_at AS "paused_at: chrono::DateTime<chrono::Utc>"
        "#,
        stripe_sub_id,
    )
    .fetch_optional(pool)
    .await?;

    Ok(sub)
}

// ── Suspension pause/resume ──

/// Get all active subscriptions to a creator's projects (for pausing on suspension).
#[tracing::instrument(skip_all)]
pub(crate) async fn get_active_subscriptions_by_creator(
    pool: &PgPool,
    creator_id: UserId,
) -> Result<Vec<DbSubscription>> {
    let subs = sqlx::query_as!(
        DbSubscription,
        r#"
        SELECT s.id AS "id: SubscriptionId", s.subscriber_id AS "subscriber_id: UserId",
               s.tier_id AS "tier_id: SubscriptionTierId", s.project_id AS "project_id: ProjectId",
               s.stripe_subscription_id, s.stripe_customer_id,
               s.status AS "status: super::SubscriptionStatus",
               s.current_period_start AS "current_period_start: chrono::DateTime<chrono::Utc>",
               s.current_period_end AS "current_period_end: chrono::DateTime<chrono::Utc>",
               s.canceled_at AS "canceled_at: chrono::DateTime<chrono::Utc>",
               s.created_at AS "created_at: chrono::DateTime<chrono::Utc>",
               s.updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
               s.item_id AS "item_id: ItemId",
               s.paused_at AS "paused_at: chrono::DateTime<chrono::Utc>"
        FROM subscriptions s
        WHERE s.project_id IN (SELECT id FROM projects WHERE user_id = $1)
          AND s.status = 'active'
          AND s.paused_at IS NULL
        "#,
        creator_id as UserId,
    )
    .fetch_all(pool)
    .await?;

    Ok(subs)
}

/// Mark all active subscriptions to a creator's projects as paused.
#[tracing::instrument(skip_all)]
pub(crate) async fn pause_subscriptions_for_creator(
    pool: &PgPool,
    creator_id: UserId,
) -> Result<u64> {
    let result = sqlx::query!(
        r#"
        UPDATE subscriptions SET paused_at = NOW()
        WHERE project_id IN (SELECT id FROM projects WHERE user_id = $1)
          AND status = 'active'
          AND paused_at IS NULL
        "#,
        creator_id as UserId,
    )
    .execute(pool)
    .await?;

    Ok(result.rows_affected())
}

/// Get all paused subscriptions to a creator's projects (for cancelling on termination).
#[tracing::instrument(skip_all)]
pub(crate) async fn get_paused_subscriptions_by_creator(
    pool: &PgPool,
    creator_id: UserId,
) -> Result<Vec<DbSubscription>> {
    let subs = sqlx::query_as!(
        DbSubscription,
        r#"
        SELECT s.id AS "id: SubscriptionId", s.subscriber_id AS "subscriber_id: UserId",
               s.tier_id AS "tier_id: SubscriptionTierId", s.project_id AS "project_id: ProjectId",
               s.stripe_subscription_id, s.stripe_customer_id,
               s.status AS "status: super::SubscriptionStatus",
               s.current_period_start AS "current_period_start: chrono::DateTime<chrono::Utc>",
               s.current_period_end AS "current_period_end: chrono::DateTime<chrono::Utc>",
               s.canceled_at AS "canceled_at: chrono::DateTime<chrono::Utc>",
               s.created_at AS "created_at: chrono::DateTime<chrono::Utc>",
               s.updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
               s.item_id AS "item_id: ItemId",
               s.paused_at AS "paused_at: chrono::DateTime<chrono::Utc>"
        FROM subscriptions s
        WHERE s.project_id IN (SELECT id FROM projects WHERE user_id = $1)
          AND s.status = 'active'
          AND s.paused_at IS NOT NULL
        "#,
        creator_id as UserId,
    )
    .fetch_all(pool)
    .await?;

    Ok(subs)
}

/// Resume all paused subscriptions for a creator's projects.
#[tracing::instrument(skip_all)]
pub(crate) async fn resume_subscriptions_for_creator(
    pool: &PgPool,
    creator_id: UserId,
) -> Result<Vec<DbSubscription>> {
    let subs = sqlx::query_as!(
        DbSubscription,
        r#"
        UPDATE subscriptions SET paused_at = NULL
        WHERE project_id IN (SELECT id FROM projects WHERE user_id = $1)
          AND status = 'active'
          AND paused_at IS NOT NULL
        RETURNING id AS "id: SubscriptionId", subscriber_id AS "subscriber_id: UserId",
                  tier_id AS "tier_id: SubscriptionTierId", project_id AS "project_id: ProjectId",
                  stripe_subscription_id, stripe_customer_id,
                  status AS "status: super::SubscriptionStatus",
                  current_period_start AS "current_period_start: chrono::DateTime<chrono::Utc>",
                  current_period_end AS "current_period_end: chrono::DateTime<chrono::Utc>",
                  canceled_at AS "canceled_at: chrono::DateTime<chrono::Utc>",
                  created_at AS "created_at: chrono::DateTime<chrono::Utc>",
                  updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
                  item_id AS "item_id: ItemId",
                  paused_at AS "paused_at: chrono::DateTime<chrono::Utc>"
        "#,
        creator_id as UserId,
    )
    .fetch_all(pool)
    .await?;

    Ok(subs)
}

// ── Access control ──

/// SQL predicate identifying a `subscriptions` row that currently grants access
/// to its scope. The `current_period_end` clause is defense-in-depth against a
/// missed/delayed `customer.subscription.deleted` webhook, `status = 'active'`
/// alone trusts Stripe to push the cancellation promptly.
///
/// What a subscription access check is scoped to. Both arms run the SAME sealed
/// predicate inside [`gate`], so a project gate and an item gate cannot diverge.
#[derive(Debug, Clone, Copy)]
pub enum SubscriptionScope {
    Project(ProjectId),
    Item(super::ItemId),
}

pub(crate) use gate::SubscriptionGate;

/// Sealed home of the "does a subscription grant access right now" predicate.
///
/// The predicate text lives in exactly ONE place, [`SubscriptionGate`]'s
/// private `PREDICATE` associated const, and is unreachable from the rest of
/// this module, let alone other modules. The only way to learn "this user has
/// access" is [`SubscriptionGate::check`] (or [`SubscriptionGate::accessible_item_ids`]
/// for the batch shape), each of which runs that predicate. A `SubscriptionGate`
/// value is a witness: its field is private and there is no public constructor,
/// so access-granting code can neither fabricate one nor hand-write a divergent
/// gate.
///
/// The predicate is a private associated const inside a sealed submodule rather
/// than a shareable `&str`: a shared const is copy-pasteable, and a copy that
/// drops the `current_period_end` clause grants access it should not. Sealing it
/// here makes the divergence unwritable rather than discouraged.
mod gate {
    use super::SubscriptionScope;
    use crate::db::{ItemId, UserId};
    use crate::error::Result;
    use sqlx::PgPool;
    use std::collections::HashMap;

    /// Proof that a subscription currently grants access. Constructible ONLY via
    /// [`SubscriptionGate::check`], the private `()` field seals the type so no
    /// other code can mint one.
    #[derive(Debug, Clone, Copy)]
    pub struct SubscriptionGate(());

    impl SubscriptionGate {
        /// The single source of truth for "grants access right now". Private to
        /// this submodule: nothing outside can read it as a string, so it cannot
        /// be copy-pasted into a divergent query. (Compile-time constant, never
        /// user input, so the `format!` interpolation is injection-safe; the
        /// `$N` placeholders stay bound.)
        const PREDICATE: &'static str = "status = 'active' AND paused_at IS NULL \
            AND (current_period_end IS NULL OR current_period_end > NOW())";

        /// Does `user_id` hold a subscription that currently grants access to
        /// `scope`? Returns `Some(gate)` iff so, the sole gate constructor and
        /// the single entry point for project- and item-level access checks.
        #[tracing::instrument(skip_all)]
        pub async fn check(
            pool: &PgPool,
            user_id: UserId,
            scope: SubscriptionScope,
        ) -> Result<Option<SubscriptionGate>> {
            // runtime-checked: dynamically-built SQL, the access predicate is
            // interpolated from the sealed `PREDICATE` const via format!, so the
            // statement text isn't a compile-time literal the macro can verify.
            let exists: bool = match scope {
                SubscriptionScope::Project(project_id) => {
                    sqlx::query_scalar(&format!(
                        "SELECT EXISTS(SELECT 1 FROM subscriptions \
                         WHERE subscriber_id = $1 AND project_id = $2 AND {})",
                        Self::PREDICATE
                    ))
                    .bind(user_id)
                    .bind(project_id)
                    .fetch_one(pool)
                    .await?
                }
                SubscriptionScope::Item(item_id) => {
                    sqlx::query_scalar(&format!(
                        "SELECT EXISTS(SELECT 1 FROM subscriptions \
                         WHERE subscriber_id = $1 AND item_id = $2 AND {})",
                        Self::PREDICATE
                    ))
                    .bind(user_id)
                    .bind(item_id)
                    .fetch_one(pool)
                    .await?
                }
            };

            Ok(exists.then_some(SubscriptionGate(())))
        }

        /// Every item ID `user_id` currently has access to via subscription
        /// (batch gate). Runs the same sealed predicate as [`check`], so the
        /// batch path cannot drift from the single-item gate.
        #[tracing::instrument(skip_all)]
        pub async fn accessible_item_ids(pool: &PgPool, user_id: UserId) -> Result<Vec<ItemId>> {
            // runtime-checked: dynamically-built SQL, the access predicate is
            // interpolated from the sealed `PREDICATE` const via format!, so the
            // statement text isn't a compile-time literal the macro can verify.
            let item_ids: Vec<ItemId> = sqlx::query_scalar(&format!(
                "SELECT DISTINCT item_id FROM subscriptions \
                 WHERE subscriber_id = $1 AND item_id IS NOT NULL AND {}",
                Self::PREDICATE
            ))
            .bind(user_id)
            .fetch_all(pool)
            .await?;

            Ok(item_ids)
        }

        /// Map of every item `user_id` currently has subscription access to →
        /// its access proof. The witness-bearing batch shape used by the project
        /// page, where each item's [`AccessContext`](crate::pricing::AccessContext)
        /// needs its own gate. Runs the sealed predicate once.
        #[tracing::instrument(skip_all)]
        #[allow(
            clippy::zero_sized_map_values,
            reason = "SubscriptionGate is a deliberate zero-sized capability witness (sealed constructor); the map value carries type-level proof of access, not data, so a HashSet would lose the witness semantics"
        )]
        pub async fn subscribed_item_gates(
            pool: &PgPool,
            user_id: UserId,
        ) -> Result<HashMap<ItemId, SubscriptionGate>> {
            let ids = Self::accessible_item_ids(pool, user_id).await?;
            Ok(ids
                .into_iter()
                .map(|id| (id, SubscriptionGate(())))
                .collect())
        }

        /// Test-only constructor. Real gates can only be minted by running the
        /// predicate against the DB; unit tests (e.g. `pricing`) need to
        /// fabricate the "access granted" state without a database. Gated to
        /// test builds so production code still cannot forge a witness.
        #[cfg(test)]
        pub(crate) fn test_witness() -> Self {
            SubscriptionGate(())
        }
    }
}

/// Does `user_id` hold a subscription that currently grants access to `scope`?
///
/// Thin boolean wrapper over the sealed [`SubscriptionGate::check`]; prefer
/// taking the [`SubscriptionGate`] witness directly where a proof of access is
/// useful downstream.
#[tracing::instrument(skip_all)]
pub(crate) async fn has_access(
    pool: &PgPool,
    user_id: UserId,
    scope: SubscriptionScope,
) -> Result<bool> {
    Ok(SubscriptionGate::check(pool, user_id, scope)
        .await?
        .is_some())
}

/// Get user subscriptions joined with project and tier data (for library display).
#[tracing::instrument(skip_all)]
pub(crate) async fn get_user_subscriptions_with_details(
    pool: &PgPool,
    user_id: UserId,
) -> Result<Vec<DbUserSubscriptionRow>> {
    let rows = sqlx::query_as!(
        DbUserSubscriptionRow,
        r#"
        SELECT s.id AS "id: SubscriptionId", s.project_id AS "project_id!: ProjectId",
               p.title AS project_title, p.slug AS "project_slug: super::Slug",
               t.name AS tier_name, t.price_cents, s.status AS "status: super::SubscriptionStatus",
               s.current_period_end AS "current_period_end: chrono::DateTime<chrono::Utc>",
               s.stripe_subscription_id,
               u.settlement_currency AS "settlement_currency: crate::currency::SettlementCurrency"
        FROM subscriptions s
        JOIN projects p ON p.id = s.project_id
        JOIN users u ON u.id = p.user_id
        JOIN subscription_tiers t ON t.id = s.tier_id
        WHERE s.subscriber_id = $1
        ORDER BY s.created_at DESC
        LIMIT 1000
        "#,
        user_id as UserId,
    )
    .fetch_all(pool)
    .await?;

    Ok(rows)
}

/// Get the number of active subscribers to a project (for dashboard display).
///
/// NOT an access gate, this is a creator-facing headcount, so it deliberately
/// counts `status = 'active'` rows regardless of `current_period_end` (a sub in
/// its grace window is still a subscriber). Do not "align" it with
/// [`GRANTS_ACCESS_PREDICATE`]; the divergence here is intentional.
#[tracing::instrument(skip_all)]
pub(crate) async fn get_project_subscriber_count(
    pool: &PgPool,
    project_id: ProjectId,
) -> Result<i64> {
    let count: i64 = sqlx::query_scalar!(
        r#"SELECT COUNT(*) AS "count!" FROM subscriptions WHERE project_id = $1 AND status = 'active' AND paused_at IS NULL"#,
        project_id as ProjectId,
    )
    .fetch_one(pool)
    .await?;

    Ok(count)
}

// ── Export ──

/// Export all subscribers across a creator's projects.
///
/// Returns username, display_name, tier name, subscription status, and when.
#[tracing::instrument(skip_all)]
/// One page of a creator's project subscribers for CSV export, newest first.
/// Paginated for bounded-memory streaming; stable
/// `(created_at, id)` ordering keeps OFFSET batches consistent.
/// How many rows the subscriber half of the follower export will page through.
///
/// The same set `get_project_subscribers_for_export_page` walks. Asked once
/// before the export starts, so `follower_exports` can record what was about to
/// be handed over (`159a7a20`).
#[tracing::instrument(skip_all)]
pub(crate) async fn count_project_subscribers_for_export(
    pool: &PgPool,
    user_id: UserId,
) -> Result<i64> {
    let count = sqlx::query_scalar::<_, i64>(
        r"
        SELECT COUNT(*)
        FROM subscriptions s
        WHERE s.project_id IN (SELECT id FROM projects WHERE user_id = $1)
        ",
    )
    .bind(user_id)
    .fetch_one(pool)
    .await?;
    Ok(count)
}

pub(crate) async fn get_project_subscribers_for_export_page(
    pool: &PgPool,
    user_id: UserId,
    limit: i64,
    offset: i64,
) -> Result<Vec<SubscriberExportRow>> {
    let rows = sqlx::query_as!(
        SubscriberExportRow,
        r#"
        SELECT u.username, u.display_name, t.name AS tier_name,
               s.status AS "status: super::SubscriptionStatus",
               s.created_at AS "created_at: chrono::DateTime<chrono::Utc>"
        FROM subscriptions s
        JOIN users u ON u.id = s.subscriber_id
        JOIN subscription_tiers t ON t.id = s.tier_id
        WHERE s.project_id IN (SELECT id FROM projects WHERE user_id = $1)
        ORDER BY s.created_at DESC, s.id DESC
        LIMIT $2 OFFSET $3
        "#,
        user_id as UserId,
        limit,
        offset,
    )
    .fetch_all(pool)
    .await?;

    Ok(rows)
}

/// Export all subscriptions across a creator's projects with full detail.
///
/// Returns project name, tier name, price, subscriber username, status,
/// billing period dates, and cancellation date.
#[tracing::instrument(skip_all)]
/// One page of a creator's subscriptions for CSV export, newest first.
/// Paginated for bounded-memory streaming; stable
/// `(created_at, id)` ordering keeps OFFSET batches consistent.
pub(crate) async fn get_subscriptions_for_export_page(
    pool: &PgPool,
    user_id: UserId,
    limit: i64,
    offset: i64,
) -> Result<Vec<SubscriptionExportRow>> {
    let rows = sqlx::query_as!(
        SubscriptionExportRow,
        r#"
        SELECT p.title AS project_title, t.name AS tier_name, t.price_cents,
               u.username, s.status AS "status: super::SubscriptionStatus",
               s.current_period_start AS "current_period_start: chrono::DateTime<chrono::Utc>",
               s.current_period_end AS "current_period_end: chrono::DateTime<chrono::Utc>",
               s.canceled_at AS "canceled_at: chrono::DateTime<chrono::Utc>",
               s.created_at AS "created_at: chrono::DateTime<chrono::Utc>"
        FROM subscriptions s
        JOIN users u ON u.id = s.subscriber_id
        JOIN subscription_tiers t ON t.id = s.tier_id
        JOIN projects p ON p.id = s.project_id
        WHERE s.project_id IN (SELECT id FROM projects WHERE user_id = $1)
        ORDER BY s.created_at DESC, s.id DESC
        LIMIT $2 OFFSET $3
        "#,
        user_id as UserId,
        limit,
        offset,
    )
    .fetch_all(pool)
    .await?;

    Ok(rows)
}

// ── Event log ──

/// Log a subscription webhook event for debugging, idempotency, and revenue
/// reconciliation. The `ON CONFLICT (stripe_event_id) DO NOTHING` makes a
/// redelivered event a silent no-op, never an error, so an `Err` return is
/// always a genuine DB failure, which callers log at `error!` (a dropped
/// reconciliation row is worth alerting on, not burying at `warn!`).
///
/// `event_type` is an [`MnwEventName`], not a string. The names in this table
/// are MNW's own vocabulary rather than Stripe's, and they were previously
/// spelled out at 26 call sites across three files, where a typo produced a row
/// nobody would ever match on. Taking the enum makes the spelling
/// unmisspellable and puts every name in one place.
#[tracing::instrument(skip_all)]
pub(crate) async fn log_subscription_event(
    pool: &PgPool,
    subscription_id: Option<SubscriptionId>,
    stripe_event_id: &str,
    event_type: crate::payments::MnwEventName,
    payload: &serde_json::Value,
) -> Result<()> {
    let event_type = event_type.as_str();
    sqlx::query!(
        r#"
        INSERT INTO subscription_events (subscription_id, stripe_event_id, event_type, payload)
        VALUES ($1, $2, $3, $4)
        ON CONFLICT (stripe_event_id) DO NOTHING
        "#,
        subscription_id as Option<SubscriptionId>,
        stripe_event_id,
        event_type,
        payload as &serde_json::Value,
    )
    .execute(pool)
    .await?;

    Ok(())
}