Coverage / crates/hub/src/db/runner_updates.rs 93.1%

    pub fn is_empty(&self) -> bool {

        self.updates.is_none() && self.channel.is_none() && self.window_from.is_none() && self.dry_run.is_none()

    }

pub async fn policies(pool: &Pool) -> sqlx::Result<Vec<Policy>> {

    sqlx::query_as("SELECT * FROM runner_update_policies").fetch_all(pool).await

}

pub async fn set_policy(pool: &Pool, p: &Policy) -> sqlx::Result<()> {

    sqlx::query("DELETE FROM runner_update_policies WHERE scope = $1 AND key = $2").bind(&p.scope).bind(&p.key).execute(pool).await?;

    if p.is_empty() {

    }

    sqlx::query(

        "INSERT INTO runner_update_policies (scope, key, updates, channel, window_from, window_to, dry_run)

         VALUES ($1, $2, $3, $4, $5, $6, $7)",

    )

    .bind(&p.scope)

    .bind(&p.key)

    .bind(p.updates)

    .bind(&p.channel)

    .bind(p.window_from)

    .bind(p.window_to)

    .bind(p.dry_run)

    .execute(pool)

    .await?;

    Ok(())

}

pub async fn groups(pool: &Pool) -> sqlx::Result<Vec<Group>> {

    sqlx::query_as("SELECT * FROM runner_groups ORDER BY owner, name").fetch_all(pool).await

}

pub async fn group(pool: &Pool, id: i64) -> sqlx::Result<Option<Group>> {

    sqlx::query_as("SELECT * FROM runner_groups WHERE id = $1").bind(id).fetch_optional(pool).await

}

pub async fn create_group(pool: &Pool, owner: &str, name: &str, percent: Option<i64>) -> sqlx::Result<Group> {

    sqlx::query_as("INSERT INTO runner_groups (owner, name, percent, created_ms) VALUES ($1, $2, $3, $4) RETURNING *")

        .bind(owner)

        .bind(name)

        .bind(percent)

        .bind(now_ms())

        .fetch_one(pool)

        .await

}

pub async fn set_percent(pool: &Pool, id: i64, percent: Option<i64>) -> sqlx::Result<()> {

    sqlx::query("UPDATE runner_groups SET percent = $1 WHERE id = $2").bind(percent).bind(id).execute(pool).await?;

    Ok(())

}

pub async fn set_wave(pool: &Pool, id: i64, wave: Option<i64>) -> sqlx::Result<()> {

    sqlx::query("UPDATE runner_groups SET wave = $1 WHERE id = $2").bind(wave).bind(id).execute(pool).await?;

    Ok(())

}

pub async fn delete_group(pool: &Pool, id: i64) -> sqlx::Result<()> {

    sqlx::query("DELETE FROM runner_group_members WHERE group_id = $1").bind(id).execute(pool).await?;

    sqlx::query("DELETE FROM runner_update_policies WHERE scope = 'group' AND key = $1").bind(id.to_string()).execute(pool).await?;

    sqlx::query("DELETE FROM runner_groups WHERE id = $1").bind(id).execute(pool).await?;

    Ok(())

}

pub async fn members(pool: &Pool) -> sqlx::Result<Vec<(i64, i64)>> {

    sqlx::query_as("SELECT group_id, runner_id FROM runner_group_members").fetch_all(pool).await

}

pub async fn set_members(pool: &Pool, group: i64, runners: &[i64]) -> sqlx::Result<()> {

    sqlx::query("DELETE FROM runner_group_members WHERE group_id = $1").bind(group).execute(pool).await?;

    for r in runners {

        sqlx::query("INSERT INTO runner_group_members (group_id, runner_id) VALUES ($1, $2)").bind(group).bind(r).execute(pool).await?;

    Ok(())

}

pub async fn report(pool: &Pool, runner: i64, note: &str) -> sqlx::Result<()> {

    sqlx::query("UPDATE flow_runners SET update_note = $1, update_ms = $2 WHERE id = $3")

        .bind(note)

        .bind(now_ms())

        .bind(runner)

        .execute(pool)

        .await?;

    Ok(())

}

pub async fn hold(pool: &Pool, version: &str, runner: &str, why: &str) -> sqlx::Result<()> {

    sqlx::query("INSERT INTO runner_update_holds (version, runner, why, at_ms) VALUES ($1, $2, $3, $4) ON CONFLICT (version) DO NOTHING")

        .bind(version)

        .bind(runner)

        .bind(why)

        .bind(now_ms())

        .execute(pool)

        .await?;

    Ok(())

}

pub async fn holds(pool: &Pool) -> sqlx::Result<Vec<Hold>> {

    sqlx::query_as("SELECT * FROM runner_update_holds ORDER BY at_ms DESC").fetch_all(pool).await

}

pub async fn held(pool: &Pool, version: &str) -> sqlx::Result<Option<Hold>> {

    sqlx::query_as("SELECT * FROM runner_update_holds WHERE version = $1").bind(version).fetch_optional(pool).await

}

pub async fn lift(pool: &Pool, version: &str) -> sqlx::Result<bool> {

    Ok(sqlx::query("DELETE FROM runner_update_holds WHERE version = $1").bind(version).execute(pool).await?.rows_affected() > 0)

}

pub async fn soaks(pool: &Pool, version: &str) -> sqlx::Result<Vec<Soak>> {

    sqlx::query_as(

        "SELECT runner_id, COUNT(*) AS jobs, SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) AS errors, MIN(started_ms) AS first_ms

         FROM flow_jobs WHERE runner_version = $1 AND runner_id IS NOT NULL AND status IN ('passed', 'failed', 'error')

         GROUP BY runner_id",

    )

    .bind(version)

    .fetch_all(pool)

    .await

}

pub async fn first_error(pool: &Pool, version: &str, runner: i64) -> sqlx::Result<Option<String>> {

    let row: Option<(String, String, String, i64, String)> = sqlx::query_as(

        "SELECT j.name, p.owner, p.name, r.number, j.error FROM flow_jobs j

         JOIN flow_runs r ON r.id = j.run_id JOIN repositories p ON p.id = r.repo_id

         WHERE j.runner_version = $1 AND j.runner_id = $2 AND j.status = 'error'

         ORDER BY j.finished_ms LIMIT 1",

    )

    .bind(version)

    .bind(runner)

    .fetch_optional(pool)

    .await?;

    Ok(row.map(|(job, owner, name, n, error)| {

        let why: String = error.chars().take(120).collect();

        format!("job {job} of {owner}/{name} run #{n}{}", if why.is_empty() { String::new() } else { format!(": {why}") })

    }))

}