PiCloud/crates/manager-core/tests/queue_release.rs

//! Audit fix #4: a TRANSIENT queue release (handler never ran — gate saturated
//! or script disabled at fire time) must NOT count against `max_attempts`.
//!
//! `claim` pre-increments `attempt` before the handler runs; a real failure
//! then legitimately counts. But the gate-saturation / disabled-at-fire paths
//! re-queue WITHOUT executing, so they must undo that pre-increment via
//! `release` — otherwise sustained overload could dead-letter a message that
//! executed zero times. This pins `release` (undoes the increment) against
//! `nack` (keeps it). Skips cleanly when `DATABASE_URL` is unset.

use picloud_manager_core::queue_repo::{PostgresQueueRepo, QueueRepo};
use picloud_shared::AppId;
use sqlx::postgres::PgPoolOptions;
use sqlx::PgPool;
use uuid::Uuid;

async fn pool_or_skip() -> Option<PgPool> {
    let Ok(url) = std::env::var("DATABASE_URL") else {
        eprintln!("queue_release: DATABASE_URL unset — skipping");
        return None;
    };
    let pool = PgPoolOptions::new()
        .max_connections(2)
        .connect(&url)
        .await
        .expect("connect");
    sqlx::migrate!("./migrations")
        .run(&pool)
        .await
        .expect("migrate");
    Some(pool)
}

async fn mk_app(pool: &PgPool, slug: &str) -> Uuid {
    // apps.group_id is NOT NULL — create a root group to hang the app on.
    let g: (Uuid,) = sqlx::query_as("INSERT INTO groups (slug, name) VALUES ($1, $1) RETURNING id")
        .bind(format!("{slug}-grp"))
        .fetch_one(pool)
        .await
        .expect("insert group");
    let r: (Uuid,) =
        sqlx::query_as("INSERT INTO apps (slug, name, group_id) VALUES ($1, $1, $2) RETURNING id")
            .bind(slug)
            .bind(g.0)
            .fetch_one(pool)
            .await
            .expect("insert app");
    r.0
}

async fn attempt_of(pool: &PgPool, id: Uuid) -> i32 {
    let r: (i32,) = sqlx::query_as("SELECT attempt FROM queue_messages WHERE id = $1")
        .bind(id)
        .fetch_one(pool)
        .await
        .expect("read attempt");
    r.0
}

fn unique(p: &str) -> String {
    // No Date/rand in scripts, but tests can use a UUID for uniqueness.
    format!("{p}-{}", Uuid::new_v4().simple())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn release_undoes_the_claim_increment_but_nack_keeps_it() {
    let Some(pool) = pool_or_skip().await else {
        return;
    };
    let repo = PostgresQueueRepo::new(pool.clone());
    let app = mk_app(&pool, &unique("qr-app")).await;
    let app_id = AppId::from(app);
    let queue = "jobs";

    // --- release: claim (attempt 0→1) then release → back to 0 ------------
    let id_rel: (Uuid,) = sqlx::query_as(
        "INSERT INTO queue_messages (app_id, queue_name, payload, max_attempts) \
         VALUES ($1, $2, '{}'::jsonb, 3) RETURNING id",
    )
    .bind(app)
    .bind(queue)
    .fetch_one(&pool)
    .await
    .expect("insert msg");
    let claimed = repo
        .claim(app_id, queue)
        .await
        .expect("claim")
        .expect("a message");
    assert_eq!(claimed.attempt, 1, "claim pre-increments attempt");
    assert!(repo
        .release(
            claimed.id,
            claimed.claim_token,
            chrono::Duration::milliseconds(0)
        )
        .await
        .expect("release"));
    assert_eq!(
        attempt_of(&pool, id_rel.0).await,
        0,
        "a transient release must undo the claim's pre-increment"
    );

    // --- nack: claim (0→1) then nack → stays 1 (a real failure counts) ----
    let id_nack: (Uuid,) = sqlx::query_as(
        "INSERT INTO queue_messages (app_id, queue_name, payload, max_attempts) \
         VALUES ($1, $2, '{}'::jsonb, 3) RETURNING id",
    )
    .bind(app)
    .bind("jobs2")
    .fetch_one(&pool)
    .await
    .expect("insert msg2");
    let claimed2 = repo
        .claim(app_id, "jobs2")
        .await
        .expect("claim2")
        .expect("a message");
    assert!(repo
        .nack(
            claimed2.id,
            claimed2.claim_token,
            chrono::Duration::milliseconds(0)
        )
        .await
        .expect("nack"));
    assert_eq!(
        attempt_of(&pool, id_nack.0).await,
        1,
        "a real-failure nack keeps the attempt (it counts toward max_attempts)"
    );

    // cleanup
    sqlx::query("DELETE FROM apps WHERE id = $1")
        .bind(app)
        .execute(&pool)
        .await
        .expect("cleanup");
}