//! Test harness for in-process integration tests.

pub(crate) mod client;
pub(crate) mod db;
pub(crate) mod email;
pub(crate) mod faults;
pub(crate) mod gitfixture;
pub(crate) mod seed;
pub(crate) mod storage;
pub(crate) mod stripe;

#[allow(unused_imports)]
pub(crate) use seed::{seed_project, seed_user};

/// CDN render base every test app uses unless `Opts::cdn_base_url` overrides it.
/// `Config::cdn_base_url` is required, so tests always have one.
pub(crate) const TEST_CDN_BASE: &str = "https://cdn.test";

/// Compute SHA-256 hash of a SyncKit API key (mirrors server's hash_api_key).
pub(crate) fn hash_api_key(api_key: &str) -> String {
    use sha2::Digest;
    hex::encode(sha2::Sha256::digest(api_key.as_bytes()))
}

use docengine::DocLoader;

/// The real published docs, parsed once per test binary.
///
/// The harness used to build a loader with `sections: vec![]` against `"."`, so
/// under test the site had no doc pages at all and nothing could cover `/docs`
/// or the docs half of the sitemap. Loading the real tree costs a parse, which
/// is why it is shared rather than rebuilt per `TestHarness::new`, and it goes
/// through `site_docs::build_doc_loader` so the config cannot drift from the
/// one production uses.
fn site_docs() -> Arc<DocLoader> {
    static DOCS: std::sync::OnceLock<Arc<DocLoader>> = std::sync::OnceLock::new();
    DOCS.get_or_init(|| {
        let assumptions = makenotwork::site_docs::load_assumptions()
            .expect("tests run from the crate root, where the assumptions file is");
        Arc::new(makenotwork::site_docs::build_doc_loader(assumptions))
    })
    .clone()
}
use makenotwork::config::{
    BuildConfig, Config, CreatorTierPricing, EmailWebhookConfig, IntegrationsConfig, ScanConfig,
    StripeConfig,
};
use makenotwork::db::UserId;
use makenotwork::email::{EmailClient, EmailConfig};
use makenotwork::payments::{PaymentProvider, StripeClient};
use makenotwork::scanning::ScanPipeline;
use makenotwork::{AppState, AppStateParts, AppStorage, build_app};
use sqlx::PgPool;
use std::sync::Arc;
use tower_sessions::cookie::SameSite;
use tower_sessions::cookie::time::Duration as CookieDuration;
use tower_sessions::{Expiry, SessionManagerLayer};
use tower_sessions_sqlx_store::PostgresStore;

use self::client::TestClient;
use self::db::TestDb;
use self::storage::InMemoryStorage;

/// Record a test's wall-clock duration to a shared timing file.
/// Call at the end of a test with the test name and start instant.
/// Results are appended to `/tmp/mnw-test-timing.csv` for analysis.
#[allow(dead_code)]
pub(crate) fn record_test_timing(name: &str, start: std::time::Instant) {
    let elapsed_ms = start.elapsed().as_millis();
    let line = format!("{name},{elapsed_ms}\n");
    use std::io::Write;
    if let Ok(mut f) = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open("/tmp/mnw-test-timing.csv")
    {
        let _ = f.write_all(line.as_bytes());
    }
}

/// Result of setting up a test creator with project and item.
#[allow(dead_code)]
pub(crate) struct CreatorSetup {
    pub user_id: UserId,
    pub project_id: String,
    pub item_id: String,
    pub slug: String,
}

