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
419 linesCodeBlameRaw
1//! Opening and changing work items: `open_work_item`, `update_work_item`,
2//! `close_work_item` and `reopen_work_item`, and what each says.
3
4use g1t_contracts::access::{self, Capability, RepoRef};
5use g1t_contracts::events::{EventLabel, WorkItemEvent};
6use g1t_contracts::repos::Repo;
7use g1t_contracts::time::rfc3339;
8use g1t_contracts::work::*;
9use g1t_contracts::{FailureCode, Outcome, User, new_id};
10use g1t_kit::now_ms;
11use worker::Result;
12
13use super::landing::Landing;
14use super::rows::{Held, no_item, path_of, workspace_of};
15use crate::{UNVERIFIED, Work, optional, optional_number, valid_title};
16
17/// What closing work item `number` by a commit in `landed_in` says when it
18/// stays open: the targets it still has to land in (`missing`, by path; an
19/// empty one for a repository the actor may not read).
20pub(crate) fn still_open(number: u32, landed_in: &str, missing: &[String]) -> String {
21 let named: Vec<&str> = missing.iter().map(String::as_str).filter(|path| !path.is_empty()).collect();
22 let hidden = missing.len() - named.len();
23 let mut targets = named.join(", ");
24 if hidden > 0 {
25 let more = if hidden == 1 { "1 other repository".to_owned() } else { format!("{hidden} other repositories") };
26 targets = if targets.is_empty() { more } else { format!("{targets} and {more}") };
27 }
28 format!("Recorded the commit as landed in {landed_in}, but #{number} stays open until it lands in {targets} too.")
29}
30
31impl Work {
32 pub(super) fn item_event(held: &Held) -> WorkItemEvent {
33 WorkItemEvent {
34 item_id: held.item.id.clone(),
35 workspace: held.item.workspace.clone(),
36 number: held.item.number,
37 repo_ids: held.repo_ids(),
38 author: Some((&held.item.author).into()),
39 requested_by: held.item.requested_by.as_ref().map(Into::into),
40 ..WorkItemEvent::default()
41 }
42 }
43
44 /// Publishes a `work_item.*` event about `held`, filed under its origin.
45 pub(super) async fn publish_item(&self, kind: &'static str, held: &Held, actor: Option<&User>, data: WorkItemEvent) -> Result<()> {
46 let data = g1t_contracts::events::marked(data, actor);
47 self.publish_on(kind, held.event_repo(), actor.map(|actor| actor.id.clone()), data).await
48 }
49
50 /// Notes something that happened to a work item, in its conversation.
51 pub(super) async fn note_item<'a>(&self, item_id: &str, author: impl Into<crate::rows::By<'a>>, text: &str) -> Result<()> {
52 self.note_on(crate::On::Item(item_id), author.into(), text).await
53 }
54
55 // --- RPC methods ----------------------------------------------------------
56
57 pub(crate) async fn open_work_item(&self, a: OpenWorkItemArgs) -> Result<Outcome<WorkItem>> {
58 if !a.actor.verified {
59 return Ok(Outcome::fail(FailureCode::Forbidden, UNVERIFIED));
60 }
61 let title = match valid_title(&a.title) {
62 Ok(title) => title.to_owned(),
63 Err(message) => return Ok(Outcome::fail(FailureCode::Invalid, message)),
64 };
65 let Some(labels) = normalize_labels(&a.labels) else {
66 return Ok(Outcome::fail(
67 FailureCode::Invalid,
68 format!("A work item can have up to {MAX_LABELS} labels of up to {MAX_LABEL_CHARS} characters each."),
69 ));
70 };
71 let workspace = workspace_of(&a.workspace, a.repo.as_ref());
72 if workspace.is_empty() {
73 return Ok(Outcome::fail(FailureCode::Invalid, "Say which workspace, or which repository it is filed from."));
74 }
75 // Numbered only once the issues before it are, so every new number
76 // comes after the old ones.
77 if !self.items_moved().await? {
78 return Ok(super::labels::still_moving());
79 }
80 let actor = Some(a.actor.clone());
81 // Where it is filed from: someone who may take part there may file
82 // it; without one, only a member of the workspace.
83 let origin = match &a.repo {
84 Some(path) => {
85 let repo = check!(self.repo(path, &actor).await?);
86 if !repo.namespace.eq_ignore_ascii_case(&workspace) {
87 return Ok(Outcome::fail(FailureCode::Invalid, format!("{} is not a repository of {workspace}.", path_of(&repo))));
88 }
89 check!(crate::retired::writable(&repo));
90 check!(crate::allowed(Some(&a.actor), &repo, Capability::Participate));
91 if !self.work_items_enabled(&repo).await? {
92 return Ok(Outcome::fail(FailureCode::Forbidden, crate::settings::work_items_off(&repo)));
93 }
94 self.remember_repo(&repo).await?;
95 Some(repo)
96 }
97 None => {
98 if access::workspace_role(&a.actor, &workspace).is_none() {
99 return Ok(Outcome::fail(
100 FailureCode::Forbidden,
101 format!("Only members of {workspace} can file a work item there without a repository to file it from."),
102 ));
103 }
104 None
105 }
106 };
107 let origin_ref = origin.as_ref().map(RepoRef::from);
108 let may = |capability| access::can_item(Some(&a.actor), &workspace, origin_ref, capability);
109 // Where it should land: repositories of the workspace the actor can see.
110 let mut targets: Vec<Repo> = Vec::new();
111 for path in &a.targets {
112 let repo = check!(self.repo(path, &actor).await?);
113 if !repo.namespace.eq_ignore_ascii_case(&workspace) {
114 return Ok(Outcome::fail(FailureCode::Invalid, format!("{} is not a repository of {workspace}.", path_of(&repo))));
115 }
116 if origin.as_ref().is_some_and(|origin| origin.id == repo.id) || targets.iter().any(|known| known.id == repo.id) {
117 continue;
118 }
119 self.remember_repo(&repo).await?;
120 targets.push(repo);
121 }
122 if !targets.is_empty() && origin.is_some() && !may(Capability::Triage) {
123 return Ok(Outcome::fail(FailureCode::Forbidden, access::needs_on_item(Capability::Triage, &workspace)));
124 }
125 let colors = check!(self.ensure_labels(&a.actor, &workspace, &labels).await?);
126 let milestone = match a.milestone.filter(|number| *number > 0) {
127 Some(number) => {
128 if !may(Capability::Triage) {
129 return Ok(Outcome::fail(FailureCode::Forbidden, access::needs_on_item(Capability::Triage, &workspace)));
130 }
131 check!(self.milestone_ref(&workspace, number).await?)
132 }
133 None => None,
134 };
135 let now = now_ms();
136 let id = new_id("wki", now);
137 let number = self.next_number(&workspace, origin.as_ref().map(|repo| repo.id.as_str())).await?;
138 let timestamp = rfc3339(now);
139 let (author, requested_by) = authorship(&a.actor);
140 let mut statements = vec![self
141 .db
142 .prepare(
143 "INSERT INTO work_items
144 (id, workspace, number, title, body, labels, author_id, author_name, requested_by_id,
145 requested_by_name, created_at, updated_at, milestone)
146 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
147 )
148 .bind(&[
149 id.as_str().into(),
150 workspace.as_str().into(),
151 number.into(),
152 title.as_str().into(),
153 a.body.trim().into(),
154 serde_json::to_string(&labels)?.into(),
155 author.id.as_str().into(),
156 author.username.as_str().into(),
157 optional(&requested_by.as_ref().map(|user| user.id.clone())),
158 optional(&requested_by.as_ref().map(|user| user.username.clone())),
159 timestamp.as_str().into(),
160 timestamp.as_str().into(),
161 optional_number(milestone.as_ref().map(|milestone| milestone.number)),
162 ])?];
163 let links = origin.iter().map(|repo| (repo, LinkRole::Origin)).chain(targets.iter().map(|repo| (repo, LinkRole::Target)));
164 for (repo, role) in links {
165 statements.push(
166 self.db
167 .prepare(
168 "INSERT INTO work_item_repos (item_id, repo_id, role, path, added_at, added_by)
169 VALUES (?, ?, ?, ?, ?, ?)",
170 )
171 .bind(&[
172 id.as_str().into(),
173 repo.id.as_str().into(),
174 role.as_str().into(),
175 path_of(repo).into(),
176 timestamp.as_str().into(),
177 a.actor.id.as_str().into(),
178 ])?,
179 );
180 }
181 self.db.batch(statements).await?;
182 let Some(held) = self.item_by_id(&id).await? else {
183 return Ok(no_item());
184 };
185 self.publish_item("work_item.opened", &held, Some(&a.actor), WorkItemEvent { title: Some(held.item.title.clone()), ..Self::item_event(&held) })
186 .await?;
187 // Opened with labels and a milestone: each is said, as it would be
188 // if they were added afterwards, without notes in the conversation.
189 for label in &held.item.labels {
190 let color = colors.iter().find(|(name, _)| name == label).map_or_else(|| label_color_for(label), |(_, c)| c.clone());
191 let label = Some(EventLabel { name: label.clone(), color });
192 self.publish_item("work_item.labeled", &held, Some(&a.actor), WorkItemEvent { label, ..Self::item_event(&held) }).await?;
193 }
194 if let Some(milestone) = milestone {
195 self.publish_item(
196 "work_item.milestoned",
197 &held,
198 Some(&a.actor),
199 WorkItemEvent { milestone: Some(milestone), ..Self::item_event(&held) },
200 )
201 .await?;
202 }
203 Ok(Outcome::Ok(held.shown(Some(&a.actor))))
204 }
205
206
207 pub(crate) async fn update_work_item(&self, a: UpdateWorkItemArgs) -> Result<Outcome<WorkItem>> {
208 let held = check!(self.manageable_item(&a.actor, &a.workspace, a.number).await?);
209 let title = match a.title.as_deref().map(valid_title) {
210 Some(Err(message)) => return Ok(Outcome::fail(FailureCode::Invalid, message)),
211 Some(Ok(title)) => Some(title.to_owned()),
212 None => None,
213 };
214 if a.labels.as_deref().is_some_and(|labels| normalize_labels(labels).is_none()) {
215 return Ok(Outcome::fail(
216 FailureCode::Invalid,
217 format!("A work item can have up to {MAX_LABELS} labels of up to {MAX_LABEL_CHARS} characters each."),
218 ));
219 }
220 let assignees = match a.assignees {
221 Some(names) => Some(check!(self.valid_assignees(names).await?)),
222 None => None,
223 };
224 // What it is blocked by: checked before anything changes.
225 let blocked_by = match &a.blocked_by {
226 Some(asked) => Some(check!(self.checked_blocked_by(&held, &a.actor, asked).await?)),
227 None => None,
228 };
229 // Its labels and milestone first: either can be refused, and then
230 // nothing else changes.
231 if let Some(number) = a.milestone {
232 check!(self.set_milestone(&a.actor, &super::labels::Labeled::Item(held.clone()), number).await?);
233 }
234 if let Some(labels) = &a.labels {
235 check!(self.relabel(&a.actor, &super::labels::Labeled::Item(held.clone()), labels).await?);
236 }
237 let assigned = assignees.as_ref().map(serde_json::to_string).transpose()?;
238 let blocked_ids = match &blocked_by {
239 Some(numbers) => Some(serde_json::to_string(&self.item_ids(&held.item.workspace, numbers).await?)?),
240 None => None,
241 };
242 let body = a.body.map(|body| body.trim().to_owned());
243 let updated_at = rfc3339(now_ms());
244 let dependency = match &blocked_by {
245 Some(after) => self.dependency_change(&held, after, &a.actor).await?,
246 None => None,
247 };
248 let mut statements = vec![self
249 .db
250 .prepare(
251 "UPDATE work_items
252 SET title = COALESCE(?, title), body = COALESCE(?, body),
253 assignees = COALESCE(?, assignees), blocked_by = COALESCE(?, blocked_by),
254 updated_at = ?
255 WHERE id = ?",
256 )
257 .bind(&[
258 optional(&title),
259 optional(&body),
260 optional(&assigned),
261 optional(&blocked_ids),
262 updated_at.as_str().into(),
263 held.item.id.as_str().into(),
264 ])?];
265 // A change to what it is blocked by reaches the sessions it concerns
266 // with the change itself.
267 let dependency_sent = match (&dependency, &blocked_ids) {
268 (Some(data), Some(blocked)) => crate::coordination::updates::push_delivery(
269 &mut statements,
270 self.delivery_to(
271 "dependency.changed",
272 &held.repo_ids(),
273 data.clone(),
274 &format!("{}:blocked_by:{updated_at}", held.item.id),
275 "EXISTS (SELECT 1 FROM work_items WHERE id = ?7 AND blocked_by = ?8 AND updated_at = ?9)",
276 vec![held.item.id.as_str().into(), blocked.as_str().into(), updated_at.as_str().into()],
277 )?,
278 ),
279 _ => None,
280 };
281 self.db.batch(statements).await?;
282 let before = held.item.assignees.clone();
283 let blocked_before = held.item.blocked_by.clone();
284 let Some(held) = self.item_by_id(&held.item.id).await? else {
285 return Ok(no_item());
286 };
287 self.publish_item("work_item.updated", &held, Some(&a.actor), Self::item_event(&held)).await?;
288 if let Some(data) = dependency {
289 self.delivered(dependency_sent, "dependency.changed", held.event_repo().unwrap_or_default(), Some(a.actor.id.clone()), data)
290 .await?;
291 self.blocked_by_changed(&held, &blocked_before, &a.actor).await?;
292 }
293 if let Some(assignees) = assignees {
294 self.note_changes(crate::On::Item(&held.item.id), &a.actor, &before, &assignees, ("assigned", "unassigned")).await?;
295 let added: Vec<String> = assignees.iter().filter(|name| !before.contains(name)).cloned().collect();
296 self.publish_item(
297 "work_item.assigned",
298 &held,
299 Some(&a.actor),
300 WorkItemEvent { assignees: Some(assignees), added: Some(added), ..Self::item_event(&held) },
301 )
302 .await?;
303 }
304 Ok(Outcome::Ok(held.shown(Some(&a.actor))))
305 }
306
307
308 pub(crate) async fn close_work_item(&self, args: CloseWorkItemArgs) -> Result<Outcome<WorkItem>> {
309 let CloseWorkItemArgs { item: a, repo, commit } = args;
310 let held = check!(self.manageable_item(&a.actor, &a.workspace, a.number).await?);
311 if held.item.state == State::Closed {
312 return Ok(Outcome::fail(FailureCode::Conflict, "This work item is already closed."));
313 }
314 // Closed by a commit on a repository's default branch: as its push
315 // would have.
316 if let Some(commit) = commit.as_deref() {
317 if a.reason == Some(CloseReason::NotPlanned) {
318 return Ok(Outcome::fail(FailureCode::Invalid, "A commit closes a work item as completed, not as not planned."));
319 }
320 let Some(path) = repo.as_ref() else {
321 return Ok(Outcome::fail(FailureCode::Invalid, "Say which repository the commit is in (repo)."));
322 };
323 let found = check!(self.repo(path, &Some(a.actor.clone())).await?);
324 let (commit, named) = check!(self.closing_commit(&found, &held, commit, &a.actor).await?);
325 let session = match named {
326 Some(session) => Some(session),
327 None => self.caller_session(&held, &a.actor).await?,
328 };
329 let landing = Landing { pull: None, commit: Some(commit), pushed_by: Some(a.actor.username.clone()) };
330 self.land(&found, &landing, &a.actor, std::slice::from_ref(&held.item.id), &[], session.as_deref(), false).await?;
331 return Ok(match self.item_by_id(&held.item.id).await? {
332 Some(after) if after.item.state == State::Open && !after.item.missing().is_empty() => {
333 let missing: Vec<String> = after
334 .item
335 .repos
336 .iter()
337 .filter(|repo| repo.role == LinkRole::Target && !repo.landed)
338 .map(|repo| if after.may_read_repo(Some(&a.actor), &repo.repo_id) { repo.path.clone() } else { String::new() })
339 .collect();
340 Outcome::fail(FailureCode::Conflict, still_open(after.item.number, &path_of(&found), &missing))
341 }
342 Some(after) => Outcome::Ok(after.shown(Some(&a.actor))),
343 None => no_item(),
344 });
345 }
346 let session = self.caller_session(&held, &a.actor).await?;
347 let reason = a.reason.unwrap_or(CloseReason::Completed);
348 let now = rfc3339(now_ms());
349 self.db
350 .prepare("UPDATE work_items SET state = 'closed', reason = ?, closed_at = ?, updated_at = ? WHERE id = ?")
351 .bind(&[reason.as_str().into(), now.as_str().into(), now.as_str().into(), held.item.id.as_str().into()])?
352 .run()
353 .await?;
354 self.publish_item(
355 "work_item.closed",
356 &held,
357 Some(&a.actor),
358 WorkItemEvent { reason: Some(reason.as_str()), session: session.clone(), ..Self::item_event(&held) },
359 )
360 .await?;
361 let note = match reason {
362 CloseReason::Completed => "closed this as completed",
363 CloseReason::NotPlanned => "closed this as not planned",
364 };
365 self.note_item(&held.item.id, &a.actor, &crate::pushed::closed_note(note, session.as_deref())).await?;
366 // Its claims end, and the work items it blocks hear (coordination/).
367 self.item_resolved(&held, &a.actor, true, None).await;
368 let mut item = held.shown(Some(&a.actor));
369 item.state = State::Closed;
370 item.reason = Some(reason);
371 item.closed_at = Some(now.clone());
372 item.updated_at = now;
373 Ok(Outcome::Ok(item))
374 }
375
376 pub(crate) async fn reopen_work_item(&self, a: WorkItemActionArgs) -> Result<Outcome<WorkItem>> {
377 let held = check!(self.manageable_item(&a.actor, &a.workspace, a.number).await?);
378 if held.item.state == State::Open {
379 return Ok(Outcome::fail(FailureCode::Conflict, "This work item is already open."));
380 }
381 let now = rfc3339(now_ms());
382 // Reopened, nothing has landed for it any more.
383 self.db
384 .batch(vec![
385 self.db
386 .prepare(
387 "UPDATE work_items
388 SET state = 'open', reason = NULL, resolutions = '[]', closed_at = NULL, updated_at = ?
389 WHERE id = ?",
390 )
391 .bind(&[now.as_str().into(), held.item.id.as_str().into()])?,
392 self.db.prepare("UPDATE work_item_repos SET landed_at = NULL WHERE item_id = ?").bind(&[held.item.id.as_str().into()])?,
393 ])
394 .await?;
395 self.publish_item("work_item.reopened", &held, Some(&a.actor), Self::item_event(&held)).await?;
396 self.note_item(&held.item.id, &a.actor, "reopened this").await?;
397 Ok(match self.item_by_id(&held.item.id).await? {
398 Some(held) => Outcome::Ok(held.shown(Some(&a.actor))),
399 None => no_item(),
400 })
401 }
402}
403
404#[cfg(test)]
405mod tests {
406 use super::still_open;
407
408 #[test]
409 fn a_commit_that_does_not_close_says_where_it_has_still_to_land() {
410 assert_eq!(
411 still_open(12, "acme/web", &["acme/api".to_owned()]),
412 "Recorded the commit as landed in acme/web, but #12 stays open until it lands in acme/api too."
413 );
414 assert_eq!(
415 still_open(12, "acme/web", &["acme/api".to_owned(), String::new()]),
416 "Recorded the commit as landed in acme/web, but #12 stays open until it lands in acme/api and 1 other repository too."
417 );
418 }
419}