leadcast/backend/tests/maintenance_reset.rs

use backend::db::{Db, querys::job, querys::maintenance};
use backend::run_control::RunControl;
use deadpool_postgres::{Config as PoolSettings, Runtime};
use tokio_postgres::NoTls;
use uuid::Uuid;

async fn test_db() -> Db {
    let database_url = std::env::var("DATABASE_URL")
        .unwrap_or_else(|_| "postgres://leads:leads@127.0.0.1:5432/leads_extractor".to_string());
    let mut settings = PoolSettings::new();
    settings.url = Some(database_url);
    let pool = settings
        .create_pool(Some(Runtime::Tokio1), NoTls)
        .expect("pool");
    Db::new(pool)
}

/// Reproduz o bug: um job travado em `running` com uma tentativa `running`
/// correspondente deve, após `reset_interrupted_work`, voltar a ficar
/// reivindicável sem colidir com `job_attempts_job_attempt_unique`.
#[tokio::test]
async fn reset_interrupted_work_frees_stuck_job_for_reclaim() {
    let db = test_db().await;
    let client = db.client().await.expect("client");

    let test_kind = format!("test_maintenance_reset_{}", Uuid::new_v4());
    let job_id: Uuid = client
        .query_one(
            "INSERT INTO jobs (kind, status, attempt_count, max_attempts, locked_at, locked_by, heartbeat_at, started_at)
             VALUES ($1, 'running', 1, 6, now(), 'worker-a', now(), now())
             RETURNING id",
            &[&test_kind],
        )
        .await
        .expect("insert job")
        .get(0);

    client
        .execute(
            "INSERT INTO job_attempts (job_id, attempt_no, worker_id, status)
             VALUES ($1, 1, 'worker-a', 'running')",
            &[&job_id],
        )
        .await
        .expect("insert job_attempt");

    drop(client);

    let runs = RunControl::default();
    maintenance::reset_interrupted_work(&db, &runs, "test-request")
        .await
        .expect("reset_interrupted_work");

    // O job deve estar de volta na fila, com o attempt_count preservado
    // (não zerado) para não colidir com o histórico de tentativas.
    let client = db.client().await.expect("client");
    let row = client
        .query_one(
            "SELECT status, attempt_count FROM jobs WHERE id = $1",
            &[&job_id],
        )
        .await
        .expect("select job");
    let status: String = row.get(0);
    let attempt_count: i32 = row.get(1);
    assert_eq!(status, "queued");
    assert_eq!(attempt_count, 1);

    let attempt_status: String = client
        .query_one(
            "SELECT status FROM job_attempts WHERE job_id = $1 AND attempt_no = 1",
            &[&job_id],
        )
        .await
        .expect("select job_attempt")
        .get(0);
    assert_eq!(attempt_status, "failed");
    drop(client);

    // Reivindicar o job não deve mais falhar por chave duplicada, e o novo
    // attempt_no deve ser 2, não 1.
    let claimed = job::claim_next(&db, "worker-b", Some(&test_kind))
        .await
        .expect("claim_next should not error")
        .expect("job should be claimable");
    assert_eq!(claimed.job.id, job_id);
    assert_eq!(claimed.attempt_no, 2);

    let client = db.client().await.expect("client");
    client
        .execute("DELETE FROM jobs WHERE id = $1", &[&job_id])
        .await
        .expect("cleanup");
}

/// Um run parado em `cancelling` no momento do reinício não pode ficar
/// travado nesse estado para sempre: `reset_interrupted_work` deve finalizá-lo
/// como `cancelled` e cancelar (não reenfileirar) os jobs pendentes/em
/// execução amarrados a ele, já que `claim_next` só reivindica jobs de runs
/// `pending`/`running`.
#[tokio::test]
async fn reset_interrupted_work_finalizes_cancelling_runs() {
    let db = test_db().await;
    let client = db.client().await.expect("client");

    let run_id: Uuid = client
        .query_one(
            "INSERT INTO pipeline_runs (kind, status, cancel_requested_at, started_at)
             VALUES ('interviewee_extraction', 'cancelling', now(), now())
             RETURNING id",
            &[],
        )
        .await
        .expect("insert run")
        .get(0);

    let running_job_id: Uuid = client
        .query_one(
            "INSERT INTO jobs (run_id, kind, status, attempt_count, max_attempts, locked_at, locked_by, heartbeat_at, started_at)
             VALUES ($1, 'interviewee_extraction', 'running', 1, 6, now(), 'worker-a', now(), now())
             RETURNING id",
            &[&run_id],
        )
        .await
        .expect("insert running job")
        .get(0);

    let queued_job_id: Uuid = client
        .query_one(
            "INSERT INTO jobs (run_id, kind, status)
             VALUES ($1, 'interviewee_extraction', 'queued')
             RETURNING id",
            &[&run_id],
        )
        .await
        .expect("insert queued job")
        .get(0);

    drop(client);

    let runs = RunControl::default();
    maintenance::reset_interrupted_work(&db, &runs, "test-request")
        .await
        .expect("reset_interrupted_work");

    let client = db.client().await.expect("client");
    let run_row = client
        .query_one(
            "SELECT status, finished_at FROM pipeline_runs WHERE id = $1",
            &[&run_id],
        )
        .await
        .expect("select run");
    let run_status: String = run_row.get(0);
    let finished_at: Option<chrono::DateTime<chrono::Utc>> = run_row.get(1);
    assert_eq!(run_status, "cancelled");
    assert!(finished_at.is_some());

    for job_id in [running_job_id, queued_job_id] {
        let job_status: String = client
            .query_one("SELECT status FROM jobs WHERE id = $1", &[&job_id])
            .await
            .expect("select job")
            .get(0);
        assert_eq!(job_status, "cancelled", "job {job_id} should be cancelled");
    }

    client
        .execute("DELETE FROM pipeline_runs WHERE id = $1", &[&run_id])
        .await
        .expect("cleanup");
}