/// Options for customizing a test harness build.
#[derive(Default)]
pub(crate) struct BuildOptions {
    pub storage: Option<Arc<InMemoryStorage>>,
    pub synckit_storage: Option<Arc<InMemoryStorage>>,
    pub stripe_client: Option<Arc<dyn PaymentProvider>>,
    pub scanner: Option<Arc<ScanPipeline>>,
    pub admin_user_id: Option<UserId>,
    pub existing_db: Option<TestDb>,
    pub postmark_webhook_token: Option<String>,
    pub postmark_broadcast_webhook_token: Option<String>,
    pub git_repos_path: Option<String>,
    pub build_trigger_token: Option<String>,
    pub postmark_inbound_webhook_token: Option<String>,
    pub mt_base_url: Option<String>,
    pub internal_shared_secret: Option<String>,
    pub cli_service_token: Option<String>,
    pub mock_email: Option<Arc<email::MockEmailTransport>>,
    /// Overrides the CDN render base. `None` uses [`TEST_CDN_BASE`]; the base is
    /// required config, so there is no "no CDN" mode to opt into.
    pub cdn_base_url: Option<String>,
    /// Site access gate. Defaults to `Open`; set to `FanPlusOrCreator` to test
    /// the testnot-style gate.
    pub access_gate: makenotwork::config::AccessGate,
    /// Delegated-login (SSO) provider config. `None` = local password form.
    pub sso: Option<makenotwork::config::SsoConfig>,
    /// Sticker monthly creator-tier price IDs. `None` = empty (creator-tier
    /// checkout is unconfigured and bails). Set to exercise creator-tier flows.
    pub creator_tier_prices:
        Option<std::collections::HashMap<makenotwork::db::CreatorTier, String>>,
    /// Rate-limit profile. `None` relaxes the limits, which is what almost every
    /// test wants: a test that logs in six times is not asking to be throttled.
    /// Set `Some(RateLimits::production())` to assert the thresholds that ship,
    /// which is what `workflows::rate_limiting` does.
    pub rate_limits: Option<makenotwork::constants::RateLimits>,
    /// Whether founder pricing is on offer. Defaults to closed, which is the
    /// state every pre-existing test was written against. Set it to exercise
    /// the surfaces that only exist while the window is open.
    pub founder_window_open: bool,
}

/// Full test harness: isolated database, in-process app, cookie-aware client.
#[allow(dead_code)]
pub(crate) struct TestHarness {
    pub client: TestClient,
    pub db: PgPool,
    pub storage: Option<Arc<InMemoryStorage>>,
    /// Mock email transport, if configured. Use `.sent()` to inspect sent emails.
    pub mock_email: Option<Arc<email::MockEmailTransport>>,
    /// Mock payment provider, if configured. Use `.checkouts()` to inspect created sessions.
    pub mock_stripe: Option<Arc<stripe::MockPaymentProvider>>,
    /// Pieces needed to drain the scan worker synchronously from tests
    /// (`drain_scan_jobs`). `None` when the harness wasn't built with a scanner.
    scan_deps: Option<ScanDeps>,
    /// The assembled state, kept so tests can drive scheduler jobs that take it.
    /// `build_app` borrows rather than consumes it, so this costs a cheap clone.
    state: makenotwork::AppState,
    _test_db: TestDb,
}

struct ScanDeps {
    s3: Arc<dyn makenotwork::storage::StorageBackend>,
    pipeline: Arc<ScanPipeline>,
    semaphore: Arc<tokio::sync::Semaphore>,
}

impl TestHarness {
    /// Spin up a fresh database, build the app, and return a ready-to-use harness.
    pub(crate) async fn new() -> Self {
        Self::build(BuildOptions::default()).await
    }

    /// Harness whose router carries the rate limits that ship, rather than the
    /// relaxed profile every other harness uses. This is the only way to assert
    /// the real thresholds, and the reason the limits are runtime config: under
    /// the old `fast-tests` feature the two profiles could not coexist in one
    /// run, so CI exercised the relaxed limiter and never the production one.
    #[allow(dead_code)]
    pub(crate) async fn with_production_rate_limits() -> Self {
        Self::build(BuildOptions {
            rate_limits: Some(makenotwork::constants::RateLimits::production()),
            ..Default::default()
        })
        .await
    }

    /// Harness with founder pricing on offer. Every other harness runs with the
    /// window shut, so this is the only way to reach the banner and the
    /// list/founder toggle on `/pricing`.
    #[allow(dead_code)]
    pub(crate) async fn with_founder_window_open() -> Self {
        Self::build(BuildOptions {
            founder_window_open: true,
            ..Default::default()
        })
        .await
    }

