Skip to content
Sign up

flagon-io/g1t

public
0
428 linesCodeBlameRaw

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

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