//! Scheduled cleanup jobs: sandbox expiry, terminated accounts, content
//! removal, IP scrubbing, stale pending transactions, orphaned uploads, cart
//! items, soft-deleted item purges, and pending S3 deletion retries.

use crate::AppState;
use crate::constants;
use crate::db;
use crate::storage::S3DeleteAuthority;

/// Hourly: purge `scan_jobs` rows in a terminal state older than the
/// retention window. Queued/running rows are not touched.
#[tracing::instrument(skip_all, name = "scheduler::purge_old_scan_jobs")]
pub(super) async fn purge_old_scan_jobs(state: &AppState) {
    let window = chrono::Duration::days(constants::SCAN_JOB_RETENTION_DAYS as i64);
    match db::scan_jobs::purge_old_terminal(&state.db, window).await {
        Ok(n) => {
            if n > 0 {
                tracing::info!(purged = n, "scan_jobs retention sweep");
            }
            let _ = db::scheduler_jobs::record_job_run(&state.db, "scan_jobs_retention", n as i64)
                .await;
        }
        Err(e) => tracing::error!(error = ?e, "scan_jobs retention sweep failed"),
    }
}

/// Drain `ids` through `cleanup_user_s3_and_delete` with bounded concurrency
/// (`CLEANUP_PARALLELISM` in flight). Returns the count of successful deletes.
///
/// The single bounded-drain primitive shared by every per-user cleanup sweep,
/// each cleanup holds a few pool conns + an S3 delete-prefix, so the cap (4)
/// leaves the bulk of the ~25-conn pool for request handlers even on a
/// mass-expiry tick. Hoisting this means the cap can't be forgotten or drift
/// between sweeps (Run #12: the two former copies were the predicted recurrence
/// of the "guard-by-convention across siblings" meta-pattern).
async fn drain_cleanup_bounded(
    state: &AppState,
    ids: Vec<db::UserId>,
    event: &'static str,
    label: &'static str,
) -> i64 {
    const CLEANUP_PARALLELISM: usize = 4;
    let mut set = tokio::task::JoinSet::new();
    let mut iter = ids.into_iter();
    let mut deleted = 0i64;
    loop {
        while set.len() < CLEANUP_PARALLELISM {
            match iter.next() {
                Some(uid) => {
                    let state = state.clone();
                    set.spawn(async move {
                        cleanup_user_s3_and_delete(&state, uid, event, label).await
                    });
                }
                None => break,
            }
        }
        match set.join_next().await {
            Some(Ok(true)) => deleted += 1,
            Some(Ok(false)) => {}
            Some(Err(e)) => tracing::warn!(error = ?e, %label, "cleanup task panicked"),
            None => break,
        }
    }
    deleted
}

/// Delete expired sandbox accounts and their S3 objects.
#[tracing::instrument(skip_all, name = "scheduler::cleanup_sandbox_accounts")]
pub(super) async fn cleanup_sandbox_accounts(state: &AppState) {
    let expired_ids = match db::users::get_expired_sandbox_ids(&state.db).await {
        Ok(ids) => ids,
        Err(e) => {
            tracing::error!(error = ?e, "failed to query expired sandbox accounts");
            return;
        }
    };

    // Drained inline on the tick (sandbox sweep is cheap + frequent); bounded at
    // 4 by the shared helper.
    let deleted = drain_cleanup_bounded(state, expired_ids, "sandbox_expired", "sandbox").await;
    let _ = db::scheduler_jobs::record_job_run(&state.db, "sandbox_cleanup", deleted).await;
}

