//! Reading purchases back: what a buyer owns, what a seller sold, and the
//! paginated exports of both.

use super::super::{
    Cents, ClaimToken, DbPurchaseRow, DbTransaction, DbTransactionExportRow, DownloadToken, ItemId,
    KeyCode, PgPool, ProjectId, PromoCodeId, Result, TransactionId, UserId,
};

/// List transactions where the user is the buyer, newest first.
///
/// Pass `limit: None` for all rows (exports), or `Some(n)` for dashboard display.
#[tracing::instrument(skip_all)]
pub async fn get_transactions_by_buyer(
    pool: &PgPool,
    buyer_id: UserId,
    limit: Option<i64>,
) -> Result<Vec<DbTransaction>> {
    let txs = sqlx::query_as!(
        DbTransaction,
        r#"
        SELECT
            id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
            item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
            currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
            created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
            item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
            parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
            guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
            download_token AS "download_token: DownloadToken",
            presentment_amount_cents, presentment_currency
        FROM transactions WHERE buyer_id = $1 ORDER BY created_at DESC LIMIT $2
        "#,
        buyer_id as UserId,
        limit,
    )
    .fetch_all(pool)
    .await?;

    Ok(txs)
}

/// One page of a buyer's purchases for CSV export, newest first.
///
/// Paginated so the purchases export streams in bounded batches rather than
/// loading the buyer's whole history with `limit: None`.
/// Stable `(created_at, id)` ordering keeps OFFSET batches consistent.
pub async fn get_buyer_transactions_for_export_page(
    pool: &PgPool,
    buyer_id: UserId,
    limit: i64,
    offset: i64,
) -> Result<Vec<DbTransaction>> {
    let txs = sqlx::query_as!(
        DbTransaction,
        r#"
        SELECT
            id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
            item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
            currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
            created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
            item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
            parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
            guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
            download_token AS "download_token: DownloadToken",
            presentment_amount_cents, presentment_currency
        FROM transactions WHERE buyer_id = $1
        ORDER BY created_at DESC, id DESC
        LIMIT $2 OFFSET $3
        "#,
        buyer_id as UserId,
        limit,
        offset,
    )
    .fetch_all(pool)
    .await?;

    Ok(txs)
}

/// List transactions where the user is the seller, newest first.
///
/// Pass `limit: None` for all rows (exports), or `Some(n)` for dashboard display.
#[tracing::instrument(skip_all)]
pub async fn get_transactions_by_seller(
    pool: &PgPool,
    seller_id: UserId,
    limit: Option<i64>,
) -> Result<Vec<DbTransaction>> {
    let txs = sqlx::query_as!(
        DbTransaction,
        r#"
        SELECT
            id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
            item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
            currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
            created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
            item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
            parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
            guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
            download_token AS "download_token: DownloadToken",
            presentment_amount_cents, presentment_currency
        FROM transactions WHERE seller_id = $1 ORDER BY created_at DESC LIMIT $2
        "#,
        seller_id as UserId,
        limit,
    )
    .fetch_all(pool)
    .await?;

    Ok(txs)
}

/// Check whether a user has a completed purchase for a given item.
#[tracing::instrument(skip_all)]
pub async fn has_purchased_item(pool: &PgPool, user_id: UserId, item_id: ItemId) -> Result<bool> {
    let count: i64 = sqlx::query_scalar!(
        r#"SELECT COUNT(*) AS "count!" FROM transactions WHERE buyer_id = $1 AND item_id = $2 AND status = 'completed'"#,
        user_id as UserId,
        item_id as ItemId,
    )
    .fetch_one(pool)
    .await?;

    Ok(count > 0)
}

/// Bulk variant of `has_purchased_item`. Returns the subset of `item_ids` that
/// the buyer has a completed purchase for. Single DB roundtrip vs. N calls.
#[tracing::instrument(skip_all)]
pub async fn purchased_subset(
    pool: &PgPool,
    user_id: UserId,
    item_ids: &[ItemId],
) -> Result<std::collections::HashSet<ItemId>> {
    if item_ids.is_empty() {
        return Ok(std::collections::HashSet::new());
    }
    let rows = sqlx::query_scalar!(
        r#"SELECT DISTINCT item_id AS "item_id!: ItemId" FROM transactions
         WHERE buyer_id = $1 AND status = 'completed' AND item_id = ANY($2)"#,
        user_id as UserId,
        item_ids as &[ItemId],
    )
    .fetch_all(pool)
    .await?;
    Ok(rows.into_iter().collect())
}