    /// Harness with in-memory storage backend.
    #[allow(dead_code)]
    pub(crate) async fn with_storage() -> Self {
        let mem = Arc::new(InMemoryStorage::new());
        Self::build(BuildOptions {
            storage: Some(mem),
            ..Default::default()
        })
        .await
    }

    /// Harness with SyncKit in-memory storage backend (for OTA tests).
    #[allow(dead_code)]
    pub(crate) async fn with_synckit_storage() -> Self {
        let mem = Arc::new(InMemoryStorage::new());
        Self::build(BuildOptions {
            synckit_storage: Some(mem),
            ..Default::default()
        })
        .await
    }

    /// Harness with in-memory storage + file scanning pipeline.
    #[allow(dead_code)]
    pub(crate) async fn with_storage_and_scanner() -> Self {
        let mem = Arc::new(InMemoryStorage::new());
        let scanner = Self::no_op_scanner();
        Self::build(BuildOptions {
            storage: Some(mem),
            scanner: Some(Arc::new(scanner)),
            ..Default::default()
        })
        .await
    }

    /// Harness with admin user + in-memory storage + file scanning pipeline.
    /// Returns (harness, admin_user_id).
    #[allow(dead_code)]
    pub(crate) async fn with_admin_storage_and_scanner() -> (Self, UserId) {
        let test_db = TestDb::new().await;
        let pool = test_db.pool.clone();
        let admin_id = Self::insert_admin_user(&pool).await;

        let mem = Arc::new(InMemoryStorage::new());
        let scanner = Self::no_op_scanner();
        let harness = Self::build(BuildOptions {
            storage: Some(mem),
            scanner: Some(Arc::new(scanner)),
            admin_user_id: Some(admin_id),
            existing_db: Some(test_db),
            ..Default::default()
        })
        .await;
        (harness, admin_id)
    }

    /// Harness with Stripe client configured (fake key, known webhook secrets).
    #[allow(dead_code)]
    pub(crate) async fn with_stripe() -> Self {
        let stripe_config = StripeConfig {
            secret_key: "sk_test_fake_key_for_testing".to_string(),
            webhook_secret: vec![stripe::TEST_WEBHOOK_SECRET.to_string()],
            webhook_secret_v2: Some(stripe::TEST_WEBHOOK_SECRET_V2.to_string()),
        };
        let stripe_client: Arc<dyn PaymentProvider> =
            Arc::new(StripeClient::new(&stripe_config).expect("test Stripe client builds"));
        Self::build(BuildOptions {
            stripe_client: Some(stripe_client),
            ..Default::default()
        })
        .await
    }

    /// Harness with mock Stripe + mock email for full payment flow testing.
    /// Access mocks via `harness.mock_stripe` and `harness.mock_email`.
    #[allow(dead_code)]
    pub(crate) async fn with_mocks() -> Self {
        let mock_stripe = Arc::new(stripe::MockPaymentProvider::new());
        let mock_email = Arc::new(email::MockEmailTransport::new());
        let mem = Arc::new(InMemoryStorage::new());
        let mut harness = Self::build(BuildOptions {
            storage: Some(mem),
            stripe_client: Some(mock_stripe.clone() as Arc<dyn PaymentProvider>),
            mock_email: Some(mock_email),
            ..Default::default()
        })
        .await;
        harness.mock_stripe = Some(mock_stripe);
        harness
    }

    /// Harness wired for creator-tier checkout: mock Stripe plus a configured
    /// sticker price for the Everything tier (the founder window stays closed,
    /// so checkout uses the sticker price ID, the mock ignores it anyway).
    /// Exposes `mock_stripe` for asserting on the trial passed to Stripe.
    #[allow(dead_code)]
    pub(crate) async fn with_creator_tier_checkout() -> Self {
        let mock_stripe = Arc::new(stripe::MockPaymentProvider::new());
        let mut prices = std::collections::HashMap::new();
        prices.insert(
            makenotwork::db::CreatorTier::Everything,
            "price_test_everything".to_string(),
        );
        let mut harness = Self::build(BuildOptions {
            storage: Some(Arc::new(InMemoryStorage::new())),
            stripe_client: Some(mock_stripe.clone() as Arc<dyn PaymentProvider>),
            creator_tier_prices: Some(prices),
            ..Default::default()
        })
        .await;
        harness.mock_stripe = Some(mock_stripe);
        harness
    }

