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
192 linesCodeBlameRaw
1//! What a work item is blocked by: the work items of the same workspace
2//! that have to land first (`work_items.blocked_by`, a JSON array of ids;
3//! numbers in and out).
4//!
5//! `update_work_item` replaces the set, refusing a number that is not one
6//! of the workspace's work items, the item itself, or a set that would
7//! make a cycle. Each change is a `dependency.changed` event on the work
8//! item; what follows a blocker closing is coordination's
9//! (`coordination/dependencies.rs`).
10
11use std::collections::{HashMap, HashSet};
12
13use g1t_contracts::work::*;
14use g1t_contracts::{FailureCode, Outcome, User};
15use serde::Deserialize;
16use worker::Result;
17
18use super::reading::{held_rows, select};
19use super::rows::Held;
20use crate::Work;
21
22#[derive(Deserialize)]
23struct EdgeRow {
24 number: u32,
25 /// JSON array of the numbers it is blocked by.
26 blocked_by: Option<String>,
27}
28
29/// A work item in brief: its id, number, title and state.
30#[derive(Clone, Debug, Deserialize)]
31pub(crate) struct Brief {
32 pub id: String,
33 pub number: u32,
34 pub title: String,
35 pub state: State,
36}
37
38fn numbers(json: Option<&str>) -> Vec<u32> {
39 json.and_then(|json| serde_json::from_str(json).ok()).unwrap_or_default()
40}
41
42/// A workspace's (`?1`) work items that are blocked by something, each
43/// with what it is blocked by, by number.
44const EDGES: &str = "SELECT number,
45 (SELECT json_group_array(b.number) FROM json_each(work_items.blocked_by) j JOIN work_items b ON b.id = j.value
46 WHERE b.number IS NOT NULL) AS blocked_by
47 FROM work_items WHERE workspace = ?1 AND blocked_by <> '[]' AND number IS NOT NULL";
48
49/// The work items of workspace `?1` numbered in `?2` (a JSON array), in
50/// brief.
51const BRIEFS: &str = "SELECT id, number, title, state FROM work_items
52 WHERE workspace = ?1 AND number IN (SELECT value FROM json_each(?2))";
53
54impl Work {
55 /// `asked` as `number`'s new `blocked_by` in `workspace`, or why it
56 /// cannot be.
57 pub(super) async fn checked_blocked_by(&self, workspace: &str, number: u32, asked: &[u32]) -> Result<Outcome<Vec<u32>>> {
58 let wanted = serde_json::to_string(asked)?;
59 let results = self
60 .db
61 .batch(vec![
62 self.db.prepare(BRIEFS).bind(&[workspace.into(), wanted.into()])?,
63 self.db.prepare(EDGES).bind(&[workspace.into()])?,
64 ])
65 .await?;
66 let mut results = results.into_iter();
67 let found: Vec<Brief> = match results.next() {
68 Some(result) => result.results()?,
69 None => Vec::new(),
70 };
71 let edges: Vec<EdgeRow> = match results.next() {
72 Some(result) => result.results()?,
73 None => Vec::new(),
74 };
75 let items: HashSet<u32> = found.iter().map(|row| row.number).collect();
76 let graph: HashMap<u32, Vec<u32>> = edges.into_iter().map(|row| (row.number, numbers(row.blocked_by.as_deref()))).collect();
77 Ok(match check_blocked_by(number, asked, &items, &graph) {
78 Ok(set) => Outcome::Ok(set),
79 Err(message) => Outcome::fail(FailureCode::Invalid, message),
80 })
81 }
82
83 /// The `dependency.changed` that setting `held`'s `blocked_by` to
84 /// `after` makes, worked out before the change so that its updates are
85 /// written with it; none when nothing changes.
86 pub(super) async fn dependency_change(&self, held: &Held, after: &[u32], actor: &User) -> Result<Option<serde_json::Value>> {
87 let before = &held.item.blocked_by;
88 let added: Vec<u32> = after.iter().filter(|n| !before.contains(n)).copied().collect();
89 let removed: Vec<u32> = before.iter().filter(|n| !after.contains(n)).copied().collect();
90 if added.is_empty() && removed.is_empty() {
91 return Ok(None);
92 }
93 let remaining = self.still_open(&held.item.workspace, after, None).await?;
94 let change = DependencyChanged {
95 workspace: held.item.workspace.clone(),
96 number: held.item.number,
97 item: held.item.number,
98 repo_ids: held.repo_ids(),
99 blocked_by: after.to_vec(),
100 added,
101 removed,
102 cleared: Vec::new(),
103 remaining,
104 merged_pull: None,
105 };
106 Ok(Some(g1t_contracts::events::marked(change, Some(actor))))
107 }
108
109 /// Notes a change to `held`'s `blocked_by` (from `before`) in its
110 /// conversation.
111 pub(super) async fn blocked_by_changed(&self, held: &Held, before: &[u32], actor: &User) -> Result<()> {
112 let added: Vec<u32> = held.item.blocked_by.iter().filter(|n| !before.contains(n)).copied().collect();
113 let removed: Vec<u32> = before.iter().filter(|n| !held.item.blocked_by.contains(n)).copied().collect();
114 if added.is_empty() && removed.is_empty() {
115 return Ok(());
116 }
117 let list = |numbers: &[u32]| numbers.iter().map(|n| format!("#{n}")).collect::<Vec<_>>().join(", ");
118 if !added.is_empty() {
119 self.note_item(&held.item.id, actor, &format!("marked this as blocked by {}", list(&added))).await?;
120 }
121 if !removed.is_empty() {
122 self.note_item(&held.item.id, actor, &format!("marked this as no longer blocked by {}", list(&removed))).await?;
123 }
124 Ok(())
125 }
126
127 /// Those of `numbers` whose work item is still open, leaving out
128 /// `cleared` (a blocker that just landed).
129 pub(super) async fn still_open(&self, workspace: &str, numbers: &[u32], cleared: Option<u32>) -> Result<Vec<u32>> {
130 if numbers.is_empty() {
131 return Ok(Vec::new());
132 }
133 let open: Vec<Brief> = self
134 .db
135 .prepare(
136 "SELECT id, number, title, state FROM work_items
137 WHERE workspace = ? AND state = 'open' AND number IN (SELECT value FROM json_each(?))",
138 )
139 .bind(&[workspace.into(), serde_json::to_string(numbers)?.into()])?
140 .all()
141 .await?
142 .results()?;
143 Ok(numbers
144 .iter()
145 .filter(|number| Some(**number) != cleared && open.iter().any(|row| row.number == **number))
146 .copied()
147 .collect())
148 }
149
150 /// The open work items `held` blocks.
151 pub(super) async fn dependents(&self, held: &Held) -> Result<Vec<Held>> {
152 held_rows(
153 &self
154 .db
155 .prepare(select(
156 "workspace = ? AND state = 'open'
157 AND EXISTS (SELECT 1 FROM json_each(work_items.blocked_by) WHERE json_each.value = ?)",
158 ))
159 .bind(&[held.item.workspace.as_str().into(), held.item.id.as_str().into()])?
160 .all()
161 .await?,
162 )
163 }
164
165 /// The work items of `workspace` that `numbers` names, in brief, in no
166 /// particular order.
167 pub(super) async fn briefs(&self, workspace: &str, numbers: &[u32]) -> Result<Vec<Brief>> {
168 if numbers.is_empty() {
169 return Ok(Vec::new());
170 }
171 Ok(self.db.prepare(BRIEFS).bind(&[workspace.into(), serde_json::to_string(numbers)?.into()])?.all().await?.results()?)
172 }
173
174 /// The work items that name `held` in their `blocked_by`, by number.
175 pub(super) async fn blocked_items(&self, held: &Held) -> Result<Vec<ItemBrief>> {
176 Ok(self
177 .db
178 .prepare(
179 "SELECT id, number, title, state FROM work_items
180 WHERE workspace = ? AND number IS NOT NULL
181 AND EXISTS (SELECT 1 FROM json_each(work_items.blocked_by) WHERE json_each.value = ?)
182 ORDER BY number LIMIT 100",
183 )
184 .bind(&[held.item.workspace.as_str().into(), held.item.id.as_str().into()])?
185 .all()
186 .await?
187 .results::<Brief>()?
188 .into_iter()
189 .map(|row| ItemBrief { number: row.number, title: row.title, state: row.state })
190 .collect())
191 }
192}