neiam /dms

//! Regression cover for the re-enqueue path.
//!
//! `upsert_pending` always conflicts onto the existing (upload_id, kind) row
//! when a job is retried, so the ActiveModel's `attempts: Set(0)` never
//! applies -- only the columns named in DO UPDATE SET do. Omitting Attempts
//! there stranded any job that had reached WORKER_MAX_ATTEMPTS: retry flipped
//! status to pending, but the worker's pre-flight attempts check still saw
//! the exhausted count and bounced it straight to the DLQ.

use dms::jobs::upsert_pending;
use dms::queue::JobKind;
use sea_orm::{DatabaseBackend, MockDatabase, Value};
use std::collections::BTreeMap;
use uuid::Uuid;

async fn upsert_sql(kind: JobKind) -> String {
    // On Postgres, sea-orm 2 runs the insert as a query and reads the key
    // back from its RETURNING row, so the mock answers with that row.
    let db = MockDatabase::new(DatabaseBackend::Postgres)
        .append_query_results([[BTreeMap::from([("id", Value::from(Uuid::now_v7()))])]])
        .into_connection();
    upsert_pending(&db, Uuid::now_v7(), kind).await.expect("upsert_pending");
    let log = db.into_transaction_log();
    assert_eq!(log.len(), 1, "expected exactly one statement");
    format!("{:?}", log[0])
}

/// The whole point of the fix: a retry must clear the attempts budget.
#[tokio::test]
async fn upsert_pending_resets_attempts_on_conflict() {
    let sql = upsert_sql(JobKind::Describe).await;
    let update = sql
        .split_once("DO UPDATE SET")
        .expect("statement should be an upsert with a DO UPDATE branch")
        .1;
    assert!(
        update.contains("attempts"),
        "attempts must be reset in the conflict branch, or an exhausted job \
         can never be retried out of the DLQ; got: {sql}"
    );
}

/// Guard the columns the retry path also depends on, so a future edit to the
/// update list can't quietly drop one.
#[tokio::test]
async fn upsert_pending_resets_the_rest_of_the_retry_state() {
    let sql = upsert_sql(JobKind::Ocr).await;
    let update = sql.split_once("DO UPDATE SET").unwrap().1;
    for col in ["status", "error", "started_at", "completed_at", "updated_at"] {
        assert!(update.contains(col), "{col} missing from DO UPDATE SET; got: {sql}");
    }
}