    /// Harness with admin user configured. Returns (harness, admin_user_id).
    #[allow(dead_code)]
    pub(crate) async fn with_admin() -> (Self, UserId) {
        let test_db = TestDb::new().await;
        let pool = test_db.pool.clone();
        let admin_id = Self::insert_admin_user(&pool).await;

        let harness = Self::build(BuildOptions {
            admin_user_id: Some(admin_id),
            existing_db: Some(test_db),
            ..Default::default()
        })
        .await;
        (harness, admin_id)
    }

    /// Harness with Postmark webhook token configured.
    #[allow(dead_code)]
    pub(crate) async fn with_postmark() -> Self {
        Self::build(BuildOptions {
            postmark_webhook_token: Some("test-postmark-token".to_string()),
            postmark_broadcast_webhook_token: Some("test-broadcast-token".to_string()),
            ..Default::default()
        })
        .await
    }

    /// Harness with git repos path configured.
    #[allow(dead_code)]
    pub(crate) async fn with_git_repos(path: String) -> Self {
        Self::build(BuildOptions {
            git_repos_path: Some(path),
            ..Default::default()
        })
        .await
    }

    /// Insert an admin user and return the ID.
    async fn insert_admin_user(pool: &PgPool) -> UserId {
        let password_hash =
            makenotwork::auth::hash_password("password123").expect("hash_password for admin");
        sqlx::query_scalar(
            "INSERT INTO users (username, email, password_hash, email_verified)
             VALUES ('admin', 'admin@test.com', $1, true)
             RETURNING id",
        )
        .bind(&password_hash)
        .fetch_one(pool)
        .await
        .expect("Failed to insert admin user")
    }

    /// Create a no-op scan pipeline for tests.
    fn no_op_scanner() -> ScanPipeline {
        let scan_config = ScanConfig {
            clamav_socket: None,
            yara_rules_dir: "/nonexistent".to_string(),
            malwarebazaar_enabled: false,
            urlhaus_enabled: false,
            abuse_ch_auth_key: None,
            metadefender_api_key: None,
            yara_min_rule_files: 0,
            clamav_max_scan_bytes: None,
        };
        ScanPipeline::new(&scan_config).expect("ScanPipeline::new with no-op config")
    }