/// Clean up a user's S3 objects, git repos, and CASCADE-delete the user row.
///
/// Shared between sandbox, terminated, and content-removal account cleanup.
/// S3 objects are deleted first (before CASCADE removes the DB rows that reference them).
async fn cleanup_user_s3_and_delete(
    state: &AppState,
    user_id: db::UserId,
    event: &str,
    label: &str,
) -> bool {
    // Resolve everything we need from the DB once, the enqueue list and the
    // delete list must come from the same snapshot, or a project/app created
    // between calls will be enqueued but not deleted (or vice versa).
    // A transient DB error here must ABORT, not degrade to an empty list: an
    // empty key set would enqueue nothing for `projects/{pid}/` / `{app_id}/`,
    // then the CASCADE delete below would drop the rows and leave those S3
    // objects orphaned forever. Return false so the sweep retries next tick
    // (ultra-fuzz Run 13 Storage: orphan-on-transient-error).
    let project_ids = match db::projects::get_project_ids_for_user(&state.db, user_id).await {
        Ok(v) => v,
        Err(e) => {
            tracing::error!(error = ?e, %user_id, "{label}: failed to list projects; aborting cleanup to avoid orphaning S3 objects");
            return false;
        }
    };
    let sync_apps = match db::synckit::get_sync_apps_by_creator(&state.db, user_id).await {
        Ok(v) => v,
        Err(e) => {
            tracing::error!(error = ?e, %user_id, "{label}: failed to list sync apps; aborting cleanup to avoid orphaning S3 objects");
            return false;
        }
    };

    let user_prefix = format!("{user_id}/");
    let main = crate::storage::S3Bucket::Main.as_str().to_string();
    let synckit = crate::storage::S3Bucket::Synckit.as_str().to_string();
    let public = crate::storage::S3Bucket::Public.as_str().to_string();
    let mut keys: Vec<(String, String)> = Vec::new();
    keys.push((user_prefix.clone(), main.clone()));
    // The user's CDN-served image content (`{user_id}/c/{sha}`) lives in the
    // public bucket, same key prefix, different bucket, so sweep it there too
    // or those objects orphan on account deletion.
    keys.push((user_prefix.clone(), public.clone()));
    for pid in &project_ids {
        keys.push((format!("projects/{pid}/"), main.clone()));
    }
    for app in &sync_apps {
        keys.push((format!("{}/", app.id), synckit.clone()));
        keys.push((format!("ota/{}/", app.id), synckit.clone()));
    }

    // Enqueue all keys before any destructive work
    if let Err(e) = db::pending_s3_deletions::enqueue_deletions(&state.db, &keys, label).await {
        tracing::error!(error = ?e, %user_id, "{label}: failed to enqueue S3 deletions, aborting cleanup");
        return false;
    }

    let auth = S3DeleteAuthority::new();
    if let Some(ref s3) = state.storage.s3 {
        if let Err(e) = s3.delete_prefix(&auth, &user_prefix).await {
            tracing::warn!(error = ?e, %user_id, "{label}: failed to delete user S3 objects");
        }
        for pid in &project_ids {
            let proj_prefix = format!("projects/{pid}/");
            if let Err(e) = s3.delete_prefix(&auth, &proj_prefix).await {
                tracing::warn!(error = ?e, %user_id, %pid, "{label}: failed to delete project S3 objects");
            }
        }
    }

    // The user's CDN-served image content lives under the same `{user_id}/`
    // prefix in the public bucket; sweep it there too (the enqueue above is the
    // durable backstop if this fast path fails).
    if let Some(ref public_s3) = state.storage.public_s3
        && let Err(e) = public_s3.delete_prefix(&auth, &user_prefix).await
    {
        tracing::warn!(error = ?e, %user_id, "{label}: failed to delete user public-bucket objects");
    }

    if let Some(ref synckit_s3) = state.storage.synckit_s3 {
        for app in &sync_apps {
            let blob_prefix = format!("{}/", app.id);
            if let Err(e) = synckit_s3.delete_prefix(&auth, &blob_prefix).await {
                tracing::warn!(error = ?e, %user_id, app_id = %app.id, "{label}: failed to delete SyncKit blobs");
            }
            let ota_prefix = format!("ota/{}/", app.id);
            if let Err(e) = synckit_s3.delete_prefix(&auth, &ota_prefix).await {
                tracing::warn!(error = ?e, %user_id, app_id = %app.id, "{label}: failed to delete OTA artifacts");
            }
        }
    }

    // Git repos on disk
    if let Some(ref git_root) = state.config.build.git_repos_path
        && let Ok(Some(user)) = db::users::get_user_by_id(&state.db, user_id).await
    {
        cleanup_git_repos_on_disk(git_root, &user.username, user_id).await;
    }

    // CASCADE delete user row (+ purge domain_cache via the coupled entry point)
    if let Err(e) = crate::delete_user_account(&state.db, &state.caches, user_id).await {
        tracing::error!(error = ?e, %user_id, "{label}: failed to delete account");
        false
    } else {
        tracing::info!(%user_id, event, "{label}: account cleaned up");
        true
    }
}

/// Remove a user's bare git repositories from disk.
///
/// Must be called before `delete_user` (which CASCADE-deletes the git_repos rows).
/// Best-effort: logs warnings on failure but does not block account deletion.
/// Runs blocking I/O on a dedicated thread to avoid stalling the Tokio runtime.
#[tracing::instrument(skip_all, name = "scheduler::cleanup_git_repos_on_disk")]
pub(super) async fn cleanup_git_repos_on_disk(
    git_repos_path: &str,
    username: &str,
    user_id: db::UserId,
) {
    let user_git_dir = std::path::Path::new(git_repos_path).join(username);
    if user_git_dir.exists() {
        let path = user_git_dir.clone();
        match tokio::task::spawn_blocking(move || std::fs::remove_dir_all(&path)).await {
            Ok(Ok(())) => {
                tracing::info!(%user_id, path = %user_git_dir.display(), "deleted git repos from disk");
            }
            Ok(Err(e)) => {
                tracing::warn!(error = ?e, %user_id, path = %user_git_dir.display(), "failed to delete git repos from disk");
            }
            Err(e) => {
                tracing::warn!(error = ?e, %user_id, path = %user_git_dir.display(), "git repo cleanup task panicked");
            }
        }
    }
}

