//! DB-layer contract tests for the money read-models:
//! `db::transactions::revenue_stats` and `db::transactions::seller_contacts`.
//!
//! Revenue is what a creator is told they earned and what the platform reports
//! to itself, so the aggregates are pinned with several distinct, non-round
//! amounts per group: a wrong aggregate operator (max instead of sum, a count
//! standing in for a total, a refunded row leaking into revenue) lands on a
//! different number rather than agreeing by accident. The empty case is pinned
//! separately, because zero rows is the one input where every wrong aggregate
//! agrees with the right one.
//!
//! Contacts are a creator's fan list, which is the data a fan can withdraw:
//! these pin that sharing is opt-in, that a revocation removes a buyer from
//! every contact surface, that the platform-notification query deliberately
//! ignores both, and that the paginated export walks each buyer exactly once.
//!
//! The Fan+ subscription store that used to share this file is
//! `db_fan_plus_layer`.
//!
//! Delete this file and nothing checks that completed-only,
//! currency-separated revenue is what the dashboards read, or that a revoked
//! fan stays revoked.

use crate::harness::db::TestDb;
use crate::harness::seed_user;
use makenotwork::currency::SettlementCurrency;
use makenotwork::db::{ItemId, ProjectId, UserId, transactions};

// ── seeding ──

/// Insert a project owned by `user` with a distinguishable title.
async fn seed_titled_project(
    pool: &sqlx::PgPool,
    user: UserId,
    slug: &str,
    title: &str,
) -> ProjectId {
    sqlx::query_scalar::<_, ProjectId>(
        "INSERT INTO projects (user_id, slug, title) VALUES ($1, $2, $3) RETURNING id",
    )
    .bind(user)
    .bind(slug)
    .bind(title)
    .fetch_one(pool)
    .await
    .expect("seed project")
}

async fn seed_item(pool: &sqlx::PgPool, project: ProjectId, slug: &str) -> ItemId {
    sqlx::query_scalar::<_, ItemId>(
        "INSERT INTO items (project_id, title, item_type, price_cents, slug)
         VALUES ($1, $2, 'digital', 1000, $3) RETURNING id",
    )
    .bind(project)
    .bind(format!("Item {slug}"))
    .bind(slug)
    .fetch_one(pool)
    .await
    .expect("seed item")
}

/// One transaction row, described by the fields these read-models group on.
struct Sale {
    buyer: UserId,
    seller: UserId,
    item: ItemId,
    amount_cents: i32,
    currency: &'static str,
    status: &'static str,
    share_contact: bool,
    /// How long ago the sale happened. Relative offsets rather than fixed
    /// timestamps, so ordering is deterministic without any wall-clock
    /// assumption beyond "NOW() moves forward".
    hours_ago: i32,
}

impl Sale {
    /// A completed USD sale, the shape most of these tests vary from.
    fn completed(buyer: UserId, seller: UserId, item: ItemId, amount_cents: i32) -> Self {
        Sale {
            buyer,
            seller,
            item,
            amount_cents,
            currency: "usd",
            status: "completed",
            share_contact: false,
            hours_ago: 1,
        }
    }

    async fn insert(&self, pool: &sqlx::PgPool, session: &str) {
        sqlx::query(
            "INSERT INTO transactions
               (buyer_id, seller_id, item_id, amount_cents, platform_fee_cents, currency,
                status, stripe_checkout_session_id, item_title, seller_username, share_contact,
                created_at, completed_at)
             VALUES ($1, $2, $3, $4, 0, $5, $6, $7, 'Item', 'seller', $8,
                     NOW() - make_interval(hours => $9::int),
                     CASE WHEN $6 IN ('completed', 'refunded')
                          THEN NOW() - make_interval(hours => $9::int) END)",
        )
        .bind(self.buyer)
        .bind(self.seller)
        .bind(self.item)
        .bind(self.amount_cents)
        .bind(self.currency)
        .bind(self.status)
        .bind(session)
        .bind(self.share_contact)
        .bind(self.hours_ago)
        .execute(pool)
        .await
        .expect("seed transaction");
    }
}

// ── revenue_stats ──