    /// Builder shared by all constructors. Public so workflow tests can use custom `BuildOptions`.
    pub(crate) async fn build(opts: BuildOptions) -> Self {
        // `main` does this for the real binary; tests never run `main`, and
        // without it the first Stripe connector build panics inside rustls.
        makenotwork::crypto::install_default_crypto_provider();

        let t0 = std::time::Instant::now();
        let test_db = match opts.existing_db {
            Some(db) => db,
            None => TestDb::new().await,
        };
        let pool = test_db.pool.clone();

        // Create session store (migration already applied in template DB)
        let session_store = PostgresStore::new(pool.clone());
        if !test_db.session_migrated {
            session_store
                .migrate()
                .await
                .expect("Failed to migrate session store");
        }

        let session_layer = SessionManagerLayer::new(session_store)
            .with_secure(false)
            .with_same_site(SameSite::Lax)
            .with_expiry(Expiry::OnInactivity(CookieDuration::days(1)));

        // Minimal config, no S3, no Stripe (those come from opts)
        let config = Config {
            host: "127.0.0.1".parse().unwrap(),
            port: 0,
            database_url: String::new(),
            host_url: std::sync::Arc::from("http://localhost:3000"),
            signing_secret: "test-signing-secret-for-integration-tests".to_string(),
            storage: None,
            synckit_storage: None,
            public_storage: None,
            stripe: None,
            admin_user_id: opts.admin_user_id,
            synckit_jwt_secret: Some("test-synckit-jwt-secret".to_string()),
            scan: None,
            cdn_base_url: opts
                .cdn_base_url
                .clone()
                .unwrap_or_else(|| TEST_CDN_BASE.to_string()),
            user_pages_host: std::sync::Arc::from("u.localhost"),
            access_gate: opts.access_gate,
            sso: opts.sso.clone(),
            rate_limits: opts
                .rate_limits
                .unwrap_or_else(makenotwork::constants::RateLimits::relaxed),
            build: BuildConfig {
                trigger_token: opts.build_trigger_token,
                host_linux: None,
                host_darwin: None,
                git_repos_path: opts.git_repos_path,
                git_ssh_host: None,
            },
            email_webhooks: EmailWebhookConfig {
                webhook_token: opts.postmark_webhook_token,
                broadcast_webhook_token: opts.postmark_broadcast_webhook_token,
                inbound_webhook_token: opts.postmark_inbound_webhook_token,
                enforce_sender_auth: true,
            },
            creator_pricing: CreatorTierPricing {
                fan_plus_price_id: None,
                tier_prices: opts.creator_tier_prices.unwrap_or_default(),
                tier_annual_prices: std::collections::HashMap::new(),
                tier_founder_prices: std::collections::HashMap::new(),
                tier_founder_annual_prices: std::collections::HashMap::new(),
                founder_window_open: opts.founder_window_open,
            },
            integrations: IntegrationsConfig {
                mt_base_url: None,
                wam_url: None,
                internal_shared_secret: opts.internal_shared_secret.clone(),
                cli_service_token: opts.cli_service_token.clone(),
                alerts_ingest_token: None,
            },
        };

        let mock_email_ref = opts.mock_email.clone();
        let email = if let Some(ref mock) = opts.mock_email {
            EmailClient::with_transport(mock.clone() as Arc<dyn makenotwork::email::EmailTransport>)
        } else {
            EmailClient::new(
                EmailConfig {
                    postmark_token: None,
                    from_address: "test@makenot.work".to_string(),
                    from_name: "Test".to_string(),
                },
                Some(pool.clone()),
            )
        };

        let rp_origin = url::Url::parse(&config.host_url).expect("test HOST_URL");
        let rp_id = rp_origin
            .host_str()
            .expect("test HOST_URL host")
            .to_string();
        let webauthn = Arc::new(
            webauthn_rs::WebauthnBuilder::new(&rp_id, &rp_origin)
                .expect("WebauthnBuilder")
                .rp_name("Test")
                .build()
                .expect("Webauthn"),
        );

        // Convert InMemoryStorage to trait object
        let storage = opts.storage;
        let s3 = storage
            .clone()
            .map(|s| s as Arc<dyn makenotwork::storage::StorageBackend>);
        let synckit_s3 = opts
            .synckit_storage
            .map(|s| s as Arc<dyn makenotwork::storage::StorageBackend>);
        // Public bucket shares the same in-memory backend as `s3` (one flat map,
        // no bucket isolation) so the cross-bucket image promote resolves.
        let public_s3 = s3.clone();

        // Route through the same `AppState::build` constructor production uses
        // (main.rs), so the derived in-memory state, start timestamps, the empty
        // cache maps, the concurrency semaphores, has a single source of truth.
        // The harness only supplies the externally-wired dependencies.
        let state = AppState::build(AppStateParts {
            db: pool.clone(),
            config,
            storage: AppStorage {
                s3,
                synckit_s3,
                public_s3,
            },
            stripe: opts.stripe_client,
            email,
            docs: site_docs(),
            tier_prices: {
                // Install the process-global TierPrices so handler code paths
                // that call CreatorTier::{price_cents,max_file_bytes,
                // max_storage_bytes} work under test (they read the global,
                // same as production). Idempotent across tests.
                makenotwork::tier_prices::TierPrices::install_test_default();
                makenotwork::tier_prices::TierPrices::global().clone()
            },
            runway_config: makenotwork::tier_prices::RunwayConfig::default(),
            fee_calculator: makenotwork::fee_calculator::FeeCalculator::load(
                "docs/business/assumptions.toml",
            ),
            scanner: opts.scanner,
            webauthn,
            syntax: None,
            mt_client: opts
                .mt_base_url
                .zip(opts.internal_shared_secret)
                .map(|(url, secret)| makenotwork::mt_client::MtClient::new(url, secret)),
            wam: None,
            domain_cache: Arc::new(dashmap::DashMap::new()),
            metrics_handle: None,
            page_view_tx: makenotwork::db::page_views::spawn_batcher(pool.clone()),
            bg: makenotwork::background::spawn_pool_detached(),
        });

        // Capture scan deps before `build_app` consumes `state`.
        let scan_deps = match (state.scanner.clone(), state.storage.s3.clone()) {
            (Some(pipeline), Some(s3)) => Some(ScanDeps {
                s3,
                pipeline,
                semaphore: state.limiters.scan_semaphore.clone(),
            }),
            _ => None,
        };

        let app = build_app(&state, session_layer);
        let client = TestClient::new(app);

        // Extract mock_stripe: if the stripe_client is a MockPaymentProvider,
        // we stored the Arc in BuildOptions.stripe_client. We can't downcast the
        // trait object, so with_mocks() stores the mock ref separately. For the
        // general build path, mock_stripe is None.
        let mock_stripe = None; // Set by with_mocks() post-build via direct field access

        let build_ms = t0.elapsed().as_millis();
        if build_ms > 1000 {
            eprintln!("[test-harness] SLOW harness build: {build_ms}ms");
        }

        TestHarness {
            client,
            db: pool,
            storage,
            mock_email: mock_email_ref,
            mock_stripe,
            scan_deps,
            state,
            _test_db: test_db,
        }
    }