/// Delete accounts that were terminated >30 days ago.
///
/// Per-user S3 sweeps + git-repo removal + CASCADE delete are spawned off the
/// scheduler's advisory-lock-held tick so a backlog of expired accounts
/// (5 creators × 20 projects of multi-page `delete_prefix` calls) doesn't
/// extend the lock hold past `TICK_DURATION_ALERT_SECS`. Each step in
/// `cleanup_user_s3_and_delete` is idempotent: `delete_prefix` on a missing
/// prefix is a no-op, the CASCADE `delete_user` on a non-existent row affects
/// zero rows. If a second tick fires before all spawned cleanups complete
/// (extremely large accounts), the next `get_expired_terminated_ids` query
/// excludes already-deleted rows; the worst-case race is one extra harmless
/// no-op call against an in-flight target.
#[tracing::instrument(skip_all, name = "scheduler::delete_expired_terminated_accounts")]
pub(super) async fn delete_expired_terminated_accounts(state: &AppState) {
    spawn_expired_account_cleanups(
        state,
        db::users::get_expired_terminated_ids(&state.db).await,
        "termination_expired",
        "terminated account",
        "terminated_account_cleanup",
    )
    .await;
}

/// Delete creator accounts whose 90-day content removal grace period has expired.
/// Off-lock spawn pattern, see `delete_expired_terminated_accounts`.
#[tracing::instrument(skip_all, name = "scheduler::delete_expired_content_removal_accounts")]
pub(super) async fn delete_expired_content_removal_accounts(state: &AppState) {
    spawn_expired_account_cleanups(
        state,
        db::users::get_expired_content_removal_ids(&state.db).await,
        "content_removal_expired",
        "content removal",
        "content_removal_cleanup",
    )
    .await;
}

async fn spawn_expired_account_cleanups(
    state: &AppState,
    fetched: Result<Vec<db::UserId>, crate::error::AppError>,
    event: &'static str,
    label: &'static str,
    job_name: &'static str,
) {
    let expired_ids = match fetched {
        Ok(ids) if ids.is_empty() => {
            let _ = db::scheduler_jobs::record_job_run(&state.db, job_name, 0).await;
            return;
        }
        Ok(ids) => ids,
        Err(e) => {
            tracing::error!(error = ?e, %job_name, "failed to query expired accounts");
            return;
        }
    };

    let scheduled = expired_ids.len() as i64;

    // Run the per-user sweeps in a SINGLE supervisor task spawned off the tick,
    // draining through the shared bounded-concurrency helper. This keeps the
    // off-lock property the doc comment relies on (the scheduler tick returns
    // immediately, never holding the advisory lock across the sweeps) AND caps
    // how many sweeps run at once. A bare `tokio::spawn` per expired account had
    // no cap: a mass expiry (terminated-account or 90-day content-removal purge)
    // could fan out N concurrent tasks, each holding several pool connections
    // plus an S3 delete-prefix, and drain the connection pool out from under
    // request handlers (Run #11 SERIOUS).
    let supervisor_state = state.clone();
    tokio::spawn(async move {
        drain_cleanup_bounded(&supervisor_state, expired_ids, event, label).await;
    });

    // Record the count *scheduled*, not *deleted*: the deletes finish out of band.
    // Operators read this metric to confirm the scheduler saw the work, not to
    // count completed sweeps. cleanup_user_s3_and_delete logs its own success/
    // failure per user.
    let _ = db::scheduler_jobs::record_job_run(&state.db, job_name, scheduled).await;
}

/// Delete pending transactions older than 25 hours and release promo code reservations.
#[tracing::instrument(skip_all, name = "scheduler::cleanup_stale_pending_transactions")]
pub(super) async fn cleanup_stale_pending_transactions(state: &AppState) {
    let promo_ids =
        match db::transactions::cleanup_stale_pending(&state.db, chrono::Duration::hours(25)).await
        {
            Ok(ids) => ids,
            Err(e) => {
                tracing::error!(error = ?e, "failed to clean up stale pending transactions");
                return;
            }
        };

    // Cart checkouts produce N pending-tx rows that share a promo_code_id;
    // release once per reservation, not once per row.
    let unique_promo_ids: std::collections::HashSet<_> = promo_ids.into_iter().flatten().collect();

    let mut released = 0i64;
    for pc_id in unique_promo_ids {
        if let Err(e) = db::promo_codes::release_use_count(&state.db, pc_id).await {
            tracing::warn!(promo_code_id = %pc_id, error = ?e, "failed to release promo code use count");
        } else {
            released += 1;
        }
    }

    if released > 0 {
        tracing::info!(
            released,
            "released promo code reservations from stale pending transactions"
        );
    }
    let _ = db::scheduler_jobs::record_job_run(&state.db, "stale_pending_cleanup", released).await;
}

/// NULL out IP addresses older than 30 days in user_sessions.
#[tracing::instrument(skip_all, name = "scheduler::scrub_stale_ip_addresses")]
pub(super) async fn scrub_stale_ip_addresses(state: &AppState) {
    let cutoff = chrono::Utc::now() - chrono::Duration::days(30);

    match sqlx::query(
        "UPDATE user_sessions SET ip_address = NULL WHERE ip_address IS NOT NULL AND created_at < $1",
    )
    .bind(cutoff)
    .execute(&state.db)
    .await
    {
        Ok(r) => {
            if r.rows_affected() > 0 {
                tracing::info!(scrubbed = r.rows_affected(), "scrubbed stale IP addresses (30-day retention)");
            }
            let _ = db::scheduler_jobs::record_job_run(&state.db, "ip_scrub", r.rows_affected() as i64).await;
        }
        Err(e) => tracing::error!(error = ?e, "failed to scrub IPs from user_sessions"),
    }
}