#[tokio::test]
async fn project_revenue_sums_completed_sales_and_ignores_every_other_status() {
    let db = TestDb::new().await;
    let seller = seed_user(&db.pool, "rev_proj_seller").await;
    let buyer = seed_user(&db.pool, "rev_proj_buyer").await;
    let project = seed_titled_project(&db.pool, seller, "rev-proj", "Rev").await;
    let item = seed_item(&db.pool, project, "rev-a").await;

    // Three completed sales at distinct, non-round amounts: 1234 + 5678 + 9012.
    // No two of them share a digit pattern that a max/min/first-row read would
    // land on, and the total (15924) differs from any single row and from any
    // pairwise sum.
    //
    // One item per completed sale, all three inside this project. The schema
    // carries `idx_transactions_buyer_item_completed`, a unique index on
    // (buyer_id, item_id) WHERE status = 'completed', so one buyer cannot hold
    // two completed rows for the same item; the aggregate groups on the
    // project, which is unaffected.
    for (i, amount) in [1234, 5678, 9012].into_iter().enumerate() {
        let sold = seed_item(&db.pool, project, &format!("rev-sale-{i}")).await;
        Sale::completed(buyer, seller, sold, amount)
            .insert(&db.pool, &format!("cs_rev_proj_{i}"))
            .await;
    }
    // Three rows that must contribute nothing. Their amounts are distinct from
    // each other, so leaking any single one moves the total to a different
    // wrong number rather than to a shared one.
    for (status, amount) in [("pending", 7777), ("refunded", 4444), ("failed", 3333)] {
        Sale {
            status,
            ..Sale::completed(buyer, seller, item, amount)
        }
        .insert(&db.pool, &format!("cs_rev_proj_{status}"))
        .await;
    }

    let (revenue, sales) = transactions::get_revenue_by_project(&db.pool, project)
        .await
        .expect("project revenue query ok");

    assert_eq!(
        revenue.in_currency(SettlementCurrency::Usd),
        15_924,
        "completed revenue is the sum of 1234 + 5678 + 9012, with pending, refunded and failed excluded"
    );
    assert_eq!(
        revenue.currency_count(),
        1,
        "one seller settling in one currency reports one currency"
    );
    assert_eq!(
        sales, 3,
        "the sales count counts completed rows only, not all six"
    );
}

#[tokio::test]
async fn project_revenue_is_empty_rather_than_zero_currency_noise_with_no_sales() {
    let db = TestDb::new().await;
    let seller = seed_user(&db.pool, "rev_empty_seller").await;
    let project = seed_titled_project(&db.pool, seller, "rev-empty", "Empty").await;
    seed_item(&db.pool, project, "rev-empty-a").await;

    let (revenue, sales) = transactions::get_revenue_by_project(&db.pool, project)
        .await
        .expect("project revenue query ok");

    assert!(
        revenue.is_empty(),
        "a project with no transactions reports no money in any currency"
    );
    assert_eq!(
        revenue.currency_count(),
        0,
        "no sales must not invent a zero-valued currency entry"
    );
    assert_eq!(revenue.in_currency(SettlementCurrency::Usd), 0);
    assert_eq!(sales, 0, "no completed rows means no sales");
}

#[tokio::test]
async fn project_revenue_keeps_two_currencies_apart_instead_of_adding_them() {
    let db = TestDb::new().await;
    let seller = seed_user(&db.pool, "rev_cur_seller").await;
    let buyer = seed_user(&db.pool, "rev_cur_buyer").await;
    let project = seed_titled_project(&db.pool, seller, "rev-cur", "Cur").await;

    // Two sales in each currency, so both groups exercise the SUM rather than
    // reading a single row: usd 2500 + 1725 = 4225, gbp 3175 + 1025 = 4200.
    // The two totals are close but unequal, so swapping the groups is visible,
    // and their sum (8425) is a third distinct number.
    for (i, (amount, currency)) in [(2500, "usd"), (1725, "usd"), (3175, "gbp"), (1025, "gbp")]
        .into_iter()
        .enumerate()
    {
        // One item per completed row: the same buyer may hold only one
        // completed transaction per item (idx_transactions_buyer_item_completed).
        let sold = seed_item(&db.pool, project, &format!("rev-cur-{i}")).await;
        Sale {
            currency,
            ..Sale::completed(buyer, seller, sold, amount)
        }
        .insert(&db.pool, &format!("cs_rev_cur_{i}"))
        .await;
    }

    let (revenue, sales) = transactions::get_revenue_by_project(&db.pool, project)
        .await
        .expect("project revenue query ok");

    assert_eq!(
        revenue.currency_count(),
        2,
        "pounds and dollars are reported as two totals, never one"
    );
    assert_eq!(
        revenue.in_currency(SettlementCurrency::Usd),
        4225,
        "the dollar total is the dollar rows only"
    );
    assert_eq!(
        revenue.in_currency(SettlementCurrency::Gbp),
        4200,
        "the pound total is the pound rows only"
    );
    assert_eq!(
        revenue.in_currency(SettlementCurrency::Eur),
        0,
        "a currency with no sales reports nothing, not the other currencies' money"
    );
    assert_eq!(sales, 4, "the count is currency-free and spans both groups");
}

