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