tireless/crates/tireless-data/tests/migrations.rs

//! Migration tests against a real Postgres.
//!
//! Skipped unless `TIRELESS_TEST_DATABASE_URL` is set, because CI has no
//! database and the house pattern is not to give it one (see `store.rs` on why
//! queries are checked at runtime). Run them by hand against a throwaway server:
//!
//! ```sh
//! podman run -d --name pg -e POSTGRES_PASSWORD=test -e POSTGRES_DB=tireless_test \
//!     -p 55432:5432 docker.io/library/postgres:18-alpine
//! export TIRELESS_TEST_DATABASE_URL=postgres://postgres:test@127.0.0.1:55432/tireless_test
//! cargo test -p tireless-data -- --ignored
//! ```
//!
//! They are `#[ignore]`d rather than silently passing on a missing variable, so
//! a green `cargo test` never implies the schema was exercised.

use sqlx::{Connection, Executor, PgConnection, Row};
use tireless_data::store::MIGRATOR;

fn url() -> String {
    std::env::var("TIRELESS_TEST_DATABASE_URL")
        .expect("set TIRELESS_TEST_DATABASE_URL to run the ignored migration tests")
}

/// A connection scoped to its own empty schema.
///
/// Per test, not per run: `cargo test` runs these concurrently against one
/// server, so a shared `public` schema means each test's reset truncates
/// whatever its neighbours are midway through. The failure looks like a schema
/// bug — a unique violation on a row the test never inserted — which is an
/// expensive thing to misread.
async fn fresh_db(schema: &str) -> PgConnection {
    let mut conn = PgConnection::connect(&url()).await.expect("connect");
    conn.execute(
        format!(
            "drop schema if exists {schema} cascade; \
             create schema {schema}; \
             set search_path to {schema};"
        )
        .as_str(),
    )
    .await
    .expect("private schema");
    conn
}

/// A second connection into an existing test schema, for concurrency tests.
async fn join_db(schema: &str) -> PgConnection {
    let mut conn = PgConnection::connect(&url()).await.expect("connect");
    conn.execute(format!("set search_path to {schema};").as_str())
        .await
        .expect("search_path");
    conn
}

#[tokio::test]
#[ignore = "needs TIRELESS_TEST_DATABASE_URL"]
async fn migrations_apply_to_an_empty_database() {
    const SCHEMA: &str = "t_apply";
    let mut conn = fresh_db(SCHEMA).await;
    MIGRATOR.run(&mut conn).await.expect("first run");

    let tables: Vec<String> = sqlx::query(
        format!("select tablename from pg_tables where schemaname='{SCHEMA}'").as_str(),
    )
    .fetch_all(&mut conn)
    .await
    .expect("list tables")
    .iter()
    .map(|r| r.get::<String, _>("tablename"))
    .collect();

    for expected in ["tracked_repo", "job", "agent_run"] {
        assert!(
            tables.contains(&expected.to_string()),
            "missing table {expected}"
        );
    }
}

#[tokio::test]
#[ignore = "needs TIRELESS_TEST_DATABASE_URL"]
async fn applying_twice_is_a_no_op() {
    const SCHEMA: &str = "t_twice";
    // The runner, poller and api all migrate on start, and systemd may restart
    // any of them at any time — including all three at once after a deploy. A
    // second run must be silent, not an error and not a duplicate.
    let mut conn = fresh_db(SCHEMA).await;
    MIGRATOR.run(&mut conn).await.expect("first run");
    MIGRATOR
        .run(&mut conn)
        .await
        .expect("second run must be a no-op");

    let applied: i64 = sqlx::query("select count(*) as n from _sqlx_migrations")
        .fetch_one(&mut conn)
        .await
        .expect("count migrations")
        .get("n");
    assert_eq!(applied, MIGRATOR.iter().count() as i64);
}