#[tokio::test]
async fn user_project_revenue_ranks_earning_projects_and_drops_the_rest() {
    let db = TestDb::new().await;
    let seller = seed_user(&db.pool, "rev_user_seller").await;
    let buyer = seed_user(&db.pool, "rev_user_buyer").await;
    let other_seller = seed_user(&db.pool, "rev_user_other").await;

    let top = seed_titled_project(&db.pool, seller, "rev-top", "Top Seller").await;
    let mid = seed_titled_project(&db.pool, seller, "rev-mid", "Mid Seller").await;
    let quiet = seed_titled_project(&db.pool, seller, "rev-quiet", "Quiet").await;
    let foreign = seed_titled_project(&db.pool, other_seller, "rev-foreign", "Foreign").await;

    // Two items per earning project, because one buyer may hold only one
    // completed transaction per item (idx_transactions_buyer_item_completed).
    // Both sales still land in the same project, which is what the roll-up
    // groups on.
    let top_item_a = seed_item(&db.pool, top, "top-a").await;
    let top_item_b = seed_item(&db.pool, top, "top-b").await;
    let mid_item_usd = seed_item(&db.pool, mid, "mid-a").await;
    let mid_item_gbp = seed_item(&db.pool, mid, "mid-b").await;
    let quiet_item = seed_item(&db.pool, quiet, "quiet-a").await;
    let foreign_item = seed_item(&db.pool, foreign, "foreign-a").await;

    // Top: 5075 + 2425 = 7500 usd, two rows so the fold is exercised.
    Sale::completed(buyer, seller, top_item_a, 5075)
        .insert(&db.pool, "cs_rev_user_top_a")
        .await;
    Sale::completed(buyer, seller, top_item_b, 2425)
        .insert(&db.pool, "cs_rev_user_top_b")
        .await;
    // Mid: two currencies, both below the top project's single total, so the
    // ordering cannot be explained by "whichever project has more rows".
    Sale::completed(buyer, seller, mid_item_usd, 4200)
        .insert(&db.pool, "cs_rev_user_mid_usd")
        .await;
    Sale {
        currency: "gbp",
        ..Sale::completed(buyer, seller, mid_item_gbp, 6000)
    }
    .insert(&db.pool, "cs_rev_user_mid_gbp")
    .await;
    // Quiet: a pending sale only. The HAVING clause must drop it entirely.
    Sale {
        status: "pending",
        ..Sale::completed(buyer, seller, quiet_item, 9999)
    }
    .insert(&db.pool, "cs_rev_user_quiet")
    .await;
    // Another creator's earning project must not appear in this creator's list.
    Sale::completed(buyer, other_seller, foreign_item, 8888)
        .insert(&db.pool, "cs_rev_user_foreign")
        .await;

    let rows = transactions::get_revenue_by_user_projects(&db.pool, seller)
        .await
        .expect("user project revenue query ok");

    assert_eq!(
        rows.len(),
        2,
        "only the two projects with completed revenue are listed, got {:?}",
        rows.iter().map(|(_, t, _)| t.clone()).collect::<Vec<_>>()
    );
    assert_eq!(
        rows[0].0, top,
        "the largest single-currency total ranks first"
    );
    assert_eq!(rows[0].1, "Top Seller");
    assert_eq!(
        rows[0].2.in_currency(SettlementCurrency::Usd),
        7500,
        "the top project's two sales are summed, not counted"
    );
    assert_eq!(rows[1].0, mid);
    assert_eq!(
        rows[1].2.currency_count(),
        2,
        "a project that sold in two currencies keeps both"
    );
    assert_eq!(rows[1].2.in_currency(SettlementCurrency::Gbp), 6000);
    assert_eq!(rows[1].2.in_currency(SettlementCurrency::Usd), 4200);
}