/// Get all item IDs that a user has purchased (for batch access checks)
#[tracing::instrument(skip_all)]
pub async fn get_user_purchased_item_ids(pool: &PgPool, user_id: UserId) -> Result<Vec<ItemId>> {
    let item_ids: Vec<ItemId> = sqlx::query_scalar!(
        r#"SELECT DISTINCT item_id AS "item_id!: ItemId" FROM transactions WHERE buyer_id = $1 AND status = 'completed' AND item_id IS NOT NULL"#,
        user_id as UserId,
    )
    .fetch_all(pool)
    .await?;

    Ok(item_ids)
}

/// Check whether a user has a completed purchase for a given project.
#[tracing::instrument(skip_all)]
pub async fn has_purchased_project(
    pool: &PgPool,
    user_id: UserId,
    project_id: ProjectId,
) -> Result<bool> {
    let count: i64 = sqlx::query_scalar!(
        r#"SELECT COUNT(*) AS "count!" FROM transactions WHERE buyer_id = $1 AND project_id = $2 AND status = 'completed'"#,
        user_id as UserId,
        project_id as ProjectId,
    )
    .fetch_one(pool)
    .await?;

    Ok(count > 0)
}

/// Get items purchased by a user, including any associated license key.
///
/// Reads from the `purchases` VIEW (which filters `transactions` to
/// `status = 'completed'`), then JOINs through `items → projects → users`
/// for display fields. The LEFT JOIN on `license_keys` attaches the most
/// recent non-revoked key code so the buyer can see it in their library
/// without a separate lookup. Capped at 20 rows for the dashboard summary.
#[tracing::instrument(skip_all)]
pub async fn get_user_purchases(pool: &PgPool, user_id: UserId) -> Result<Vec<DbPurchaseRow>> {
    let purchases = sqlx::query_as!(
        DbPurchaseRow,
        r#"
        SELECT
            transaction_id AS "transaction_id!: TransactionId",
            item_id AS "item_id!: ItemId",
            title AS "title!",
            creator AS "creator!",
            item_type AS "item_type!: crate::db::ItemType",
            purchased_at AS "purchased_at!: chrono::DateTime<chrono::Utc>",
            is_free AS "is_free!",
            license_key_code AS "license_key_code?: KeyCode",
            has_new_version AS "has_new_version!"
        FROM (
            SELECT DISTINCT ON (p.item_id)
                p.transaction_id,
                p.item_id,
                i.title,
                u.username as creator,
                i.item_type,
                p.purchased_at,
                -- Badge from what the buyer actually paid, not the item's current
                -- price: a later re-price to $0 must not retroactively badge a paid
                -- purchase "Free" (nor vice-versa). The purchases view carries the
                -- transaction's own amount_cents.
                (p.amount_cents = 0) as is_free,
                lk.key_code as license_key_code,
                (vc.total_versions > 0 AND vc.total_versions > COALESCE(dc.downloaded_count, 0)) as has_new_version
            FROM purchases p
            JOIN items i ON p.item_id = i.id
            JOIN projects proj ON i.project_id = proj.id
            JOIN users u ON proj.user_id = u.id
            LEFT JOIN license_keys lk ON lk.item_id = p.item_id AND lk.owner_id = p.buyer_id AND lk.revoked_at IS NULL
            LEFT JOIN LATERAL (
                SELECT COUNT(*) AS total_versions
                FROM versions v
                WHERE v.item_id = i.id AND v.s3_key IS NOT NULL
            ) vc ON true
            LEFT JOIN LATERAL (
                SELECT COUNT(*) AS downloaded_count
                FROM user_downloads ud
                WHERE ud.user_id = p.buyer_id AND ud.item_id = i.id
            ) dc ON true
            WHERE p.buyer_id = $1
            ORDER BY p.item_id, p.purchased_at DESC
        ) deduped
        ORDER BY purchased_at DESC
        LIMIT 20
        "#,
        user_id as UserId,
    )
    .fetch_all(pool)
    .await?;

    Ok(purchases)
}