#[tokio::test]
#[ignore = "needs TIRELESS_TEST_DATABASE_URL"]
async fn one_live_job_per_issue_but_any_number_of_historical_ones() {
    const SCHEMA: &str = "t_live";
    // The property the partial unique index exists for. Getting this wrong in
    // either direction is silent: too strict and discovery never recurs, too
    // loose and one issue is worked twice at once.
    let mut conn = fresh_db(SCHEMA).await;
    MIGRATOR.run(&mut conn).await.expect("migrate");

    let insert = |state: &str| {
        format!(
            "insert into job (forge,owner,repo,number,kind,state) \
             values ('gitea','lair','tireless',1,'discover','{state}')"
        )
    };

    conn.execute(insert("pending").as_str())
        .await
        .expect("first live job");
    let second = conn.execute(insert("pending").as_str()).await;
    assert!(
        second.is_err(),
        "a second live job for one issue must be rejected"
    );

    conn.execute("update job set state='delivered'")
        .await
        .expect("finish it");
    conn.execute(insert("pending").as_str())
        .await
        .expect("a new job after a terminal one must be allowed");

    let n: i64 = sqlx::query("select count(*) as n from job")
        .fetch_one(&mut conn)
        .await
        .expect("count")
        .get("n");
    assert_eq!(n, 2, "history must be retained, not overwritten");
}

#[tokio::test]
#[ignore = "needs TIRELESS_TEST_DATABASE_URL"]
async fn a_pending_job_cannot_hold_a_claim() {
    const SCHEMA: &str = "t_claim";
    // Lease expiry returns a job to the pool, and must clear the claim as it
    // does. The database refuses the half-done version so that a buggy release
    // path cannot strand an issue.
    let mut conn = fresh_db(SCHEMA).await;
    MIGRATOR.run(&mut conn).await.expect("migrate");

    let bad = conn
        .execute(
            "insert into job (forge,owner,repo,number,kind,state,claimed_by,claim_expires_at) \
             values ('gitea','lair','tireless',1,'plan','pending','w1',now())",
        )
        .await;
    assert!(
        bad.is_err(),
        "a pending job holding a claim must be rejected"
    );

    conn.execute(
        "insert into job (forge,owner,repo,number,kind,state,claimed_by,claim_expires_at) \
         values ('gitea','lair','tireless',1,'plan','claimed','w1',now() + interval '5 min')",
    )
    .await
    .expect("a properly claimed job is fine");
}

#[tokio::test]
#[ignore = "needs TIRELESS_TEST_DATABASE_URL"]
async fn concurrent_claims_never_hand_out_the_same_job() {
    const SCHEMA: &str = "t_concurrent";
    // The property the whole design rests on (design.md §4.2). Two claimers,
    // one pending job: exactly one winner, and the loser gets nothing rather
    // than blocking.
    let mut setup = fresh_db(SCHEMA).await;
    MIGRATOR.run(&mut setup).await.expect("migrate");
    setup
        .execute(
            "insert into job (forge,owner,repo,number,kind,state) \
             values ('gitea','lair','tireless',1,'plan','pending')",
        )
        .await
        .expect("seed one job");

    const CLAIM: &str = "select id from job where state='pending' \
                         order by created_at limit 1 for update skip locked";

    let mut a = join_db(SCHEMA).await;
    let mut b = join_db(SCHEMA).await;

    let mut ta = a.begin().await.expect("tx a");
    let got_a = sqlx::query(CLAIM)
        .fetch_all(&mut *ta)
        .await
        .expect("claim a");

    // b runs while a still holds the row lock.
    let mut tb = b.begin().await.expect("tx b");
    let got_b = sqlx::query(CLAIM)
        .fetch_all(&mut *tb)
        .await
        .expect("claim b");

    assert_eq!(got_a.len(), 1, "the first claimer takes the job");
    assert_eq!(
        got_b.len(),
        0,
        "the second claimer must skip the locked row, not block and not duplicate"
    );

    ta.commit().await.expect("commit a");
    tb.commit().await.expect("commit b");
}