    /// Run the orphaned-upload reaper once, synchronously.
    ///
    /// The scheduler drives this on a tick in production. Tests that want the
    /// reaper's failure branches (a transient S3 delete handed off to the
    /// durable queue, a best-effort multipart abort) need it to run at a known
    /// point instead, the same reason `drain_s3_deletions` exists.
    #[allow(dead_code)]
    pub(crate) async fn run_orphan_upload_reaper(&self) {
        makenotwork::scheduler::cleanup_orphaned_uploads_for_test(&self.state).await;
    }

    /// Sign up a new user via POST /join. Returns the user's ID.
    pub(crate) async fn signup(&mut self, username: &str, email: &str, password: &str) -> UserId {
        // Fetch a page first to establish session + CSRF
        self.client.fetch_csrf_token().await;

        let body = format!(
            "username={}&email={}&password={}",
            urlencoding::encode(username),
            urlencoding::encode(email),
            urlencoding::encode(password),
        );

        let resp = self.client.post_form("/join/step/account", &body).await;
        assert_eq!(
            resp.status, 200,
            "Signup failed with status {}: {}",
            resp.status, resp.text
        );

        // Login rotates the CSRF token, fetch the new one
        self.client.fetch_csrf_token().await;

        // Look up the user in the database
        sqlx::query_scalar::<_, UserId>("SELECT id FROM users WHERE username = $1")
            .bind(username)
            .fetch_one(&self.db)
            .await
            .expect("User not found after signup")
    }

    /// Grant creator permissions to a user via direct SQL.
    pub(crate) async fn grant_creator(&self, user_id: UserId) {
        sqlx::query("UPDATE users SET can_create_projects = true WHERE id = $1")
            .bind(user_id)
            .execute(&self.db)
            .await
            .expect("Failed to grant creator");
    }

    /// Trust a user for uploads via direct SQL.
    pub(crate) async fn trust_user(&self, user_id: UserId) {
        sqlx::query("UPDATE users SET upload_trusted = true WHERE id = $1")
            .bind(user_id)
            .execute(&self.db)
            .await
            .expect("Failed to trust user");
    }

    /// Give a user an active creator tier subscription via direct SQL.
    /// Also syncs the denormalized `creator_tier` column on the users table.
    pub(crate) async fn grant_tier(&self, user_id: UserId, tier: &str) {
        sqlx::query(
            r"INSERT INTO creator_subscriptions (user_id, stripe_subscription_id, stripe_customer_id, tier, status)
               VALUES ($1, 'sub_test_' || $1::text, 'cus_test_' || $1::text, $2, 'active')
               ON CONFLICT (user_id) DO UPDATE SET tier = $2, status = 'active'",
        )
        .bind(user_id)
        .bind(tier)
        .execute(&self.db)
        .await
        .expect("Failed to grant tier");

        sqlx::query("UPDATE users SET creator_tier = $2 WHERE id = $1")
            .bind(user_id)
            .bind(tier)
            .execute(&self.db)
            .await
            .expect("Failed to sync creator_tier");
    }