#[tokio::test]
async fn platform_revenue_stats_separate_currencies_and_count_refunds_apart() {
    let db = TestDb::new().await;
    let seller = seed_user(&db.pool, "plat_seller").await;
    let buyer = seed_user(&db.pool, "plat_buyer").await;
    let project = seed_titled_project(&db.pool, seller, "plat", "Platform").await;
    let item = seed_item(&db.pool, project, "plat-a").await;

    // usd completed: 1234 + 5678 = 6912. gbp completed: 3175.
    // Refunded rows carry their own distinct amounts (4444, 2222): if a refund
    // leaked into revenue the dollar total would be 13578 or 11134, never 6912.
    for (i, (amount, currency, status)) in [
        (1234, "usd", "completed"),
        (5678, "usd", "completed"),
        (3175, "gbp", "completed"),
        (4444, "usd", "refunded"),
        (2222, "gbp", "refunded"),
        (7777, "usd", "pending"),
    ]
    .into_iter()
    .enumerate()
    {
        // Completed rows get an item each: one buyer may hold only one
        // completed transaction per item (idx_transactions_buyer_item_completed).
        // Refunded and pending rows are outside that partial index, and this
        // roll-up is platform-wide, so which item they name is immaterial.
        let sold = if status == "completed" {
            seed_item(&db.pool, project, &format!("plat-{i}")).await
        } else {
            item
        };
        Sale {
            currency,
            status,
            ..Sale::completed(buyer, seller, sold, amount)
        }
        .insert(&db.pool, &format!("cs_plat_{i}"))
        .await;
    }

    let (revenue, completed, refunded) = transactions::get_platform_revenue_stats(&db.pool)
        .await
        .expect("platform revenue query ok");

    assert_eq!(
        revenue.in_currency(SettlementCurrency::Usd),
        6912,
        "dollar revenue counts the two completed dollar sales only"
    );
    assert_eq!(
        revenue.in_currency(SettlementCurrency::Gbp),
        3175,
        "pound revenue counts the completed pound sale only"
    );
    assert_eq!(
        revenue.currency_count(),
        2,
        "the platform roll-up spans currencies without adding them"
    );
    assert_eq!(
        completed, 3,
        "three completed rows across both currencies, counts add where money does not"
    );
    assert_eq!(
        refunded, 2,
        "both refunded rows are counted, in whichever currency they were written"
    );
}

#[tokio::test]
async fn item_sales_list_completed_and_refunded_rows_for_that_seller_newest_first() {
    let db = TestDb::new().await;
    let seller = seed_user(&db.pool, "item_sales_seller").await;
    let rogue = seed_user(&db.pool, "item_sales_rogue").await;
    let buyer = seed_user(&db.pool, "item_sales_buyer").await;
    // A second buyer for the other seller's row on this item: the same buyer
    // may hold only one completed transaction per item
    // (idx_transactions_buyer_item_completed), and the point of that row is the
    // seller predicate, not the buyer.
    let rogue_buyer = seed_user(&db.pool, "item_sales_rogue_buyer").await;
    let project = seed_titled_project(&db.pool, seller, "item-sales", "Sales").await;
    let item = seed_item(&db.pool, project, "sales-a").await;
    let other_item = seed_item(&db.pool, project, "sales-b").await;

    // Newest completed sale, then an older refund: both belong in the tab.
    Sale {
        hours_ago: 1,
        ..Sale::completed(buyer, seller, item, 1500)
    }
    .insert(&db.pool, "cs_item_sales_new")
    .await;
    Sale {
        status: "refunded",
        hours_ago: 5,
        ..Sale::completed(buyer, seller, item, 2600)
    }
    .insert(&db.pool, "cs_item_sales_refunded")
    .await;
    // Excluded: still pending, a different item, and a row tagged to another seller.
    Sale {
        status: "pending",
        hours_ago: 2,
        ..Sale::completed(buyer, seller, item, 3700)
    }
    .insert(&db.pool, "cs_item_sales_pending")
    .await;
    Sale {
        hours_ago: 3,
        ..Sale::completed(buyer, seller, other_item, 4800)
    }
    .insert(&db.pool, "cs_item_sales_other_item")
    .await;
    Sale {
        hours_ago: 4,
        ..Sale::completed(rogue_buyer, rogue, item, 5900)
    }
    .insert(&db.pool, "cs_item_sales_other_seller")
    .await;

    let rows = transactions::get_sales_by_item(&db.pool, item, seller)
        .await
        .expect("item sales query ok");

    let amounts: Vec<i64> = rows.iter().map(|t| t.amount_cents.as_i64()).collect();
    assert_eq!(
        amounts,
        vec![1500, 2600],
        "completed and refunded sales of this item by this seller, newest first, got {amounts:?}"
    );
    assert!(
        rows.iter().all(|t| t.seller_id == Some(seller)),
        "another seller's row on the same item is never listed"
    );
}