/// Permanently delete items that were soft-deleted more than 7 days ago.
/// Cleans up S3 objects (item files + version files) and decrements storage
/// before DB deletion to prevent orphaned storage and accounting drift.
///
/// Accepted residual (ultra-fuzz Run 4 Perf, decision 2026-06-23): this gathers
/// all expired-item S3 keys into one Vec before the batch delete. It is a daily
/// cron off the request path; steady-state (7-day window) is small, so the only
/// large allocation is transient, after a rare mass-delete event. Bounding it
/// per-tick means coupling a LIMIT on the key-gather to the same slice as the
/// CASCADE delete; deferred as low-value for a cron path. Revisit if it matters.
#[tracing::instrument(skip_all, name = "scheduler::purge_expired_deleted_items")]
pub(super) async fn purge_expired_deleted_items(state: &AppState) {
    // Collect S3 keys from items AND their versions before CASCADE delete destroys the data
    let mut all_s3_keys: Vec<(String, String)> = Vec::new();

    // Item keys are audio/video (private bucket) mixed with the item cover
    // (public bucket); enqueue each under both so the reaper hits the right one
    // (see `both_bucket_delete`).
    match db::items::get_expired_deleted_item_s3_keys(&state.db).await {
        Ok(keys) => {
            for key in &keys {
                all_s3_keys.extend(crate::storage::both_bucket_delete(key));
            }
        }
        Err(e) => {
            tracing::error!(error = ?e, "failed to query item S3 keys for items pending purge");
        }
    }

    match db::items::get_expired_deleted_item_version_s3_keys(&state.db).await {
        Ok(keys) => {
            for key in &keys {
                // Version downloads are gated media, always the private bucket.
                all_s3_keys.push((
                    key.clone(),
                    crate::storage::S3Bucket::Main.as_str().to_string(),
                ));
            }
        }
        Err(e) => {
            tracing::error!(error = ?e, "failed to query version S3 keys for items pending purge");
        }
    }

    // Gallery images (item_images) cascade away with the purged items; collect
    // their keys too or the objects orphan with no durable record (Run #18 B2).
    // CDN-served → public bucket post-promote (or staging in main); enqueue both.
    match db::gallery_images::s3_keys_for_expired_purged_items(&state.db).await {
        Ok(keys) => {
            for key in &keys {
                all_s3_keys.extend(crate::storage::both_bucket_delete(key));
            }
        }
        Err(e) => {
            tracing::error!(error = ?e, "failed to query gallery S3 keys for items pending purge");
        }
    }

    // Enqueue all keys as a durable safety net before any destructive work
    if !all_s3_keys.is_empty()
        && let Err(e) = db::pending_s3_deletions::enqueue_deletions(
            &state.db,
            &all_s3_keys,
            "purge_deleted_items",
        )
        .await
    {
        tracing::error!(error = ?e, "failed to enqueue S3 deletions for purged items, aborting purge");
        return;
    }

    if let Some(ref s3) = state.storage.s3
        && !all_s3_keys.is_empty()
    {
        let keys_only: Vec<crate::storage::S3Key> = all_s3_keys
            .iter()
            .map(|(k, _)| crate::storage::S3Key::from_stored(k))
            .collect();
        if let Err(e) = s3
            .delete_objects(&S3DeleteAuthority::new(), &keys_only)
            .await
        {
            tracing::warn!(error = ?e, "batch S3 delete failed for purged items; pending_s3_deletions queue will retry");
        }
        tracing::info!(
            count = all_s3_keys.len(),
            "deleted S3 objects for purged items"
        );
    }

    // Decrement each affected user's storage AND purge the items in ONE
    // transaction. Previously these were two pool calls: a crash between them
    // left the items un-purged but never re-measured, so the next tick decremented
    // the SAME items again → `storage_used_bytes` under-counted (fuzz-2026-07-06
    // LOW). Folding them means a crash rolls both back and the next tick redoes the
    // pair cleanly. The measure and the DELETE share the transaction's fixed
    // `NOW()`, so both act on exactly the same expired set.
    let mut tx = match state.db.begin().await {
        Ok(tx) => tx,
        Err(e) => {
            tracing::error!(error = ?e, "failed to open tx for item purge; will retry next tick");
            return;
        }
    };

    match db::items::get_expired_deleted_item_storage_by_user(&mut *tx).await {
        Ok(user_sizes) => {
            for (user_id, total_bytes) in &user_sizes {
                if *total_bytes > 0
                    && let Err(e) =
                        db::creator_tiers::decrement_storage_used(&mut *tx, *user_id, *total_bytes)
                            .await
                {
                    // A failed decrement poisons the tx; abort the whole purge so
                    // items are never deleted without their storage being credited
                    // back. The next tick retries the pair atomically.
                    tracing::warn!(user_id = %user_id, bytes = total_bytes, error = ?e,
                        "failed to decrement storage for purged items; aborting purge tx");
                    return;
                }
            }
        }
        Err(e) => {
            tracing::error!(error = ?e, "failed to query storage sizes for items pending purge; aborting purge tx");
            return;
        }
    }

    let purged = match db::items::purge_expired_deleted_items(&mut *tx).await {
        Ok(n) => n,
        Err(e) => {
            tracing::error!(error = ?e, "failed to purge expired soft-deleted items; aborting purge tx");
            return;
        }
    };

    if let Err(e) = tx.commit().await {
        tracing::error!(error = ?e, "failed to commit item purge tx; will retry next tick");
        return;
    }

    if purged > 0 {
        tracing::info!(deleted = purged, "purged expired soft-deleted items");
    }
    let _ = db::scheduler_jobs::record_job_run(&state.db, "soft_delete_purge", purged as i64).await;
}

