| 1 | |
|
| 2 | |
|
| 3 | |
|
| 4 | | use super::{now_ms, Pool}; |
| 5 | | use sqlx::FromRow; |
| 6 | |
|
| 7 | |
|
| 8 | | #[derive(Clone, Debug, Default, PartialEq, Eq, FromRow)] |
| 9 | | pub struct Policy { |
| 10 | | pub scope: String, |
| 11 | | pub key: String, |
| 12 | | pub updates: Option<bool>, |
| 13 | | pub channel: Option<String>, |
| 14 | | pub window_from: Option<i64>, |
| 15 | | pub window_to: Option<i64>, |
| 16 | | pub dry_run: Option<bool>, |
| 17 | | } |
| 18 | |
|
| 19 | | impl Policy { |
| 20 | | |
| 21 | 10× | pub fn is_empty(&self) -> bool { |
| 22 | 10× | self.updates.is_none() && self.channel.is_none() && self.window_from.is_none() && self.dry_run.is_none() |
| 23 | 10× | } |
| 24 | | } |
| 25 | |
|
| 26 | | #[derive(Clone, Debug, PartialEq, Eq, FromRow)] |
| 27 | | pub struct Group { |
| 28 | | pub id: i64, |
| 29 | | |
| 30 | | pub owner: String, |
| 31 | | pub name: String, |
| 32 | | |
| 33 | | |
| 34 | | pub percent: Option<i64>, |
| 35 | | pub created_ms: i64, |
| 36 | | |
| 37 | | |
| 38 | | |
| 39 | | pub wave: Option<i64>, |
| 40 | | } |
| 41 | |
|
| 42 | 80× | pub async fn policies(pool: &Pool) -> sqlx::Result<Vec<Policy>> { |
| 43 | 80× | sqlx::query_as("SELECT * FROM runner_update_policies").fetch_all(pool).await |
| 44 | 80× | } |
| 45 | |
|
| 46 | 0× | pub async fn policy(pool: &Pool, scope: &str, key: &str) -> sqlx::Result<Policy> { |
| 47 | 0× | let p: Option<Policy> = sqlx::query_as("SELECT * FROM runner_update_policies WHERE scope = $1 AND key = $2") |
| 48 | 0× | .bind(scope) |
| 49 | 0× | .bind(key) |
| 50 | 0× | .fetch_optional(pool) |
| 51 | 0× | .await?; |
| 52 | 0× | Ok(p.unwrap_or(Policy { scope: scope.into(), key: key.into(), ..Policy::default() })) |
| 53 | 0× | } |
| 54 | |
|
| 55 | |
|
| 56 | 10× | pub async fn set_policy(pool: &Pool, p: &Policy) -> sqlx::Result<()> { |
| 57 | 10× | sqlx::query("DELETE FROM runner_update_policies WHERE scope = $1 AND key = $2").bind(&p.scope).bind(&p.key).execute(pool).await?; |
| 58 | 10× | if p.is_empty() { |
| 59 | 0× | return Ok(()); |
| 60 | 10× | } |
| 61 | 10× | sqlx::query( |
| 62 | 10× | "INSERT INTO runner_update_policies (scope, key, updates, channel, window_from, window_to, dry_run) |
| 63 | 10× | VALUES ($1, $2, $3, $4, $5, $6, $7)", |
| 64 | 10× | ) |
| 65 | 10× | .bind(&p.scope) |
| 66 | 10× | .bind(&p.key) |
| 67 | 10× | .bind(p.updates) |
| 68 | 10× | .bind(&p.channel) |
| 69 | 10× | .bind(p.window_from) |
| 70 | 10× | .bind(p.window_to) |
| 71 | 10× | .bind(p.dry_run) |
| 72 | 10× | .execute(pool) |
| 73 | 10× | .await?; |
| 74 | 10× | Ok(()) |
| 75 | 10× | } |
| 76 | |
|
| 77 | 87× | pub async fn groups(pool: &Pool) -> sqlx::Result<Vec<Group>> { |
| 78 | 87× | sqlx::query_as("SELECT * FROM runner_groups ORDER BY owner, name").fetch_all(pool).await |
| 79 | 87× | } |
| 80 | |
|
| 81 | 7× | pub async fn group(pool: &Pool, id: i64) -> sqlx::Result<Option<Group>> { |
| 82 | 7× | sqlx::query_as("SELECT * FROM runner_groups WHERE id = $1").bind(id).fetch_optional(pool).await |
| 83 | 7× | } |
| 84 | |
|
| 85 | 5× | pub async fn create_group(pool: &Pool, owner: &str, name: &str, percent: Option<i64>) -> sqlx::Result<Group> { |
| 86 | 5× | sqlx::query_as("INSERT INTO runner_groups (owner, name, percent, created_ms) VALUES ($1, $2, $3, $4) RETURNING *") |
| 87 | 5× | .bind(owner) |
| 88 | 5× | .bind(name) |
| 89 | 5× | .bind(percent) |
| 90 | 5× | .bind(now_ms()) |
| 91 | 5× | .fetch_one(pool) |
| 92 | 5× | .await |
| 93 | 5× | } |
| 94 | |
|
| 95 | 1× | pub async fn set_percent(pool: &Pool, id: i64, percent: Option<i64>) -> sqlx::Result<()> { |
| 96 | 1× | sqlx::query("UPDATE runner_groups SET percent = $1 WHERE id = $2").bind(percent).bind(id).execute(pool).await?; |
| 97 | 1× | Ok(()) |
| 98 | 1× | } |
| 99 | |
|
| 100 | 5× | pub async fn set_wave(pool: &Pool, id: i64, wave: Option<i64>) -> sqlx::Result<()> { |
| 101 | 5× | sqlx::query("UPDATE runner_groups SET wave = $1 WHERE id = $2").bind(wave).bind(id).execute(pool).await?; |
| 102 | 5× | Ok(()) |
| 103 | 5× | } |
| 104 | |
|
| 105 | |
|
| 106 | 1× | pub async fn delete_group(pool: &Pool, id: i64) -> sqlx::Result<()> { |
| 107 | 1× | sqlx::query("DELETE FROM runner_group_members WHERE group_id = $1").bind(id).execute(pool).await?; |
| 108 | 1× | sqlx::query("DELETE FROM runner_update_policies WHERE scope = 'group' AND key = $1").bind(id.to_string()).execute(pool).await?; |
| 109 | 1× | sqlx::query("DELETE FROM runner_groups WHERE id = $1").bind(id).execute(pool).await?; |
| 110 | 1× | Ok(()) |
| 111 | 1× | } |
| 112 | |
|
| 113 | |
|
| 114 | 80× | pub async fn members(pool: &Pool) -> sqlx::Result<Vec<(i64, i64)>> { |
| 115 | 80× | sqlx::query_as("SELECT group_id, runner_id FROM runner_group_members").fetch_all(pool).await |
| 116 | 80× | } |
| 117 | |
|
| 118 | |
|
| 119 | 5× | pub async fn set_members(pool: &Pool, group: i64, runners: &[i64]) -> sqlx::Result<()> { |
| 120 | 5× | sqlx::query("DELETE FROM runner_group_members WHERE group_id = $1").bind(group).execute(pool).await?; |
| 121 | 5× | for r in runners { |
| 122 | 4× | sqlx::query("INSERT INTO runner_group_members (group_id, runner_id) VALUES ($1, $2)").bind(group).bind(r).execute(pool).await?; |
| 123 | | } |
| 124 | 5× | Ok(()) |
| 125 | 5× | } |
| 126 | |
|
| 127 | |
|
| 128 | 2× | pub async fn report(pool: &Pool, runner: i64, note: &str) -> sqlx::Result<()> { |
| 129 | 2× | sqlx::query("UPDATE flow_runners SET update_note = $1, update_ms = $2 WHERE id = $3") |
| 130 | 2× | .bind(note) |
| 131 | 2× | .bind(now_ms()) |
| 132 | 2× | .bind(runner) |
| 133 | 2× | .execute(pool) |
| 134 | 2× | .await?; |
| 135 | 2× | Ok(()) |
| 136 | 2× | } |
| 137 | |
|
| 138 | |
|
| 139 | | #[derive(Clone, Debug, PartialEq, Eq, FromRow)] |
| 140 | | pub struct Hold { |
| 141 | | pub version: String, |
| 142 | | |
| 143 | | pub runner: String, |
| 144 | | pub why: String, |
| 145 | | pub at_ms: i64, |
| 146 | | } |
| 147 | |
|
| 148 | |
|
| 149 | 2× | pub async fn hold(pool: &Pool, version: &str, runner: &str, why: &str) -> sqlx::Result<()> { |
| 150 | 2× | sqlx::query("INSERT INTO runner_update_holds (version, runner, why, at_ms) VALUES ($1, $2, $3, $4) ON CONFLICT (version) DO NOTHING") |
| 151 | 2× | .bind(version) |
| 152 | 2× | .bind(runner) |
| 153 | 2× | .bind(why) |
| 154 | 2× | .bind(now_ms()) |
| 155 | 2× | .execute(pool) |
| 156 | 2× | .await?; |
| 157 | 2× | Ok(()) |
| 158 | 2× | } |
| 159 | |
|
| 160 | 30× | pub async fn holds(pool: &Pool) -> sqlx::Result<Vec<Hold>> { |
| 161 | 30× | sqlx::query_as("SELECT * FROM runner_update_holds ORDER BY at_ms DESC").fetch_all(pool).await |
| 162 | 30× | } |
| 163 | |
|
| 164 | 41× | pub async fn held(pool: &Pool, version: &str) -> sqlx::Result<Option<Hold>> { |
| 165 | 41× | sqlx::query_as("SELECT * FROM runner_update_holds WHERE version = $1").bind(version).fetch_optional(pool).await |
| 166 | 41× | } |
| 167 | |
|
| 168 | |
|
| 169 | 3× | pub async fn lift(pool: &Pool, version: &str) -> sqlx::Result<bool> { |
| 170 | 3× | Ok(sqlx::query("DELETE FROM runner_update_holds WHERE version = $1").bind(version).execute(pool).await?.rows_affected() > 0) |
| 171 | 3× | } |
| 172 | |
|
| 173 | |
|
| 174 | |
|
| 175 | |
|
| 176 | | #[derive(Clone, Debug, Default, PartialEq, Eq, FromRow)] |
| 177 | | pub struct Soak { |
| 178 | | pub runner_id: i64, |
| 179 | | pub jobs: i64, |
| 180 | | pub errors: i64, |
| 181 | | pub first_ms: Option<i64>, |
| 182 | | } |
| 183 | |
|
| 184 | |
|
| 185 | 9× | pub async fn soaks(pool: &Pool, version: &str) -> sqlx::Result<Vec<Soak>> { |
| 186 | 9× | sqlx::query_as( |
| 187 | 9× | "SELECT runner_id, COUNT(*) AS jobs, SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) AS errors, MIN(started_ms) AS first_ms |
| 188 | 9× | FROM flow_jobs WHERE runner_version = $1 AND runner_id IS NOT NULL AND status IN ('passed', 'failed', 'error') |
| 189 | 9× | GROUP BY runner_id", |
| 190 | 9× | ) |
| 191 | 9× | .bind(version) |
| 192 | 9× | .fetch_all(pool) |
| 193 | 9× | .await |
| 194 | 9× | } |
| 195 | |
|
| 196 | |
|
| 197 | |
|
| 198 | 2× | pub async fn first_error(pool: &Pool, version: &str, runner: i64) -> sqlx::Result<Option<String>> { |
| 199 | 2× | let row: Option<(String, String, String, i64, String)> = sqlx::query_as( |
| 200 | 2× | "SELECT j.name, p.owner, p.name, r.number, j.error FROM flow_jobs j |
| 201 | 2× | JOIN flow_runs r ON r.id = j.run_id JOIN repositories p ON p.id = r.repo_id |
| 202 | 2× | WHERE j.runner_version = $1 AND j.runner_id = $2 AND j.status = 'error' |
| 203 | 2× | ORDER BY j.finished_ms LIMIT 1", |
| 204 | 2× | ) |
| 205 | 2× | .bind(version) |
| 206 | 2× | .bind(runner) |
| 207 | 2× | .fetch_optional(pool) |
| 208 | 2× | .await?; |
| 209 | 2× | Ok(row.map(|(job, owner, name, n, error)| { |
| 210 | 2× | let why: String = error.chars().take(120).collect(); |
| 211 | 2× | format!("job {job} of {owner}/{name} run #{n}{}", if why.is_empty() { String::new() } else { format!(": {why}") }) |
| 212 | 2× | })) |
| 213 | 2× | } |