// ── seller_contacts ──

#[tokio::test]
async fn seller_contacts_aggregate_only_shared_completed_purchases() {
    let db = TestDb::new().await;
    let seller = seed_user(&db.pool, "contacts_seller").await;
    let sharer = seed_user(&db.pool, "contacts_sharer").await;
    let revoker = seed_user(&db.pool, "contacts_revoker").await;
    let private_buyer = seed_user(&db.pool, "contacts_private").await;
    let project = seed_titled_project(&db.pool, seller, "contacts", "Contacts").await;
    let item = seed_item(&db.pool, project, "contacts-a").await;

    // The sharer bought twice: 1975 + 2225 = 4200 across two purchases. One
    // item each, because the same buyer may hold only one completed
    // transaction per item (idx_transactions_buyer_item_completed); the
    // contact roll-up groups by buyer, so the split changes nothing it reads.
    for (i, (amount, hours_ago)) in [(1975, 6), (2225, 2)].into_iter().enumerate() {
        let sold = seed_item(&db.pool, project, &format!("contacts-sharer-{i}")).await;
        Sale {
            share_contact: true,
            hours_ago,
            ..Sale::completed(sharer, seller, sold, amount)
        }
        .insert(&db.pool, &format!("cs_contacts_sharer_{i}"))
        .await;
    }
    // A pending purchase of theirs must not inflate either figure: 9999 would
    // be visible in both the total (14199) and the count (3).
    Sale {
        share_contact: true,
        status: "pending",
        ..Sale::completed(sharer, seller, item, 9999)
    }
    .insert(&db.pool, "cs_contacts_sharer_pending")
    .await;
    // Shared, then revoked: the buyer disappears from the list entirely.
    Sale {
        share_contact: true,
        ..Sale::completed(revoker, seller, item, 3050)
    }
    .insert(&db.pool, "cs_contacts_revoker")
    .await;
    // Never shared at all.
    Sale::completed(private_buyer, seller, item, 4100)
        .insert(&db.pool, "cs_contacts_private")
        .await;

    transactions::revoke_contact_sharing(&db.pool, revoker, seller)
        .await
        .expect("revoke ok");

    let contacts = transactions::get_seller_contacts(&db.pool, seller)
        .await
        .expect("seller contacts query ok");

    let usernames: Vec<&str> = contacts.iter().map(|c| c.username.as_str()).collect();
    assert_eq!(
        usernames,
        vec!["contacts_sharer"],
        "only the un-revoked opted-in buyer is a contact, got {usernames:?}"
    );
    assert_eq!(
        contacts[0].total_purchases, 2,
        "the pending purchase is not a purchase"
    );
    assert_eq!(
        contacts[0].total_spent_cents, 4200,
        "spend is 1975 + 2225, with the pending 9999 excluded"
    );
    assert_eq!(contacts[0].email, "contacts_sharer@test.com");
}