/// Outcome of a guarded single-key orphan S3 delete.
enum GuardedDelete {
    /// Object deleted (S3 returned Ok).
    Deleted,
    /// Skipped: a live DB row still references the key (delete-then-reupload
    /// race). The caller should clear its queue/record without deleting.
    SkippedLive,
    /// S3 delete failed (already logged). The caller decides retry vs. clear.
    Failed,
}

/// The single funnel for deleting one non-prefix orphan S3 object.
///
/// Performs the `is_s3_key_live` check FIRST, so an object that a fresh upload
/// reclaimed under the same key is never torpedoed. Both the pending-deletions
/// retry worker and the orphaned-upload reaper route every single-key delete
/// through here, neither can delete a key without the live-check. (Storage S2
/// / CHRONIC 2: the worker had this guard inline; its sibling reaper did not,
/// and could delete a now-live deterministic-key object.)
async fn delete_orphan_key_guarded(
    db: &sqlx::PgPool,
    s3: &dyn crate::storage::StorageBackend,
    bucket: &str,
    s3_key: &str,
) -> GuardedDelete {
    match db::pending_s3_deletions::is_s3_key_live(db, bucket, s3_key).await {
        Ok(true) => {
            tracing::info!(s3_key = %s3_key, bucket = %bucket,
                "orphan S3 delete skipped, key reclaimed by a live row (delete-then-reupload)");
            return GuardedDelete::SkippedLive;
        }
        Ok(false) => {}
        Err(e) => {
            tracing::warn!(s3_key = %s3_key, error = ?e,
                "live-key check failed; proceeding with delete attempt");
        }
    }
    match s3
        .delete_object(
            &S3DeleteAuthority::new(),
            &crate::storage::S3Key::from_stored(s3_key),
        )
        .await
    {
        Ok(()) => GuardedDelete::Deleted,
        Err(e) => {
            tracing::warn!(s3_key = %s3_key, bucket = %bucket, error = ?e, "orphan S3 delete failed");
            GuardedDelete::Failed
        }
    }
}