/// Fetch a single transaction by ID.
///
/// # Authorization
///
/// This lookup is intentionally **unscoped**, it does not filter by buyer or
/// seller, because the two callers need the row to *decide* authorization
/// (a receipt page shown to buyer-or-seller; a refund restricted to the
/// seller). Every caller MUST therefore check ownership against the returned
/// `buyer_id`/`seller_id` before acting on it. A new caller that returns this
/// row's contents without such a check would be an IDOR, there is no implicit
/// scoping here to lean on.
#[tracing::instrument(skip_all)]
pub async fn get_transaction_by_id(
    pool: &PgPool,
    id: TransactionId,
) -> Result<Option<DbTransaction>> {
    let tx = sqlx::query_as!(
        DbTransaction,
        r#"
        SELECT
            id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
            item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
            currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
            created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
            item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
            parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
            guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
            download_token AS "download_token: DownloadToken",
            presentment_amount_cents, presentment_currency
        FROM transactions WHERE id = $1
        "#,
        id as TransactionId,
    )
    .fetch_optional(pool)
    .await?;
    Ok(tx)
}

#[tracing::instrument(skip_all)]
/// One page of a seller's sales for CSV export, newest first.
///
/// Paginated (`LIMIT`/`OFFSET`) so the export streams in bounded batches instead
/// of loading the seller's entire transaction history into memory in one query.
/// The `(created_at, id)` ordering is stable so OFFSET
/// batches don't reorder. (Keyset pagination would avoid OFFSET's deep-scan cost
/// and is the future optimization; OFFSET is sufficient at current scale and
/// keeps peak memory + per-query result bounded, which is the DoS fix.)
pub async fn get_seller_transactions_for_export_page(
    pool: &PgPool,
    seller_id: UserId,
    limit: i64,
    offset: i64,
) -> Result<Vec<DbTransactionExportRow>> {
    let rows = sqlx::query_as!(
        DbTransactionExportRow,
        r#"
        SELECT
            t.created_at AS "created_at: chrono::DateTime<chrono::Utc>",
            t.item_id AS "item_id: ItemId",
            t.item_title,
            t.amount_cents AS "amount_cents: Cents",
            t.status AS "status: crate::db::TransactionStatus",
            CASE WHEN t.share_contact AND NOT EXISTS (
                SELECT 1 FROM contact_revocations cr
                WHERE cr.buyer_id = t.buyer_id AND cr.seller_id = t.seller_id
            ) THEN u.email ELSE NULL END as buyer_email
        FROM transactions t
        LEFT JOIN users u ON u.id = t.buyer_id
        WHERE t.seller_id = $1
        ORDER BY t.created_at DESC, t.id DESC
        LIMIT $2 OFFSET $3
        "#,
        seller_id as UserId,
        limit,
        offset,
    )
    .fetch_all(pool)
    .await?;

    Ok(rows)
}

/// All of a seller's export rows accumulated into one Vec, bounded to
/// `EXPORT_ACCUMULATE_CAP` rows. For admin / internal-API callers that need the
/// full set in memory; the public creator-facing export streams page-by-page via
/// [`get_seller_transactions_for_export_page`] instead of materializing here.
pub async fn get_seller_transactions_for_export(
    pool: &PgPool,
    seller_id: UserId,
) -> Result<Vec<DbTransactionExportRow>> {
    /// Page size for the accumulating fetch.
    const PAGE: i64 = 5_000;
    /// Cap so even an admin/internal export can't load an unbounded result set.
    const EXPORT_ACCUMULATE_CAP: usize = 1_000_000;

    let mut all = Vec::new();
    let mut offset = 0i64;
    loop {
        let page = get_seller_transactions_for_export_page(pool, seller_id, PAGE, offset).await?;
        let n = page.len();
        all.extend(page);
        offset += n as i64;
        if (n as i64) < PAGE || all.len() >= EXPORT_ACCUMULATE_CAP {
            break;
        }
    }
    Ok(all)
}