#[tokio::test]
async fn revoking_contact_sharing_is_idempotent_and_reversible() {
    let db = TestDb::new().await;
    let seller = seed_user(&db.pool, "revoke_seller").await;
    let buyer = seed_user(&db.pool, "revoke_buyer").await;
    let project = seed_titled_project(&db.pool, seller, "revoke", "Revoke").await;
    let item = seed_item(&db.pool, project, "revoke-a").await;
    Sale {
        share_contact: true,
        ..Sale::completed(buyer, seller, item, 2750)
    }
    .insert(&db.pool, "cs_revoke_1")
    .await;

    assert_eq!(
        transactions::get_seller_contacts(&db.pool, seller)
            .await
            .unwrap()
            .len(),
        1,
        "the buyer starts out as a contact"
    );

    // Revoking twice is one revocation: the second call hits ON CONFLICT DO
    // NOTHING, so it neither errors nor writes a second row.
    transactions::revoke_contact_sharing(&db.pool, buyer, seller)
        .await
        .expect("first revoke ok");
    transactions::revoke_contact_sharing(&db.pool, buyer, seller)
        .await
        .expect("repeat revoke ok");

    let revocations: i64 =
        sqlx::query_scalar("SELECT COUNT(*) FROM contact_revocations WHERE buyer_id = $1")
            .bind(buyer)
            .fetch_one(&db.pool)
            .await
            .unwrap();
    assert_eq!(revocations, 1, "a repeated revocation stays one row");
    assert!(
        transactions::get_seller_contacts(&db.pool, seller)
            .await
            .unwrap()
            .is_empty(),
        "a revoked buyer is not a contact"
    );

    // Clearing the revocation (the fan shares again on a new purchase) restores
    // the contact with its history intact.
    transactions::clear_contact_revocation(&db.pool, buyer, seller)
        .await
        .expect("clear ok");
    let restored = transactions::get_seller_contacts(&db.pool, seller)
        .await
        .unwrap();
    assert_eq!(
        restored.len(),
        1,
        "clearing the revocation restores the contact"
    );
    assert_eq!(
        restored[0].total_spent_cents, 2750,
        "the restored contact keeps the purchase history it always had"
    );
}

#[tokio::test]
async fn shared_creators_list_the_fan_still_shares_with_and_no_others() {
    let db = TestDb::new().await;
    let buyer = seed_user(&db.pool, "shared_buyer").await;
    let kept = seed_user(&db.pool, "shared_kept").await;
    let dropped = seed_user(&db.pool, "shared_dropped").await;
    let never = seed_user(&db.pool, "shared_never").await;

    for (seller, share, tag) in [
        (kept, true, "kept"),
        (dropped, true, "dropped"),
        (never, false, "never"),
    ] {
        let project = seed_titled_project(&db.pool, seller, &format!("shared-{tag}"), "S").await;
        let item = seed_item(&db.pool, project, &format!("shared-{tag}-a")).await;
        Sale {
            share_contact: share,
            ..Sale::completed(buyer, seller, item, 3300)
        }
        .insert(&db.pool, &format!("cs_shared_{tag}"))
        .await;
    }
    transactions::revoke_contact_sharing(&db.pool, buyer, dropped)
        .await
        .expect("revoke ok");

    let creators = transactions::get_shared_creators(&db.pool, buyer)
        .await
        .expect("shared creators query ok");
    let names: Vec<&str> = creators.iter().map(|c| c.username.as_str()).collect();
    assert_eq!(
        names,
        vec!["shared_kept"],
        "only the creator the fan shares with, and still shares with, is listed, got {names:?}"
    );
    assert_eq!(creators[0].seller_id, kept);
}

