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