/// Delete S3 objects from presigned uploads that were never confirmed (>24h old).
#[tracing::instrument(skip_all, name = "scheduler::cleanup_orphaned_uploads")]
pub(super) async fn cleanup_orphaned_uploads(state: &AppState) {
    let stale = match db::pending_uploads::get_stale_pending_uploads(
        &state.db,
        chrono::Duration::hours(24),
    )
    .await
    {
        Ok(rows) if rows.is_empty() => {
            let _ =
                db::scheduler_jobs::record_job_run(&state.db, "orphaned_upload_cleanup", 0).await;
            return;
        }
        Ok(rows) => rows,
        Err(e) => {
            tracing::error!(error = ?e, "failed to query stale pending uploads");
            return;
        }
    };

    let mut cleaned = 0i64;
    let mut aborted = 0i64;
    // Carry (key, bucket) pairs: the deletion is bucket-scoped so a key present
    // in two buckets only clears the record for the bucket we reaped (B7).
    let mut keys_to_delete: Vec<(String, String)> = Vec::with_capacity(stale.len());
    // Transient S3 delete failures handed off to the durable deletion queue
    // (which has a retry + dead-letter backstop) instead of dropping the only
    // tracking row, clearing it on a transient failure permanently leaked the
    // object (Run #2 Storage SERIOUS; the sibling retry worker already routes
    // failures into this ladder).
    let mut failed_keys: Vec<(String, String)> = Vec::new();

    for (s3_key, bucket) in &stale {
        let s3_client = match crate::storage::S3Bucket::from_db_str(bucket) {
            crate::storage::S3Bucket::Synckit => state.storage.synckit_s3.as_ref(),
            crate::storage::S3Bucket::Public => state.storage.public_s3.as_ref(),
            crate::storage::S3Bucket::Main => state.storage.s3.as_ref(),
        };
        if let Some(s3) = s3_client {
            // A multipart session that was started and never completed leaves NO
            // object to delete, only uploaded parts that S3 bills for until they
            // are aborted, so the delete below is a no-op against it and the
            // parts would leak forever. Abort first, keyed off S3's own list so a
            // session whose upload_id was never recorded is still caught.
            //
            // Best-effort: a failed abort must not block the object delete or
            // strand the tracking row. The bucket lifecycle rule
            // (AbortIncompleteMultipartUpload) is the backstop for anything
            // missed here.
            aborted += abort_orphan_multipart_sessions(s3.as_ref(), s3_key).await;

            // Route through the guarded funnel: if a confirm already reclaimed
            // this deterministic key, the live row owns the object and we must
            // NOT delete it, just clear the stale pending_uploads record.
            match delete_orphan_key_guarded(&state.db, s3.as_ref(), bucket, s3_key).await {
                GuardedDelete::Deleted => {
                    cleaned += 1;
                    keys_to_delete.push((s3_key.clone(), bucket.clone()));
                }
                // Live row owns the key now: clear the stale record, keep object.
                GuardedDelete::SkippedLive => keys_to_delete.push((s3_key.clone(), bucket.clone())),
                // Delete failed (transient): hand off to the durable deletion
                // queue rather than dropping the tracking row and leaking.
                GuardedDelete::Failed => failed_keys.push((s3_key.clone(), bucket.clone())),
            }
        } else {
            // S3 not configured for this bucket; remove the DB record anyway
            keys_to_delete.push((s3_key.clone(), bucket.clone()));
        }
    }

    // Enqueue transient failures into the durable deletion queue BEFORE clearing
    // their tracking rows, so a key is never dropped by both. If the handoff
    // itself fails, leave the pending_uploads rows for the next reaper tick
    // rather than leaking.
    if !failed_keys.is_empty() {
        match db::pending_s3_deletions::enqueue_deletions(
            &state.db,
            &failed_keys,
            "orphan-reaper-retry",
        )
        .await
        {
            Ok(()) => {
                tracing::warn!(
                    count = failed_keys.len(),
                    "orphan reaper: S3 delete failed; handed off to durable deletion queue for retry"
                );
                keys_to_delete.extend(failed_keys);
            }
            Err(e) => {
                tracing::error!(
                    error = ?e, count = failed_keys.len(),
                    "orphan reaper: could not enqueue transient failures to deletion queue; leaving tracking rows for next tick"
                );
            }
        }
    }

    if !keys_to_delete.is_empty()
        && let Err(e) =
            db::pending_uploads::delete_pending_uploads(&state.db, &keys_to_delete).await
    {
        tracing::error!(error = ?e, "failed to delete pending upload records");
    }

    if cleaned > 0 || aborted > 0 {
        tracing::info!(cleaned, aborted, "cleaned up orphaned presigned uploads");
    }
    let _ = db::scheduler_jobs::record_job_run(&state.db, "orphaned_upload_cleanup", cleaned).await;
}

/// Abort every in-progress multipart session for `s3_key`, returning how many
/// were aborted.
///
/// Best-effort by design: the caller still deletes the object and clears the
/// tracking row regardless, and the bucket's `AbortIncompleteMultipartUpload`
/// lifecycle rule catches whatever fails here. Returning a count rather than a
/// Result keeps a storage hiccup from stranding the rest of the reap.
/// Public via a single re-export on `scheduler` (the module itself stays
/// private) so integration tests can drive it against a test backend without a
/// database or an `AppState`.
pub async fn abort_orphan_multipart_sessions(
    s3: &dyn crate::storage::StorageBackend,
    s3_key: &str,
) -> i64 {
    let upload_ids = match s3.list_multipart_uploads_for_key(s3_key).await {
        Ok(ids) => ids,
        Err(e) => {
            tracing::warn!(
                s3_key = %s3_key, error = ?e,
                "orphan reaper: could not list multipart sessions; lifecycle rule is the backstop"
            );
            return 0;
        }
    };

    let key = crate::storage::S3Key::from_stored(s3_key);
    let mut aborted = 0i64;
    for upload_id in upload_ids {
        match s3.abort_multipart_upload(&key, &upload_id).await {
            Ok(()) => {
                aborted += 1;
                tracing::info!(
                    s3_key = %s3_key, %upload_id,
                    "orphan reaper: aborted an abandoned multipart session"
                );
            }
            Err(e) => tracing::warn!(
                s3_key = %s3_key, %upload_id, error = ?e,
                "orphan reaper: failed to abort multipart session; lifecycle rule is the backstop"
            ),
        }
    }
    aborted
}

/// Remove stale cart items (>30 days old) and items that became unavailable.
#[tracing::instrument(skip_all, name = "scheduler::cleanup_cart_items")]
pub(super) async fn cleanup_cart_items(state: &AppState) {
    match db::cart::cleanup_stale_cart_items(&state.db, chrono::Duration::days(30)).await {
        Ok(n) if n > 0 => tracing::info!(removed = n, "cleaned up stale cart items"),
        Err(e) => tracing::error!(error = ?e, "failed to clean up stale cart items"),
        _ => {}
    }
    match db::cart::cleanup_unavailable_cart_items(&state.db).await {
        Ok(n) if n > 0 => tracing::info!(removed = n, "cleaned up unavailable cart items"),
        Err(e) => tracing::error!(error = ?e, "failed to clean up unavailable cart items"),
        _ => {}
    }
}