    /// Suspend a user via direct SQL.
    #[allow(dead_code)]
    pub(crate) async fn suspend_user(&self, user_id: UserId) {
        sqlx::query("UPDATE users SET suspended_at = NOW(), suspension_reason = 'test suspension' WHERE id = $1")
            .bind(user_id)
            .execute(&self.db)
            .await
            .expect("Failed to suspend user");
    }

    /// POST a single login attempt and return the response. Refreshes the
    /// CSRF token first so the new Manual-posture `/login` (Phase 2) accepts
    /// the form even when a previous `/logout` invalidated the cached token.
    /// Use this for negative-path login tests (lockout, suspended, wrong
    /// password) that need to inspect the response rather than asserting
    /// success like `login()` does.
    pub(crate) async fn failed_login_attempt(
        &mut self,
        login: &str,
        password: &str,
    ) -> client::TestResponse {
        self.client.fetch_csrf_token().await;
        let body = format!(
            "login={}&password={}",
            urlencoding::encode(login),
            urlencoding::encode(password),
        );
        self.client.post_form("/login", &body).await
    }

    /// Synchronously drain queued scan jobs by running the worker loop in-
    /// process until the queue is empty. Mirrors the production worker pool
    /// without spawning a background task, integration tests call this
    /// between upload-confirm and any assertion on `scan_status`.
    pub(crate) async fn drain_scan_jobs(&self) {
        let Some(ctx) = self.scan_worker_context() else {
            return;
        };
        // Hard cap to avoid an infinite loop if a job re-enqueues itself.
        for _ in 0..256 {
            match makenotwork::scanning::worker::process_next_for_test(&ctx).await {
                Ok(true) => {}
                Ok(false) => return,
                Err(e) => panic!("scan worker drain failed: {e}"),
            }
        }
        panic!("drain_scan_jobs did not terminate within 256 iterations");
    }

    /// Run one scan job and return its outcome instead of panicking on failure.
    ///
    /// `drain_scan_jobs` treats a failing job as a broken test, which is right
    /// for the happy paths but makes the worker's failure branch unobservable:
    /// a job whose download fails marks itself `failed` and resets its entity to
    /// `HeldForReview`, and no test could reach that while the only entry point
    /// panicked. `Ok(true)` ran a job, `Ok(false)` found an empty queue, `Err`
    /// carries the message the worker recorded in `last_error`.
    #[allow(dead_code)]
    pub(crate) async fn try_process_one_scan_job(&self) -> Result<bool, String> {
        let Some(ctx) = self.scan_worker_context() else {
            return Ok(false);
        };
        makenotwork::scanning::worker::process_next_for_test(&ctx)
            .await
            .map_err(|e| e.to_string())
    }

    /// The worker context both drain paths run against. `None` when the harness
    /// was not built with a scanner.
    fn scan_worker_context(&self) -> Option<makenotwork::scanning::worker::WorkerContext> {
        let deps = self.scan_deps.as_ref()?;
        Some(makenotwork::scanning::worker::WorkerContext {
            db: self.db.clone(),
            s3: deps.s3.clone(),
            pipeline: deps.pipeline.clone(),
            scan_semaphore: deps.semaphore.clone(),
            wam: None,
            bg: makenotwork::background::spawn_pool_detached(),
            // No Cloudflare purge in tests; quarantine still deletes from origin.
            cloudflare: None,
            cdn_base_url: std::sync::Arc::from(TEST_CDN_BASE),
            // OTA artifacts scan from the SyncKit bucket; tests share one backend.
            synckit_s3: Some(deps.s3.clone()),
            // Public bucket shares the same backend; image promotes copy here.
            public_s3: Some(deps.s3.clone()),
            config: self.state.config.clone(),
        })
    }

