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
117 linesCodeBlameRaw
1//! What follows for coordination when a blocker resolves, and what an
2//! agent picking up a work item sees of its blockers. What a work item is
3//! blocked by, and changing it, is items/dependencies.rs.
4//!
5//! Each blocker that closes or has a pull request merged for it is a
6//! `dependency.changed` event on the work item it blocks.
7
8use g1t_contracts::work::*;
9use g1t_contracts::User;
10use serde::Deserialize;
11use worker::Result;
12
13use super::{CLAIM_COLUMNS, ClaimRow};
14use crate::Work;
15use crate::items::{self, Held};
16
17#[derive(Deserialize)]
18struct PullBriefRow {
19 number: u32,
20 title: String,
21 status: PullStatus,
22 item_id: String,
23}
24
25impl Work {
26 /// A work item closed (`closed`) or had a pull request merged for it
27 /// (`merged_pull`): its claims end if it closed, and each open work
28 /// item it blocks hears that it cleared. Never fails what caused it: a
29 /// problem is logged.
30 pub(crate) async fn item_resolved(&self, held: &Held, actor: &User, closed: bool, merged_pull: Option<u32>) {
31 let number = held.item.number;
32 let resolved = async {
33 if closed {
34 self.release_claims_on_close(held, Some(actor)).await?;
35 }
36 for dependent in items::dependents(self, held).await? {
37 let blocked_by = dependent.item.blocked_by.clone();
38 let remaining = items::still_open(self, &dependent.item.workspace, &blocked_by, Some(number)).await?;
39 let change = DependencyChanged {
40 workspace: dependent.item.workspace.clone(),
41 number: dependent.item.number,
42 item: dependent.item.number,
43 repo_ids: dependent.repo_ids(),
44 blocked_by,
45 cleared: vec![number],
46 remaining,
47 merged_pull,
48 ..DependencyChanged::default()
49 };
50 // Follows from a close or merge already made: named by it, so
51 // a repeat of the same close or merge tells each session once.
52 let how = merged_pull.map_or_else(|| "closed".to_owned(), |pull| format!("merged:{pull}"));
53 let key = format!("{}:cleared:{}:{how}", dependent.item.id, held.item.id);
54 let data = g1t_contracts::events::marked(change, Some(actor));
55 self.publish_keyed_to("dependency.changed", &dependent.repo_ids(), Some(actor.id.clone()), data, &key).await?;
56 }
57 Ok::<(), worker::Error>(())
58 };
59 if let Err(error) = resolved.await {
60 worker::console_error!("#{number} resolved, but its dependents were not told: {error}");
61 }
62 }
63
64 /// The work items of `workspace` that `numbers` names, each with its
65 /// active claims and its pull requests, in the order given.
66 pub(crate) async fn blockers(&self, workspace: &str, numbers: &[u32]) -> Result<Vec<Blocker>> {
67 let items = items::briefs(self, workspace, numbers).await?;
68 if items.is_empty() {
69 return Ok(Vec::new());
70 }
71 let ids = serde_json::to_string(&items.iter().map(|item| item.id.as_str()).collect::<Vec<_>>())?;
72 let bind = |sql: String| -> Result<worker::D1PreparedStatement> { self.db.prepare(sql).bind(&[ids.as_str().into()]) };
73 let results = self
74 .db
75 .batch(vec![
76 bind(format!(
77 "SELECT {CLAIM_COLUMNS} FROM claims
78 WHERE state = 'active' AND item_id IN (SELECT value FROM json_each(?1))
79 ORDER BY created_at, id"
80 ))?,
81 bind(
82 "SELECT number, title, status, item_id FROM pulls
83 WHERE item_id IN (SELECT value FROM json_each(?1))
84 ORDER BY created_at"
85 .to_owned(),
86 )?,
87 ])
88 .await?;
89 let mut results = results.into_iter();
90 let mut next = || results.next();
91 let claims: Vec<(String, Claim)> = match next() {
92 Some(result) => result.results::<ClaimRow>()?.into_iter().map(|row| (row.item_id.clone(), Claim::from(row))).collect(),
93 None => Vec::new(),
94 };
95 let pulls: Vec<PullBriefRow> = match next() {
96 Some(result) => result.results()?,
97 None => Vec::new(),
98 };
99 Ok(numbers
100 .iter()
101 .filter_map(|number| {
102 let item = items.iter().find(|item| item.number == *number)?;
103 Some(Blocker {
104 number: item.number,
105 title: item.title.clone(),
106 state: item.state,
107 claims: claims.iter().filter(|(id, _)| *id == item.id).map(|(_, claim)| claim.clone()).collect(),
108 pulls: pulls
109 .iter()
110 .filter(|pull| pull.item_id == item.id)
111 .map(|pull| PullBrief { number: pull.number, title: pull.title.clone(), status: pull.status })
112 .collect(),
113 })
114 })
115 .collect())
116 }
117}