/// Retry stale pending S3 deletions (older than 10 minutes, batch of 100).
#[tracing::instrument(skip_all, name = "scheduler::retry_pending_s3_deletions")]
pub(super) async fn retry_pending_s3_deletions(state: &AppState) {
    let stale = match db::pending_s3_deletions::get_stale_pending(
        &state.db,
        chrono::Duration::minutes(10),
        100,
    )
    .await
    {
        Ok(rows) => rows,
        Err(e) => {
            tracing::error!(error = ?e, "failed to fetch stale pending S3 deletions");
            return;
        }
    };

    if stale.is_empty() {
        let _ = db::scheduler_jobs::record_job_run(&state.db, "s3_deletion_retry", 0).await;
        return;
    }

    let mut completed_ids = Vec::new();
    let mut dead_letter_ids = Vec::new();
    for row in &stale {
        if row.attempts >= 10 {
            tracing::error!(s3_key = %row.s3_key, bucket = %row.bucket, source = %row.source, attempts = row.attempts,
                "S3 deletion dead-lettered after 10 attempts, moving to dead-letter table for manual triage");
            dead_letter_ids.push(row.id);
            continue;
        } else if row.attempts >= 5 {
            tracing::warn!(s3_key = %row.s3_key, bucket = %row.bucket, source = %row.source, attempts = row.attempts,
                "S3 deletion stuck after 5+ attempts");
        }

        let s3 = match crate::storage::S3Bucket::from_db_str(&row.bucket) {
            crate::storage::S3Bucket::Synckit => state.storage.synckit_s3.as_ref(),
            crate::storage::S3Bucket::Public => state.storage.public_s3.as_ref(),
            crate::storage::S3Bucket::Main => state.storage.s3.as_ref(),
        };

        if let Some(s3) = s3 {
            if row.s3_key.ends_with('/') {
                // Prefix delete (account-cascade cleanup). Bypasses the per-key
                // live-check by design, but carries a liveness guard symmetric
                // across buckets: a `{user_id}/` (main) or `{app_id}/` /
                // `ota/{app_id}/` (synckit) prefix wipes an entire creator's or
                // app's storage and is only ever enqueued by cleanup that deletes
                // the owning row in the same pass. A STILL-LIVE owner means
                // something wrongly enqueued a wipe, refuse and let the row climb
                // toward dead-letter triage rather than nuking live files. The
                // synckit branch was previously unguarded (ultra-fuzz Run 11 Storage LOW).
                let live_owner: Option<String> =
                    match crate::storage::S3Bucket::from_db_str(&row.bucket) {
                        // Both the main and public buckets key content under
                        // `{user_id}/`, so the owner guard is identical.
                        crate::storage::S3Bucket::Main | crate::storage::S3Bucket::Public => {
                            match row
                                .s3_key
                                .strip_suffix('/')
                                .and_then(|s| s.parse::<db::UserId>().ok())
                            {
                                Some(uid)
                                    if matches!(
                                        db::users::get_user_by_id(&state.db, uid).await,
                                        Ok(Some(_))
                                    ) =>
                                {
                                    Some(format!("user {uid}"))
                                }
                                _ => None,
                            }
                        }
                        crate::storage::S3Bucket::Synckit => {
                            let app_seg = row.s3_key.strip_prefix("ota/").unwrap_or(&row.s3_key);
                            match app_seg
                                .strip_suffix('/')
                                .and_then(|s| s.parse::<db::SyncAppId>().ok())
                            {
                                Some(app_id)
                                    if matches!(
                                        db::synckit::get_sync_app_by_id(&state.db, app_id).await,
                                        Ok(Some(_))
                                    ) =>
                                {
                                    Some(format!("sync app {app_id}"))
                                }
                                _ => None,
                            }
                        }
                    };
                if let Some(owner) = live_owner {
                    tracing::error!(s3_key = %row.s3_key, %owner,
                        "refusing prefix S3 delete: owner still exists, parking for dead-letter triage instead of wiping live storage");
                    continue;
                }
                match s3
                    .delete_prefix(&S3DeleteAuthority::new(), &row.s3_key)
                    .await
                {
                    Ok(()) => completed_ids.push(row.id),
                    Err(e) => {
                        tracing::warn!(s3_key = %row.s3_key, error = ?e, "retry S3 prefix deletion failed");
                    }
                }
            } else {
                // Single-key delete, routed through the one guarded funnel so the
                // delete-then-reupload live-check can never be skipped. The same
                // funnel backs the orphaned-upload reaper.
                match delete_orphan_key_guarded(&state.db, s3.as_ref(), &row.bucket, &row.s3_key)
                    .await
                {
                    GuardedDelete::Deleted | GuardedDelete::SkippedLive => {
                        completed_ids.push(row.id);
                    }
                    GuardedDelete::Failed => {} // leave queued; climbs toward dead-letter
                }
            }
        } else {
            // S3 not configured, remove from queue (can't delete what doesn't exist)
            completed_ids.push(row.id);
        }
    }

    // Move permanently-failing rows to the dead-letter table (durable, operator-
    // visible) rather than silently DELETEing them and orphaning the S3 object.
    if !dead_letter_ids.is_empty() {
        match db::pending_s3_deletions::move_to_dead_letter(&state.db, &dead_letter_ids).await {
            Ok(moved) => tracing::warn!(
                moved,
                "moved permanently-failing S3 deletions to dead-letter table, manual triage required"
            ),
            Err(e) => {
                tracing::error!(error = ?e, "failed to move S3 deletions to dead-letter table");
            }
        }
    }

    if !completed_ids.is_empty() {
        if let Err(e) = db::pending_s3_deletions::remove_completed(&state.db, &completed_ids).await
        {
            tracing::error!(error = ?e, "failed to dequeue completed S3 deletions");
        } else {
            tracing::info!(
                completed = completed_ids.len(),
                total = stale.len(),
                "retried pending S3 deletions"
            );
        }
    }

    if !completed_ids.is_empty() || !dead_letter_ids.is_empty() {
        let processed = (completed_ids.len() + dead_letter_ids.len()) as i64;
        let _ = db::scheduler_jobs::record_job_run(&state.db, "s3_deletion_retry", processed).await;
    }
}

