Skip to content
Sign up

flagon-io/g1t

public
0
428 linesCodeBlameRaw
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
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>,
45 /// The issue g1t opened for it, from before it stopped doing that.
46 #[serde(default)]
47 pub issue: Option<i64>,
48 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
63 pub fn issue(&self) -> Option<u32> {
64 self.issue.map(|n| n.max(0) as u32)
65 }
66
67 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.
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
273 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}