Skip to content
Sign up

flagon-io/g1t

public
00

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

gitagentscloudflarerustforgemcp
416 linesCodeBlameRaw
1//! What landing a work item does, and closing one by a commit.
2//!
3//! When a pull request merges into the default branch, or a commit naming
4//! a work item is pushed there, where the repository closes work items
5//! that way (`RepoSettings::auto_close_items`), the repository's link to
6//! each work item it closes is marked landed and the resolution recorded
7//! ([`LAND_SQL`]); the work item closes once it has landed in every target
8//! it has, or at once when it has none. What it is part of is linked
9//! (`touched`), and closes nothing.
10
11use g1t_contracts::events::WorkItemEvent;
12use g1t_contracts::repos::{BranchDriftArgs, BranchDrifts, Commit, LogArgs, Repo, RepoPath};
13use g1t_contracts::time::rfc3339;
14use g1t_contracts::work::*;
15use g1t_contracts::{FailureCode, Outcome, User};
16use g1t_kit::now_ms;
17use serde::Deserialize;
18use worker::Result;
19use worker::wasm_bindgen::JsValue;
20
21use super::links::PART_SQL;
22use super::rows::{Held, StoredResolution, path_of};
23use crate::Work;
24use crate::rows::{By, ValueRow};
25
26/// What landed: a pull request merged, or a commit pushed.
27#[derive(Clone, Debug, Default)]
28pub(crate) struct Landing {
29 pub pull: Option<u32>,
30 pub commit: Option<String>,
31 pub pushed_by: Option<String>,
32}
33
34/// Landing a work item (`?1`) in repository `?2` at `?3`, by `?4` (an
35/// account id), resolved by `?5` (a stored resolution, JSON), whose pull
36/// request and commit are `?6` and `?7`, in one batch: linked (as touched,
37/// unless it is already), marked landed there, the resolution recorded
38/// once, and closed when no target is left to land.
39pub(crate) const LAND_SQL: [&str; 4] = [
40 "INSERT INTO work_item_repos (item_id, repo_id, role, added_at, added_by)
41 VALUES (?1, ?2, 'touched', ?3, ?4) ON CONFLICT (item_id, repo_id) DO NOTHING",
42 "UPDATE work_item_repos SET landed_at = COALESCE(landed_at, ?3) WHERE item_id = ?1 AND repo_id = ?2",
43 "UPDATE work_items SET resolutions = json_insert(resolutions, '$[#]', json(?5)), updated_at = ?3
44 WHERE id = ?1 AND NOT EXISTS (SELECT 1 FROM json_each(work_items.resolutions) j
45 WHERE json_extract(j.value, '$.repo_id') = ?2 AND json_extract(j.value, '$.pull') IS ?6
46 AND json_extract(j.value, '$.commit') IS ?7)",
47 "UPDATE work_items SET state = 'closed', reason = 'completed', closed_at = ?3, updated_at = ?3
48 WHERE id = ?1 AND state = 'open'
49 AND NOT EXISTS (SELECT 1 FROM work_item_repos WHERE item_id = ?1 AND role = 'target' AND landed_at IS NULL)
50 RETURNING id AS value",
51];
52
53/// The work items a commit (`?2`) pushed to repository `?1` resolved.
54const CLOSED_BY_COMMIT: &str = "SELECT i.workspace, i.number, i.title, i.state = 'closed' AS closed, i.id AS id
55 FROM work_item_repos r JOIN work_items i ON i.id = r.item_id
56 WHERE r.repo_id = ?1 AND i.number IS NOT NULL
57 AND EXISTS (SELECT 1 FROM json_each(i.resolutions) j
58 WHERE json_extract(j.value, '$.repo_id') = ?1 AND json_extract(j.value, '$.commit') = ?2)
59 ORDER BY i.number LIMIT 50";
60
61/// The repositories of workspace `?1` with what earlier work items (not
62/// `?2`) landed in them, newest first.
63const LANDED_BEFORE: &str = "SELECT r.repo_id AS repo_id, i.title AS title, i.labels AS labels
64 FROM work_item_repos r JOIN work_items i ON i.id = r.item_id
65 WHERE i.workspace = ?1 AND i.id <> ?2 AND r.landed_at IS NOT NULL
66 ORDER BY r.landed_at DESC LIMIT ?3";
67
68#[derive(Deserialize)]
69struct ClosedRow {
70 id: String,
71 workspace: String,
72 number: u32,
73 title: String,
74 closed: u8,
75}
76
77/// A repository another work item of the workspace landed in.
78#[derive(Clone, Debug, Deserialize)]
79pub(crate) struct PastLanding {
80 pub repo_id: String,
81 pub title: String,
82 /// JSON array of label names.
83 pub labels: String,
84}
85
86/// Those of `rows` (what a commit resolved) that `viewer` may see.
87fn seen_closed(rows: Vec<ClosedRow>, held: &[Held], viewer: Option<&User>) -> Vec<ClosedItem> {
88 rows.into_iter()
89 .filter(|row| held.iter().any(|held| held.item.id == row.id && held.may(viewer, g1t_contracts::access::Capability::Read)))
90 .map(|row| ClosedItem { workspace: row.workspace, number: row.number, title: row.title, closed: row.closed != 0 })
91 .collect()
92}
93
94/// A note of a landing: where, and by what.
95pub(crate) fn landing_note(path: &str, pull: Option<u32>, commit: Option<&str>) -> String {
96 match (pull, commit) {
97 (Some(pull), _) => format!("{path}#{pull}"),
98 (None, Some(commit)) => format!("{path}@{}", &commit[..7.min(commit.len())]),
99 (None, None) => path.to_owned(),
100 }
101}
102
103impl Work {
104 /// Lands `closes` (work items, by id) in `repo` by `landing`, and links
105 /// `parts`: see the module's notes. Says so on each, with the session
106 /// it was done in when one is known; `agent` when an agent pushed for
107 /// `actor` (the note is then g1t's, for them). Returns the work items
108 /// it closed.
109 #[allow(clippy::too_many_arguments)]
110 pub(super) async fn land(
111 &self,
112 repo: &Repo,
113 landing: &Landing,
114 actor: &User,
115 closes: &[String],
116 parts: &[String],
117 session: Option<&str>,
118 agent: bool,
119 ) -> Result<Vec<Held>> {
120 let now = rfc3339(now_ms());
121 let path = path_of(repo);
122 let by = match By::from(actor) {
123 by if agent && by.acting_for.is_none() && by.id == actor.id => By {
124 id: g1t_contracts::identity::AGENT_ID,
125 name: g1t_contracts::identity::AGENT_NAME,
126 acting_for: Some((&actor.id, &actor.username)),
127 },
128 by => by,
129 };
130 let stored = StoredResolution {
131 repo_id: repo.id.clone(),
132 pull: landing.pull,
133 commit: landing.commit.clone(),
134 pushed_by: landing.pushed_by.clone(),
135 };
136 let resolution = Resolution {
137 repo: path.clone(),
138 pull: landing.pull,
139 commit: landing.commit.clone(),
140 pushed_by: landing.pushed_by.clone(),
141 };
142 let what = landing_note(&path, landing.pull, landing.commit.as_deref());
143 let mut closed = Vec::new();
144 for id in closes {
145 let Some(before) = self.item_by_id(id).await? else { continue };
146 let binds: [JsValue; 7] = [
147 id.as_str().into(),
148 repo.id.as_str().into(),
149 now.as_str().into(),
150 actor.id.as_str().into(),
151 serde_json::to_string(&stored)?.into(),
152 landing.pull.map_or(JsValue::NULL, JsValue::from),
153 landing.commit.as_deref().map_or(JsValue::NULL, JsValue::from),
154 ];
155 let statements = LAND_SQL
156 .iter()
157 .map(|sql| self.db.prepare(*sql).bind(&binds[..g1t_kit::rename::parameters(sql)]))
158 .collect::<Result<Vec<_>>>()?;
159 let results = self.db.batch(statements).await?;
160 let closed_now = results
161 .get(3)
162 .map(|result| result.results::<ValueRow>())
163 .transpose()?
164 .is_some_and(|rows| !rows.is_empty());
165 let Some(held) = self.item_by_id(id).await? else { continue };
166 if closed_now {
167 self.publish_item(
168 "work_item.closed",
169 &held,
170 Some(actor),
171 WorkItemEvent {
172 reason: Some(CloseReason::Completed.as_str()),
173 resolution: Some(resolution.clone()),
174 missing: Some(Vec::new()),
175 session: session.map(str::to_owned),
176 ..Self::item_event(&held)
177 },
178 )
179 .await?;
180 let text = match landing.pull {
181 Some(_) => format!("closed this by merging {what}"),
182 None => format!("closed this in {what}"),
183 };
184 self.note_item(id, by, &crate::pushed::closed_note(&text, session)).await?;
185 self.item_resolved(&held, actor, true, Some(landing)).await;
186 closed.push(held);
187 continue;
188 }
189 self.publish_item(
190 "work_item.landed",
191 &held,
192 Some(actor),
193 WorkItemEvent {
194 resolution: Some(resolution.clone()),
195 missing: Some(held.item.missing()),
196 session: session.map(str::to_owned),
197 ..Self::item_event(&held)
198 },
199 )
200 .await?;
201 if before.item.state == State::Open {
202 let missing = held.item.missing();
203 let text = if missing.is_empty() {
204 format!("landed {what}")
205 } else {
206 format!("landed {what}; still to land in {}", missing.join(", "))
207 };
208 self.note_item(id, by, &crate::pushed::closed_note(&text, session)).await?;
209 // Landed for it: what it blocks hears, though it stays open
210 // and they still wait on it.
211 self.item_resolved(&held, actor, false, Some(landing)).await;
212 }
213 }
214 for id in parts {
215 let linked = self
216 .db
217 .prepare(PART_SQL)
218 .bind(&[id.as_str().into(), repo.id.as_str().into(), now.as_str().into(), actor.id.as_str().into()])?
219 .first::<ValueRow>(None)
220 .await?;
221 if linked.is_none() {
222 continue;
223 }
224 let Some(held) = self.item_by_id(id).await? else { continue };
225 self.publish_item(
226 "work_item.linked",
227 &held,
228 Some(actor),
229 WorkItemEvent { repo: Some(path.clone()), role: Some(LinkRole::Touched), ..Self::item_event(&held) },
230 )
231 .await?;
232 self.note_item(id, by, &format!("linked {path}: {what} is part of this")).await?;
233 }
234 Ok(closed)
235 }
236
237 /// `close_work_item` with a commit: the commit, on the default branch
238 /// of `repo`, whose message names work item `held` after a closing
239 /// keyword, with the session it names, if it names one of this
240 /// repository's. Anything else is refused, saying why.
241 pub(super) async fn closing_commit(
242 &self,
243 repo: &Repo,
244 held: &Held,
245 commit: &str,
246 actor: &User,
247 ) -> Result<Outcome<(String, Option<String>)>> {
248 let number = held.item.number;
249 let commit = commit.trim().to_lowercase();
250 if !(4..=64).contains(&commit.len()) || !commit.bytes().all(|b| b.is_ascii_hexdigit()) {
251 return Ok(Outcome::fail(FailureCode::Invalid, "commit is a commit's hash."));
252 }
253 let path = RepoPath { namespace: repo.namespace.clone(), name: repo.name.clone() };
254 let log = |at: Option<String>| LogArgs { path: path.clone(), viewer: Some(actor.clone()), git_ref: at, limit: 1 };
255 let (named, tip) = (log(Some(commit.clone())), log(None));
256 let (found, head) = futures_util::future::join(
257 g1t_kit::call::<_, Outcome<Vec<Commit>>>(&self.repos, "log", &named),
258 g1t_kit::call::<_, Outcome<Vec<Commit>>>(&self.repos, "log", &tip),
259 )
260 .await;
261 let found = match found? {
262 Outcome::Ok(commits) => commits.into_iter().next().filter(|found| found.hash.starts_with(&commit)),
263 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
264 };
265 let Some(found) = found else {
266 return Ok(Outcome::fail(
267 FailureCode::NotFound,
268 format!("There is no commit {commit} in {}/{}; give its full hash.", repo.namespace, repo.name),
269 ));
270 };
271 let short = &found.hash[..7.min(found.hash.len())];
272 let mut names = false;
273 for reference in refs::named(&found.message).closes {
274 if self.ref_item(repo, &reference).await?.as_deref() == Some(held.item.id.as_str()) {
275 names = true;
276 break;
277 }
278 }
279 if !names {
280 return Ok(Outcome::fail(
281 FailureCode::Invalid,
282 format!("Commit {short} does not close #{number}: its message names it after no closing keyword (Fixes #{number})."),
283 ));
284 }
285 let head = match head? {
286 Outcome::Ok(commits) => commits.into_iter().next(),
287 Outcome::Fail(_) => None,
288 };
289 let Some(head) = head else {
290 return Ok(Outcome::fail(FailureCode::Invalid, format!("Commit {short} is not on {}.", repo.default_branch)));
291 };
292 if head.hash != found.hash {
293 let drifts: Outcome<BranchDrifts> = g1t_kit::call(
294 &self.repos,
295 "branch_drift",
296 &BranchDriftArgs { path: path.clone(), viewer: Some(actor.clone()), base: head.hash, heads: vec![found.hash.clone()] },
297 )
298 .await?;
299 let drift = match drifts {
300 Outcome::Ok(drifts) => drifts.branches.into_iter().next().and_then(|branch| branch.drift),
301 Outcome::Fail(_) => None,
302 };
303 match drift {
304 Some(drift) if drift.ahead == 0 => {}
305 Some(_) => {
306 return Ok(Outcome::fail(
307 FailureCode::Invalid,
308 format!("Commit {short} is not on {}, the default branch: only a commit there closes a work item.", repo.default_branch),
309 ));
310 }
311 None => {
312 return Ok(Outcome::fail(
313 FailureCode::Invalid,
314 format!("Could not tell whether commit {short} is on {}: it is too far back.", repo.default_branch),
315 ));
316 }
317 }
318 }
319 let session = self.trailer_session(&repo.id, &found.message).await?;
320 Ok(Outcome::Ok((found.hash, session)))
321 }
322
323 /// `closed_by_commit`: the work items a commit pushed to a repository
324 /// resolved, for anyone who may read the repository: those of them
325 /// they may see.
326 pub(crate) async fn closed_by_commit(&self, a: ClosedByCommitArgs) -> Result<Outcome<Vec<ClosedItem>>> {
327 let Some(commit) = crate::commit_comments::valid_commit(&a.commit).filter(|commit| commit.len() >= 40) else {
328 return Ok(Outcome::Ok(Vec::new()));
329 };
330 let repo = match self.repo(&a.repo, &a.viewer).await? {
331 Outcome::Ok(repo) => repo,
332 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
333 };
334 let rows = self.db.prepare(CLOSED_BY_COMMIT).bind(&[repo.id.as_str().into(), commit.into()])?.all().await?.results::<ClosedRow>()?;
335 let ids: Vec<String> = rows.iter().map(|row| row.id.clone()).collect();
336 let held = self.items_by_ids(&ids).await?;
337 let viewer = a.viewer.as_ref();
338 Ok(Outcome::Ok(seen_closed(rows, &held, viewer)))
339 }
340
341 /// Where the other work items of `held`'s workspace landed, newest
342 /// first, at most `limit`.
343 pub(super) async fn landed_before(&self, held: &Held, limit: u32) -> Result<Vec<PastLanding>> {
344 Ok(self
345 .db
346 .prepare(LANDED_BEFORE)
347 .bind(&[held.item.workspace.as_str().into(), held.item.id.as_str().into(), limit.into()])?
348 .all()
349 .await?
350 .results()?)
351 }
352}
353
354#[cfg(test)]
355mod tests {
356 use rusqlite::params;
357
358 use super::*;
359
360 #[test]
361 fn a_commit_names_only_the_work_items_the_viewer_may_see() {
362 use crate::items::rows_for_test::{held_for_test, person_for_test};
363 let row = |id: &str, number: u32| ClosedRow { id: id.into(), workspace: "acme".into(), number, title: format!("Item {number}"), closed: 1 };
364 // #1 filed in a public repository, #2 a members-only one.
365 let held = vec![held_for_test(1, &[("rep_web", "origin", "acme", false)], &[]), held_for_test(2, &[], &[])];
366 let numbers = |viewer: Option<&User>| -> Vec<u32> {
367 seen_closed(vec![row("wki_1", 1), row("wki_2", 2)], &held, viewer).iter().map(|item| item.number).collect()
368 };
369 assert_eq!(numbers(None), vec![1]);
370 assert_eq!(numbers(Some(&person_for_test("usr_9", None))), vec![1]);
371 assert_eq!(numbers(Some(&person_for_test("usr_2", Some("acme")))), vec![1, 2]);
372 }
373
374 #[test]
375 fn a_landing_is_named_by_its_pull_request_or_commit() {
376 assert_eq!(landing_note("acme/web", Some(12), None), "acme/web#12");
377 assert_eq!(landing_note("acme/web", None, Some("abcdef1234")), "acme/web@abcdef1");
378 }
379
380 #[test]
381 fn landing_statements_take_the_item_repository_and_what_landed() {
382 let counts: Vec<usize> = LAND_SQL.iter().map(|sql| g1t_kit::rename::parameters(sql)).collect();
383 assert_eq!(counts, vec![4, 3, 7, 3]);
384 assert_eq!(g1t_kit::rename::parameters(PART_SQL), 4);
385 }
386
387 #[test]
388 fn a_commit_names_what_it_resolved_in_its_repository() {
389 let db = crate::migrated::database();
390 let resolved = |repo: &str| format!(r#"[{{"repo_id":"{repo}","pull":null,"commit":"c1","pushed_by":"ana"}}]"#);
391 let item = |id: &str, number: u32, title: &str, state: &str, repo: &str| {
392 db.execute(
393 "INSERT INTO work_items (id, workspace, number, title, body, state, resolutions, author_id, author_name, created_at, updated_at)
394 VALUES (?1, 'acme', ?2, ?3, '', ?4, ?5, 'user', 'user', 'now', 'now')",
395 params![id, number, title, state, resolved(repo)],
396 )
397 .unwrap();
398 };
399 item("wki_1", 7, "Broken header", "closed", "repo");
400 item("wki_2", 8, "Footer", "open", "repo");
401 item("wki_3", 9, "Elsewhere", "closed", "other");
402 db.execute_batch(
403 "INSERT INTO work_item_repos (item_id, repo_id, role, added_at) VALUES
404 ('wki_1', 'repo', 'origin', 'now'), ('wki_2', 'repo', 'touched', 'now'), ('wki_3', 'other', 'origin', 'now');",
405 )
406 .unwrap();
407 let closed: Vec<(String, u32, String, u8)> = db
408 .prepare(CLOSED_BY_COMMIT)
409 .unwrap()
410 .query_map(params!["repo", "c1"], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)))
411 .unwrap()
412 .map(Result::unwrap)
413 .collect();
414 assert_eq!(closed, vec![("acme".to_owned(), 7, "Broken header".to_owned(), 1), ("acme".to_owned(), 8, "Footer".to_owned(), 0)]);
415 }
416}