/// Test-only synchronous drain of the pending-S3-deletion queue (`main` bucket).
///
/// Run [`cleanup_orphaned_uploads`] once, synchronously.
///
/// The reaper is `pub(super)` and the scheduler drives it on a tick, so no
/// integration test could reach it, including the branch that hands a failed S3
/// delete to the durable deletion queue instead of dropping the tracking row.
/// That branch is the fix for a leak that shipped once already (Run #2 Storage
/// SERIOUS), which is exactly the kind of code that should not be reachable only
/// in production. Exposed via `TestHarness::run_orphan_upload_reaper`.
#[doc(hidden)]
#[tracing::instrument(skip_all, name = "scheduler::cleanup_orphaned_uploads_for_test")]
pub async fn cleanup_orphaned_uploads_for_test(state: &AppState) {
    cleanup_orphaned_uploads(state).await;
}

/// The S3 delete a confirm/delete handler triggers is asynchronous: handlers
/// only [`enqueue_s3_orphan`](crate::routes::storage::enqueue_s3_orphan), and
/// the scheduler's [`retry_pending_s3_deletions`] performs the actual delete
/// later. Integration tests can't mint [`S3DeleteAuthority`] and don't run the
/// scheduler, so a test that asserts an object is gone must force the queued
/// deletes first. This runs them immediately against the given pool + storage,
/// mirroring the scheduler's per-row guarded-delete path. Returns the number of
/// objects actually deleted. Exposed via `TestHarness::drain_s3_deletions`.
#[doc(hidden)]
#[tracing::instrument(skip_all, name = "scheduler::drain_pending_s3_deletions_for_test")]
pub async fn drain_pending_s3_deletions_for_test(
    pool: &sqlx::PgPool,
    s3: &dyn crate::storage::StorageBackend,
) -> usize {
    let Ok(stale) =
        db::pending_s3_deletions::get_stale_pending(pool, chrono::Duration::zero(), 1000).await
    else {
        return 0;
    };
    let mut completed = Vec::new();
    let mut deleted = 0usize;
    for row in &stale {
        // Storage tests only exercise the main bucket; synckit needs its own s3.
        if crate::storage::S3Bucket::from_db_str(&row.bucket) != crate::storage::S3Bucket::Main {
            continue;
        }
        match delete_orphan_key_guarded(pool, s3, &row.bucket, &row.s3_key).await {
            GuardedDelete::Deleted => {
                deleted += 1;
                completed.push(row.id);
            }
            GuardedDelete::SkippedLive => completed.push(row.id),
            GuardedDelete::Failed => {}
        }
    }
    if !completed.is_empty() {
        let _ = db::pending_s3_deletions::remove_completed(pool, &completed).await;
    }
    deleted
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn cleanup_git_repos_removes_directory() {
        let tmp = tempfile::tempdir().unwrap();
        let git_root = tmp.path();
        let user_dir = git_root.join("testuser");
        std::fs::create_dir_all(user_dir.join("repo.git")).unwrap();
        std::fs::write(user_dir.join("repo.git/HEAD"), "ref: refs/heads/main\n").unwrap();

        let user_id = db::UserId::nil();
        cleanup_git_repos_on_disk(git_root.to_str().unwrap(), "testuser", user_id).await;

        assert!(!user_dir.exists(), "user git directory should be deleted");
    }

    #[tokio::test]
    async fn cleanup_git_repos_noop_if_missing() {
        let tmp = tempfile::tempdir().unwrap();
        let user_id = db::UserId::nil();
        cleanup_git_repos_on_disk(tmp.path().to_str().unwrap(), "nonexistent", user_id).await;
    }
}