#[tokio::test]
async fn platform_notification_buyers_ignore_sharing_and_honour_the_limit() {
    let db = TestDb::new().await;
    let seller = seed_user(&db.pool, "notify_seller").await;
    let other_seller = seed_user(&db.pool, "notify_other_seller").await;
    let project = seed_titled_project(&db.pool, seller, "notify", "Notify").await;
    let item = seed_item(&db.pool, project, "notify-a").await;

    // Three buyers, none of whom shared contact, one of whom also revoked:
    // platform notices about content removal reach all of them anyway.
    let buyers = [
        seed_user(&db.pool, "notify_b1").await,
        seed_user(&db.pool, "notify_b2").await,
        seed_user(&db.pool, "notify_b3").await,
    ];
    for (i, amount) in [1107, 1214, 1321].into_iter().enumerate() {
        Sale::completed(buyers[i], seller, item, amount)
            .insert(&db.pool, &format!("cs_notify_{i}"))
            .await;
    }
    transactions::revoke_contact_sharing(&db.pool, buyers[0], seller)
        .await
        .expect("revoke ok");
    // A pending purchase is not a purchase, and another creator's buyer is not
    // this creator's buyer.
    let pending_buyer = seed_user(&db.pool, "notify_pending").await;
    Sale {
        status: "pending",
        ..Sale::completed(pending_buyer, seller, item, 2100)
    }
    .insert(&db.pool, "cs_notify_pending")
    .await;
    let foreign_buyer = seed_user(&db.pool, "notify_foreign").await;
    let foreign_project =
        seed_titled_project(&db.pool, other_seller, "notify-foreign", "Foreign").await;
    let foreign_item = seed_item(&db.pool, foreign_project, "notify-foreign-a").await;
    Sale::completed(foreign_buyer, other_seller, foreign_item, 2200)
        .insert(&db.pool, "cs_notify_foreign")
        .await;

    let all = transactions::get_all_buyers_for_seller(&db.pool, seller, 100)
        .await
        .expect("buyer notification query ok");
    let mut emails: Vec<String> = all.into_iter().map(|r| r.email).collect();
    emails.sort();
    assert_eq!(
        emails,
        vec![
            "notify_b1@test.com".to_string(),
            "notify_b2@test.com".to_string(),
            "notify_b3@test.com".to_string(),
        ],
        "every completed buyer of this seller is reachable, sharing preference and revocation notwithstanding"
    );

    // The cap is what bounds outbound mail volume, and the caller detects it by
    // the returned length, so a limit below the population must return exactly
    // that many rows.
    let capped = transactions::get_all_buyers_for_seller(&db.pool, seller, 2)
        .await
        .expect("capped query ok");
    assert_eq!(capped.len(), 2, "the limit is a row cap, not a suggestion");
}

#[tokio::test]
async fn contacts_export_pages_walk_each_buyer_once_newest_first() {
    let db = TestDb::new().await;
    let seller = seed_user(&db.pool, "page_seller").await;
    let project = seed_titled_project(&db.pool, seller, "page", "Page").await;
    let item = seed_item(&db.pool, project, "page-a").await;

    // Three contacts with distinct last-purchase times: newest first is
    // page_c (1h), then page_b (4h), then page_a (9h). Seeded in a different
    // order than they should come back, so "insertion order" cannot pass.
    for (username, hours_ago, amount) in [
        ("page_b", 4, 2600),
        ("page_a", 9, 1350),
        ("page_c", 1, 3900),
    ] {
        let buyer = seed_user(&db.pool, username).await;
        Sale {
            share_contact: true,
            hours_ago,
            ..Sale::completed(buyer, seller, item, amount)
        }
        .insert(&db.pool, &format!("cs_page_{username}"))
        .await;
    }

    let first = transactions::get_seller_contacts_page(&db.pool, seller, 2, 0)
        .await
        .expect("first page ok");
    let second = transactions::get_seller_contacts_page(&db.pool, seller, 2, 2)
        .await
        .expect("second page ok");
    let third = transactions::get_seller_contacts_page(&db.pool, seller, 2, 4)
        .await
        .expect("third page ok");

    let names = |rows: &[transactions::DbContactRow]| -> Vec<String> {
        rows.iter().map(|r| r.username.clone()).collect()
    };
    assert_eq!(
        names(&first),
        vec!["page_c".to_string(), "page_b".to_string()],
        "the first page is the two most recent purchasers, newest first"
    );
    assert_eq!(
        names(&second),
        vec!["page_a".to_string()],
        "the second page continues where the first stopped, with no repeat"
    );
    assert!(
        third.is_empty(),
        "an offset past the end is an empty page, not a wrap-around"
    );
    assert_eq!(
        second[0].total_spent_cents, 1350,
        "each paged row carries that buyer's own spend"
    );
}