| 1 | //! Version updates' tables in D1 (migration 0004): when each entry runs, |
| 2 | //! the update pull requests g1t makes, and ignore conditions from comments. |
| 3 | |
| 4 | use g1t_contracts::new_id; |
| 5 | use g1t_contracts::security::{BumpArgs, UpdateState}; |
| 6 | use g1t_contracts::updates::{IgnoreCondition, UpdatePull, UpdatedDependency}; |
| 7 | use g1t_kit::now_ms; |
| 8 | use serde::Deserialize; |
| 9 | use worker::Result; |
| 10 | use worker::wasm_bindgen::JsValue; |
| 11 | |
| 12 | use crate::store::{Store, now, optional}; |
| 13 | |
| 14 | /// Where one `updates` entry's checks stand. |
| 15 | #[derive(Clone, Debug, Deserialize)] |
| 16 | pub struct RunRow { |
| 17 | pub repo_id: String, |
| 18 | pub entry: String, |
| 19 | pub next_run_at: Option<String>, |
| 20 | pub last_checked_at: Option<String>, |
| 21 | pub last_result: Option<String>, |
| 22 | pub last_error: Option<String>, |
| 23 | } |
| 24 | |
| 25 | /// An update pull request, as stored. |
| 26 | #[derive(Clone, Debug, Deserialize)] |
| 27 | pub struct PullRow { |
| 28 | pub id: String, |
| 29 | pub repo_id: String, |
| 30 | pub kind: String, |
| 31 | pub entry: String, |
| 32 | pub ecosystem: String, |
| 33 | pub subject: String, |
| 34 | pub signature: String, |
| 35 | pub group_name: Option<String>, |
| 36 | pub branch: String, |
| 37 | pub title: String, |
| 38 | pub body: String, |
| 39 | pub dependencies: String, |
| 40 | pub bump: String, |
| 41 | pub assignees: String, |
| 42 | pub reviewers: String, |
| 43 | pub state: String, |
| 44 | pub pull: Option<i64>, |
| 45 | /// The work item g1t opened for it, from before it stopped doing that. |
| 46 | #[serde(default)] |
| 47 | pub item: Option<i64>, |
| 48 | #[serde(default)] |
| 49 | pub item_id: Option<String>, |
| 50 | #[serde(default)] |
| 51 | pub item_workspace: Option<String>, |
| 52 | pub head: Option<String>, |
| 53 | pub merge_by: Option<String>, |
| 54 | pub error: Option<String>, |
| 55 | pub updated_at: String, |
| 56 | } |
| 57 | |
| 58 | impl PullRow { |
| 59 | pub fn state(&self) -> UpdateState { |
| 60 | UpdateState::parse(&self.state).unwrap_or(UpdateState::Failed) |
| 61 | } |
| 62 | |
| 63 | pub fn pull(&self) -> Option<u32> { |
| 64 | self.pull.map(|n| n.max(0) as u32) |
| 65 | } |
| 66 | |
| 67 | pub fn linked_item(&self) -> Option<crate::store::LinkedItem> { |
| 68 | crate::store::LinkedItem::new(self.item, self.item_id.as_deref(), self.item_workspace.as_deref()) |
| 69 | } |
| 70 | |
| 71 | pub fn dependencies(&self) -> Vec<UpdatedDependency> { |
| 72 | serde_json::from_str(&self.dependencies).unwrap_or_default() |
| 73 | } |
| 74 | |
| 75 | pub fn bump(&self) -> Option<BumpArgs> { |
| 76 | serde_json::from_str(&self.bump).ok() |
| 77 | } |
| 78 | |
| 79 | pub fn names(&self, column: &str) -> Vec<String> { |
| 80 | serde_json::from_str(if column == "assignees" { &self.assignees } else { &self.reviewers }).unwrap_or_default() |
| 81 | } |
| 82 | |
| 83 | /// Who asked for it to merge, by username. |
| 84 | pub fn merge_requested_by(&self) -> Option<String> { |
| 85 | self.merge_by.clone() |
| 86 | } |
| 87 | |
| 88 | pub fn to_contract(&self) -> UpdatePull { |
| 89 | UpdatePull { |
| 90 | kind: self.kind.clone(), |
| 91 | entry: self.entry.clone(), |
| 92 | ecosystem: self.ecosystem.clone(), |
| 93 | group: self.group_name.clone(), |
| 94 | branch: self.branch.clone(), |
| 95 | title: self.title.clone(), |
| 96 | state: self.state.clone(), |
| 97 | pull: self.pull(), |
| 98 | dependencies: self.dependencies(), |
| 99 | merge_requested_by: self.merge_requested_by(), |
| 100 | error: self.error.clone(), |
| 101 | updated_at: self.updated_at.clone(), |
| 102 | } |
| 103 | } |
| 104 | } |
| 105 | |
| 106 | /// A pull request to record before its sandbox starts. |
| 107 | pub struct NewPull<'a> { |
| 108 | pub repo_id: &'a str, |
| 109 | pub kind: &'a str, |
| 110 | pub entry: &'a str, |
| 111 | pub ecosystem: &'a str, |
| 112 | pub subject: &'a str, |
| 113 | pub signature: &'a str, |
| 114 | pub group: Option<&'a str>, |
| 115 | pub branch: &'a str, |
| 116 | pub title: &'a str, |
| 117 | pub body: &'a str, |
| 118 | pub dependencies: &'a [UpdatedDependency], |
| 119 | /// Without registries: credentials are never stored. |
| 120 | pub bump: &'a BumpArgs, |
| 121 | pub assignees: &'a [String], |
| 122 | pub reviewers: &'a [String], |
| 123 | } |
| 124 | |
| 125 | #[derive(Deserialize)] |
| 126 | struct IgnoreRow { |
| 127 | ecosystem: String, |
| 128 | dependency: String, |
| 129 | versions: Option<String>, |
| 130 | update_type: Option<String>, |
| 131 | by: String, |
| 132 | pull: Option<i64>, |
| 133 | at: String, |
| 134 | } |
| 135 | |
| 136 | #[derive(Deserialize)] |
| 137 | struct Claimed { |
| 138 | #[allow(dead_code)] |
| 139 | entry: String, |
| 140 | } |
| 141 | |
| 142 | /// What clears a repository's version update rows when it is purged. |
| 143 | pub const PURGED: &[&str] = &[ |
| 144 | "DELETE FROM update_runs WHERE repo_id = ?1", |
| 145 | "DELETE FROM update_pulls WHERE repo_id = ?1", |
| 146 | "DELETE FROM update_ignores WHERE repo_id = ?1", |
| 147 | ]; |
| 148 | |
| 149 | /// A run claimed this long ago and never finished is taken to have died. |
| 150 | const RUN_CLAIM_MS: u64 = 30 * 60 * 1000; |
| 151 | |
| 152 | impl Store { |
| 153 | pub async fn runs(&self, repo_id: &str) -> Result<Vec<RunRow>> { |
| 154 | self.db.prepare("SELECT * FROM update_runs WHERE repo_id = ?").bind(&[repo_id.into()])?.all().await?.results::<RunRow>() |
| 155 | } |
| 156 | |
| 157 | /// Keeps a row for each of `entries` (id, next run) and removes the rest. |
| 158 | pub async fn set_runs(&self, repo_id: &str, entries: &[(String, Option<String>)]) -> Result<()> { |
| 159 | let mut statements = Vec::new(); |
| 160 | for (entry, next) in entries { |
| 161 | statements.push( |
| 162 | self.db |
| 163 | .prepare( |
| 164 | "INSERT INTO update_runs (repo_id, entry, next_run_at) VALUES (?1, ?2, ?3) |
| 165 | ON CONFLICT (repo_id, entry) DO UPDATE SET next_run_at = ?3", |
| 166 | ) |
| 167 | .bind(&[repo_id.into(), entry.as_str().into(), optional(next.as_deref())])?, |
| 168 | ); |
| 169 | } |
| 170 | let keep: Vec<JsValue> = entries.iter().map(|(entry, _)| JsValue::from(entry.as_str())).collect(); |
| 171 | let marks = vec!["?"; keep.len()].join(", "); |
| 172 | let mut binds = vec![JsValue::from(repo_id)]; |
| 173 | binds.extend(keep); |
| 174 | let sql = if entries.is_empty() { |
| 175 | "DELETE FROM update_runs WHERE repo_id = ?".to_owned() |
| 176 | } else { |
| 177 | format!("DELETE FROM update_runs WHERE repo_id = ? AND entry NOT IN ({marks})") |
| 178 | }; |
| 179 | statements.push(self.db.prepare(sql).bind(&binds)?); |
| 180 | self.db.batch(statements).await?; |
| 181 | Ok(()) |
| 182 | } |
| 183 | |
| 184 | /// Entries whose next run has come, oldest first, not being run now. |
| 185 | pub async fn due_runs(&self, at: &str, limit: u32) -> Result<Vec<RunRow>> { |
| 186 | let stale = g1t_contracts::time::rfc3339(now_ms().saturating_sub(RUN_CLAIM_MS)); |
| 187 | self.db |
| 188 | .prepare( |
| 189 | "SELECT * FROM update_runs WHERE next_run_at IS NOT NULL AND next_run_at <= ? |
| 190 | AND (running_at IS NULL OR running_at < ?) ORDER BY next_run_at LIMIT ?", |
| 191 | ) |
| 192 | .bind(&[at.into(), stale.into(), limit.into()])? |
| 193 | .all() |
| 194 | .await? |
| 195 | .results::<RunRow>() |
| 196 | } |
| 197 | |
| 198 | /// Takes an entry's run, unless another sweep already has. |
| 199 | pub async fn claim_run(&self, repo_id: &str, entry: &str) -> Result<bool> { |
| 200 | let stale = g1t_contracts::time::rfc3339(now_ms().saturating_sub(RUN_CLAIM_MS)); |
| 201 | let claimed = self |
| 202 | .db |
| 203 | .prepare( |
| 204 | "INSERT INTO update_runs (repo_id, entry, running_at) VALUES (?1, ?2, ?3) |
| 205 | ON CONFLICT (repo_id, entry) DO UPDATE SET running_at = ?3 |
| 206 | WHERE update_runs.running_at IS NULL OR update_runs.running_at < ?4 |
| 207 | RETURNING entry", |
| 208 | ) |
| 209 | .bind(&[repo_id.into(), entry.into(), now().into(), stale.into()])? |
| 210 | .first::<Claimed>(None) |
| 211 | .await?; |
| 212 | Ok(claimed.is_some()) |
| 213 | } |
| 214 | |
| 215 | pub async fn finish_run(&self, repo_id: &str, entry: &str, next: Option<&str>, result: Option<&str>, error: Option<&str>) -> Result<()> { |
| 216 | self.db |
| 217 | .prepare( |
| 218 | "UPDATE update_runs SET running_at = NULL, last_checked_at = ?, next_run_at = ?, last_result = ?, last_error = ? |
| 219 | WHERE repo_id = ? AND entry = ?", |
| 220 | ) |
| 221 | .bind(&[now().into(), optional(next), optional(result), optional(error), repo_id.into(), entry.into()])? |
| 222 | .run() |
| 223 | .await?; |
| 224 | Ok(()) |
| 225 | } |
| 226 | |
| 227 | // --- Update pull requests ------------------------------------------------- |
| 228 | |
| 229 | pub async fn update_pulls(&self, repo_id: &str) -> Result<Vec<PullRow>> { |
| 230 | self.db |
| 231 | .prepare("SELECT * FROM update_pulls WHERE repo_id = ? ORDER BY requested_at DESC LIMIT 500") |
| 232 | .bind(&[repo_id.into()])? |
| 233 | .all() |
| 234 | .await? |
| 235 | .results::<PullRow>() |
| 236 | } |
| 237 | |
| 238 | /// The latest update pull request made on `branch`. |
| 239 | pub async fn update_pull_by_branch(&self, repo_id: &str, branch: &str) -> Result<Option<PullRow>> { |
| 240 | self.db |
| 241 | .prepare("SELECT * FROM update_pulls WHERE repo_id = ? AND branch = ? ORDER BY requested_at DESC LIMIT 1") |
| 242 | .bind(&[repo_id.into(), branch.into()])? |
| 243 | .first::<PullRow>(None) |
| 244 | .await |
| 245 | } |
| 246 | |
| 247 | pub async fn update_pull_by_number(&self, repo_id: &str, number: u32) -> Result<Option<PullRow>> { |
| 248 | self.db |
| 249 | .prepare("SELECT * FROM update_pulls WHERE repo_id = ? AND pull = ? ORDER BY requested_at DESC LIMIT 1") |
| 250 | .bind(&[repo_id.into(), number.into()])? |
| 251 | .first::<PullRow>(None) |
| 252 | .await |
| 253 | } |
| 254 | |
| 255 | /// Open update pull requests, the least recently looked at first. |
| 256 | pub async fn open_update_pulls(&self, limit: u32) -> Result<Vec<PullRow>> { |
| 257 | self.db |
| 258 | .prepare("SELECT * FROM update_pulls WHERE state = 'open' ORDER BY updated_at LIMIT ?") |
| 259 | .bind(&[limit.into()])? |
| 260 | .all() |
| 261 | .await? |
| 262 | .results::<PullRow>() |
| 263 | } |
| 264 | |
| 265 | /// Update pull requests asked of a sandbox before `before` that never pushed. |
| 266 | /// Version updates left in `needs_code`, which g1t opened a work item for |
| 267 | /// before it stopped doing that. |
| 268 | pub async fn needs_code_update_pulls(&self, limit: u32) -> Result<Vec<PullRow>> { |
| 269 | self.db |
| 270 | .prepare("SELECT * FROM update_pulls WHERE state = 'needs_code' ORDER BY updated_at LIMIT ?") |
| 271 | .bind(&[limit.into()])? |
| 272 | .all() |
| 273 | .await? |
| 274 | .results::<PullRow>() |
| 275 | } |
| 276 | |
| 277 | pub async fn stalled_update_pulls(&self, before: &str, limit: u32) -> Result<Vec<PullRow>> { |
| 278 | self.db |
| 279 | .prepare("SELECT * FROM update_pulls WHERE state = 'requested' AND updated_at < ? ORDER BY updated_at LIMIT ?") |
| 280 | .bind(&[before.into(), limit.into()])? |
| 281 | .all() |
| 282 | .await? |
| 283 | .results::<PullRow>() |
| 284 | } |
| 285 | |
| 286 | pub async fn add_update_pull(&self, new: &NewPull<'_>) -> Result<String> { |
| 287 | let id = new_id("upd", now_ms()); |
| 288 | let at = now(); |
| 289 | self.db |
| 290 | .prepare( |
| 291 | "INSERT INTO update_pulls (id, repo_id, kind, entry, ecosystem, subject, signature, group_name, branch, title, |
| 292 | body, dependencies, bump, assignees, reviewers, state, requested_at, updated_at) |
| 293 | VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, 'requested', ?16, ?16)", |
| 294 | ) |
| 295 | .bind(&[ |
| 296 | id.as_str().into(), |
| 297 | new.repo_id.into(), |
| 298 | new.kind.into(), |
| 299 | new.entry.into(), |
| 300 | new.ecosystem.into(), |
| 301 | new.subject.into(), |
| 302 | new.signature.into(), |
| 303 | optional(new.group), |
| 304 | new.branch.into(), |
| 305 | new.title.into(), |
| 306 | new.body.into(), |
| 307 | serde_json::to_string(new.dependencies)?.into(), |
| 308 | serde_json::to_string(new.bump)?.into(), |
| 309 | serde_json::to_string(new.assignees)?.into(), |
| 310 | serde_json::to_string(new.reviewers)?.into(), |
| 311 | at.as_str().into(), |
| 312 | ])? |
| 313 | .run() |
| 314 | .await?; |
| 315 | Ok(id) |
| 316 | } |
| 317 | |
| 318 | pub async fn set_update_pull(&self, id: &str, state: UpdateState, pull: Option<u32>, item: Option<u32>, error: Option<&str>) -> Result<()> { |
| 319 | self.db |
| 320 | .prepare("UPDATE update_pulls SET state = ?, pull = ?, item = ?, error = ?, updated_at = ? WHERE id = ?") |
| 321 | .bind(&[ |
| 322 | state.as_str().into(), |
| 323 | pull.map_or(JsValue::NULL, JsValue::from), |
| 324 | item.map_or(JsValue::NULL, JsValue::from), |
| 325 | optional(error), |
| 326 | now().into(), |
| 327 | id.into(), |
| 328 | ])? |
| 329 | .run() |
| 330 | .await?; |
| 331 | Ok(()) |
| 332 | } |
| 333 | |
| 334 | /// Asks for the pull request to be made again on its branch (a rebase |
| 335 | /// or a recreate): it stays open while its sandbox works. |
| 336 | pub async fn remake_update_pull(&self, id: &str, bump: &BumpArgs) -> Result<()> { |
| 337 | self.db |
| 338 | .prepare("UPDATE update_pulls SET bump = ?, head = NULL, error = NULL, updated_at = ? WHERE id = ?") |
| 339 | .bind(&[serde_json::to_string(bump)?.into(), now().into(), id.into()])? |
| 340 | .run() |
| 341 | .await?; |
| 342 | Ok(()) |
| 343 | } |
| 344 | |
| 345 | pub async fn set_update_pull_head(&self, id: &str, head: &str) -> Result<()> { |
| 346 | self.db |
| 347 | .prepare("UPDATE update_pulls SET head = ?, updated_at = ? WHERE id = ?") |
| 348 | .bind(&[head.into(), now().into(), id.into()])? |
| 349 | .run() |
| 350 | .await?; |
| 351 | Ok(()) |
| 352 | } |
| 353 | |
| 354 | /// Marks it looked at, so the sweep moves on to others. |
| 355 | pub async fn touch_update_pull(&self, id: &str) -> Result<()> { |
| 356 | self.db.prepare("UPDATE update_pulls SET updated_at = ? WHERE id = ?").bind(&[now().into(), id.into()])?.run().await?; |
| 357 | Ok(()) |
| 358 | } |
| 359 | |
| 360 | pub async fn set_merge_by(&self, id: &str, by: Option<&str>) -> Result<()> { |
| 361 | self.db.prepare("UPDATE update_pulls SET merge_by = ? WHERE id = ?").bind(&[optional(by), id.into()])?.run().await?; |
| 362 | Ok(()) |
| 363 | } |
| 364 | |
| 365 | // --- Ignore conditions ---------------------------------------------------- |
| 366 | |
| 367 | pub async fn ignores(&self, repo_id: &str) -> Result<Vec<IgnoreCondition>> { |
| 368 | Ok(self |
| 369 | .db |
| 370 | .prepare("SELECT * FROM update_ignores WHERE repo_id = ? ORDER BY dependency, at") |
| 371 | .bind(&[repo_id.into()])? |
| 372 | .all() |
| 373 | .await? |
| 374 | .results::<IgnoreRow>()? |
| 375 | .into_iter() |
| 376 | .map(|row| IgnoreCondition { |
| 377 | ecosystem: row.ecosystem, |
| 378 | dependency: row.dependency, |
| 379 | versions: row.versions, |
| 380 | update_type: row.update_type, |
| 381 | by: row.by, |
| 382 | pull: row.pull.map(|n| n.max(0) as u32), |
| 383 | at: row.at, |
| 384 | }) |
| 385 | .collect()) |
| 386 | } |
| 387 | |
| 388 | pub async fn add_ignore(&self, repo_id: &str, condition: &IgnoreCondition) -> Result<()> { |
| 389 | let key = condition.versions.clone().or_else(|| condition.update_type.clone()).unwrap_or_default(); |
| 390 | self.db |
| 391 | .prepare( |
| 392 | "INSERT INTO update_ignores (repo_id, ecosystem, dependency, condition, versions, update_type, by, pull, at) |
| 393 | VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) |
| 394 | ON CONFLICT (repo_id, ecosystem, dependency, condition) DO UPDATE SET by = excluded.by, pull = excluded.pull, at = excluded.at", |
| 395 | ) |
| 396 | .bind(&[ |
| 397 | repo_id.into(), |
| 398 | condition.ecosystem.as_str().into(), |
| 399 | condition.dependency.to_lowercase().as_str().into(), |
| 400 | key.as_str().into(), |
| 401 | optional(condition.versions.as_deref()), |
| 402 | optional(condition.update_type.as_deref()), |
| 403 | condition.by.as_str().into(), |
| 404 | condition.pull.map_or(JsValue::NULL, JsValue::from), |
| 405 | now().into(), |
| 406 | ])? |
| 407 | .run() |
| 408 | .await?; |
| 409 | Ok(()) |
| 410 | } |
| 411 | |
| 412 | /// Removes the conditions on `dependency` (`*` for every one) in |
| 413 | /// `ecosystem`, or only those `keep` refuses. Returns how many went. |
| 414 | pub async fn remove_ignores(&self, repo_id: &str, ecosystem: &str, dependency: &str, condition: Option<&str>) -> Result<usize> { |
| 415 | let found = self.ignores(repo_id).await?; |
| 416 | let mut removed = 0; |
| 417 | for item in found { |
| 418 | let named = dependency == "*" || item.dependency.eq_ignore_ascii_case(dependency); |
| 419 | let key = item.versions.clone().or_else(|| item.update_type.clone()).unwrap_or_default(); |
| 420 | if item.ecosystem != ecosystem || !named || condition.is_some_and(|wanted| wanted != key) { |
| 421 | continue; |
| 422 | } |
| 423 | self.db |
| 424 | .prepare("DELETE FROM update_ignores WHERE repo_id = ? AND ecosystem = ? AND dependency = ? AND condition = ?") |
| 425 | .bind(&[repo_id.into(), ecosystem.into(), item.dependency.as_str().into(), key.as_str().into()])? |
| 426 | .run() |
| 427 | .await?; |
| 428 | removed += 1; |
| 429 | } |
| 430 | Ok(removed) |
| 431 | } |
| 432 | } |