    /// Synchronously perform any queued S3 object deletions (`main` bucket).
    /// Handler-side deletes only enqueue to `pending_s3_deletions`; the actual
    /// delete is the scheduler's job in production. Tests that assert an object
    /// was removed from storage call this first (the deterministic mirror of
    /// the production retry worker). Returns the number of objects deleted.
    pub(crate) async fn drain_s3_deletions(&self) -> usize {
        let Some(storage) = self.storage.as_ref() else {
            return 0;
        };
        makenotwork::scheduler::drain_pending_s3_deletions_for_test(&self.db, storage.as_ref())
            .await
    }

    /// Log in as an existing user via POST /login. The client's session
    /// cookies are updated automatically.
    pub(crate) async fn login(&mut self, login: &str, password: &str) {
        // Fetch CSRF token first
        self.client.fetch_csrf_token().await;

        let body = format!(
            "login={}&password={}",
            urlencoding::encode(login),
            urlencoding::encode(password),
        );

        let resp = self.client.post_form("/login", &body).await;
        assert_eq!(
            resp.status, 303,
            "Login failed with status {}: {}",
            resp.status, resp.text
        );

        // Login rotates the CSRF token, fetch the new one
        self.client.fetch_csrf_token().await;
    }

    /// Create a test creator: signup, grant creator access, re-login.
    /// Uses password "password123" and email "{username}@test.com".
    pub(crate) async fn create_creator(&mut self, username: &str) -> UserId {
        let user_id = self
            .signup(username, &format!("{username}@test.com"), "password123")
            .await;
        self.grant_creator(user_id).await;
        self.client.post_form("/logout", "").await;
        self.login(username, "password123").await;
        user_id
    }

    /// Create a test creator with a project and one item. Creator is logged in afterward.
    /// Project slug: "{username}-proj". Returns all created IDs.
    pub(crate) async fn create_creator_with_item(
        &mut self,
        username: &str,
        item_type: &str,
        price_cents: i64,
    ) -> CreatorSetup {
        let user_id = self.create_creator(username).await;

        // Usernames may contain underscores (valid for accounts) but project
        // slugs may not, the slug charset is lowercase letters, digits, and
        // hyphens only. Sanitize so a `seller_vis` creator yields `seller-vis-proj`.
        let slug = format!("{}-proj", username.replace('_', "-"));
        let resp = self
            .client
            .post_form("/api/projects", &format!("slug={slug}&title=Test+Project"))
            .await;
        assert_eq!(resp.status, 200, "Create project failed: {}", resp.text);
        let project: serde_json::Value = resp.json();
        let project_id = project["id"].as_str().unwrap().to_string();

        let resp = self
            .client
            .post_form(
                &format!("/api/projects/{project_id}/items"),
                &format!("title=Test+Item&item_type={item_type}&price_cents={price_cents}"),
            )
            .await;
        assert_eq!(resp.status, 200, "Create item failed: {}", resp.text);
        let item: serde_json::Value = resp.json();
        let item_id = item["id"].as_str().unwrap().to_string();

        CreatorSetup {
            user_id,
            project_id,
            item_id,
            slug,
        }
    }

    /// Connect a user's Stripe account via direct SQL.
    /// Sets stripe_account_id, stripe_charges_enabled, and stripe_onboarding_complete.
    /// Use after `create_creator()` for tests that need a Stripe-connected seller.
    pub(crate) async fn connect_stripe(&self, user_id: UserId, account_id: &str) {
        sqlx::query(
            "UPDATE users SET stripe_account_id = $2, stripe_charges_enabled = true, \
             stripe_onboarding_complete = true, stripe_payouts_enabled = true WHERE id = $1",
        )
        .bind(user_id)
        .bind(account_id)
        .execute(&self.db)
        .await
        .expect("Failed to connect Stripe");
    }

    /// Publish both a project and an item.
    pub(crate) async fn publish_project_and_item(&mut self, project_id: &str, item_id: &str) {
        self.client
            .put_json(
                &format!("/api/projects/{project_id}"),
                r#"{"is_public": true}"#,
            )
            .await;
        self.client
            .put_form(&format!("/api/items/{item_id}"), "is_public=true")
            .await;
    }
}