| 1 | //! Claims: an advisory, expiring "I'm working on #N". |
| 2 | //! |
| 3 | //! A claim is never a lock: anyone may claim a work item someone else has |
| 4 | //! claimed, and is told who else is on it. It lapses unless renewed, by |
| 5 | //! claiming again or by a heartbeat of the session it was made from, and |
| 6 | //! ends when released, handed off, or when its work item closes. It is on |
| 7 | //! a work item of a workspace, and made from a checkout of one of the |
| 8 | //! workspace's repositories when it names one. |
| 9 | |
| 10 | use g1t_contracts::access::Capability; |
| 11 | use g1t_contracts::identity::UsernameArgs; |
| 12 | use g1t_contracts::repos::{Repo, RepoPath}; |
| 13 | use g1t_contracts::time::rfc3339; |
| 14 | use g1t_contracts::work::*; |
| 15 | use g1t_contracts::{FailureCode, Outcome, User, Viewer, new_id}; |
| 16 | use g1t_kit::now_ms; |
| 17 | use worker::Result; |
| 18 | use worker::wasm_bindgen::JsValue; |
| 19 | |
| 20 | use g1t_contracts::credentials::Principal; |
| 21 | |
| 22 | use super::updates::push_delivery; |
| 23 | use super::{CLAIM_COLUMNS, ClaimRow, SessionRow, after_minutes, tidy, unwrap_or_fail}; |
| 24 | use crate::Work; |
| 25 | use crate::items::{Held, workspace_of}; |
| 26 | |
| 27 | /// The most claims one list returns. |
| 28 | const LIST_LIMIT: u32 = 100; |
| 29 | |
| 30 | /// Resolve ownership and insert/renew in one SQLite statement. In particular, |
| 31 | /// two requests that both read no claim must converge on the same row. |
| 32 | /// Sessionless claims retain the caller-wide semantics of claim_is_mine. |
| 33 | /// `?2` is the work item, `?3` the repository it is claimed from. |
| 34 | pub(crate) fn upsert_claim_sql() -> String { |
| 35 | format!( |
| 36 | "INSERT INTO claims |
| 37 | (id, item_id, repo_id, session_id, actor_id, actor_name, harness, note, state, ttl_minutes, |
| 38 | created_at, expires_at) |
| 39 | VALUES (COALESCE((SELECT id FROM claims |
| 40 | WHERE item_id = ?2 AND actor_id = ?5 AND state = 'active' |
| 41 | AND (session_id IS NULL OR ?4 IS NULL OR session_id = ?4) |
| 42 | ORDER BY created_at, id LIMIT 1), ?1), |
| 43 | ?2, ?3, ?4, ?5, ?6, ?7, ?8, 'active', ?9, ?10, ?11) |
| 44 | ON CONFLICT(id) DO UPDATE SET |
| 45 | expires_at = MAX(claims.expires_at, excluded.expires_at), ttl_minutes = excluded.ttl_minutes, |
| 46 | note = COALESCE(excluded.note, claims.note), |
| 47 | session_id = COALESCE(excluded.session_id, claims.session_id), |
| 48 | repo_id = COALESCE(excluded.repo_id, claims.repo_id), |
| 49 | harness = COALESCE(excluded.harness, claims.harness) |
| 50 | WHERE claims.state = 'active' |
| 51 | RETURNING {CLAIM_COLUMNS}" |
| 52 | ) |
| 53 | } |
| 54 | |
| 55 | fn text_or_null(value: &Option<String>) -> JsValue { |
| 56 | value.as_deref().map_or(JsValue::NULL, JsValue::from) |
| 57 | } |
| 58 | |
| 59 | /// The repositories a claim's updates go to: those its work item is linked |
| 60 | /// to, and the one it was made from. |
| 61 | fn claim_repos(held: &Held, repo_id: Option<&str>) -> Vec<String> { |
| 62 | let mut repos = held.repo_ids(); |
| 63 | if let Some(repo_id) = repo_id |
| 64 | && !repos.iter().any(|known| known == repo_id) |
| 65 | { |
| 66 | repos.push(repo_id.to_owned()); |
| 67 | } |
| 68 | repos |
| 69 | } |
| 70 | |
| 71 | impl Work { |
| 72 | /// The caller's active claim on a work item, if they hold one. |
| 73 | async fn own_claim(&self, item_id: &str, actor: &User, session: Option<&str>) -> Result<Option<Claim>> { |
| 74 | Ok(self |
| 75 | .active_claims(item_id) |
| 76 | .await? |
| 77 | .into_iter() |
| 78 | .find(|claim| claim_is_mine(&claim.actor.id, claim.session_id.as_deref(), &actor.id, session))) |
| 79 | } |
| 80 | |
| 81 | /// One of `actor`'s sessions, in `repo` when one is named, else in any |
| 82 | /// repository. |
| 83 | async fn actor_session(&self, repo: Option<&Repo>, actor: &User, id: &str) -> Result<Option<SessionRow>> { |
| 84 | let row = match repo { |
| 85 | Some(repo) => self.session_row(&repo.id, id).await?, |
| 86 | None => self |
| 87 | .db |
| 88 | .prepare(format!("SELECT {} FROM sessions WHERE id = ?", super::SESSION_COLUMNS)) |
| 89 | .bind(&[id.into()])? |
| 90 | .first::<SessionRow>(None) |
| 91 | .await?, |
| 92 | }; |
| 93 | Ok(row.filter(|row| row.actor_id == actor.id)) |
| 94 | } |
| 95 | |
| 96 | /// The repository a claim is made from, if `actor` may take part in it. |
| 97 | async fn claiming_repo(&self, repo: Option<&RepoPath>, actor: &User) -> Result<Outcome<Option<Repo>>> { |
| 98 | Ok(match repo { |
| 99 | Some(path) => Outcome::Ok(Some(unwrap_or_fail!(self.participating_repo(path, actor).await?))), |
| 100 | None => { |
| 101 | if !actor.verified { |
| 102 | return Ok(Outcome::fail(FailureCode::Forbidden, crate::UNVERIFIED)); |
| 103 | } |
| 104 | Outcome::Ok(None) |
| 105 | } |
| 106 | }) |
| 107 | } |
| 108 | |
| 109 | /// `claim_work_item`: the work item named, or with a `title` and no |
| 110 | /// number, a new one opened for it, with the claim's session attached |
| 111 | /// to it. |
| 112 | pub(crate) async fn claim_work_item(&self, a: ClaimArgs) -> Result<Outcome<Claimed>> { |
| 113 | if a.number == 0 { |
| 114 | return self.open_and_claim(a).await; |
| 115 | } |
| 116 | self.claim_number(a).await |
| 117 | } |
| 118 | |
| 119 | /// Opens a work item from `title` and `body`, filed from `repo` when |
| 120 | /// given, claims it, and attaches the claim's session to it. The |
| 121 | /// session is checked before anything is opened, so a claim that would |
| 122 | /// be refused opens nothing. |
| 123 | async fn open_and_claim(&self, a: ClaimArgs) -> Result<Outcome<Claimed>> { |
| 124 | let Some(title) = a.title.as_deref().map(str::trim).filter(|title| !title.is_empty()).map(str::to_owned) else { |
| 125 | return Ok(Outcome::fail(FailureCode::Invalid, "Name the work item by its number, or give a title (and body) to open one.")); |
| 126 | }; |
| 127 | let repo = unwrap_or_fail!(self.claiming_repo(a.repo.as_ref(), &a.actor).await?); |
| 128 | let session_id = a.session_id.as_deref().map(str::trim).filter(|id| !id.is_empty()).map(str::to_owned); |
| 129 | let mut session_repo = None; |
| 130 | if let Some(id) = &session_id { |
| 131 | match self.actor_session(repo.as_ref(), &a.actor, id).await? { |
| 132 | Some(row) => { |
| 133 | if row.clone().at(now_ms()).state == SessionState::Ended { |
| 134 | return Ok(Outcome::fail(FailureCode::Invalid, "That session has ended. Start a new one.")); |
| 135 | } |
| 136 | session_repo = Some(row.repo_id.clone()); |
| 137 | } |
| 138 | None => return Ok(Outcome::fail(FailureCode::Invalid, "session_id is not one of your sessions in this repository.")), |
| 139 | } |
| 140 | } |
| 141 | let opened = unwrap_or_fail!( |
| 142 | self.open_work_item(OpenWorkItemArgs { |
| 143 | actor: a.actor.clone(), |
| 144 | workspace: workspace_of(&a.workspace, a.repo.as_ref()), |
| 145 | repo: a.repo.clone(), |
| 146 | targets: Vec::new(), |
| 147 | title, |
| 148 | body: a.body.clone().unwrap_or_default(), |
| 149 | labels: Vec::new(), |
| 150 | milestone: None, |
| 151 | }) |
| 152 | .await? |
| 153 | ); |
| 154 | let claimed = unwrap_or_fail!( |
| 155 | self.claim_number(ClaimArgs { |
| 156 | workspace: opened.workspace.clone(), |
| 157 | number: opened.number, |
| 158 | title: None, |
| 159 | body: None, |
| 160 | session_id: session_id.clone(), |
| 161 | ..a.clone() |
| 162 | }) |
| 163 | .await? |
| 164 | ); |
| 165 | // The session is on the work item now, not only tied to it by the claim. |
| 166 | if let (Some(id), Some(session_repo)) = (session_id, session_repo) { |
| 167 | let path: Option<RepoPath> = g1t_kit::call(&self.repos, "path_by_id", &g1t_contracts::repos::PathByIdArgs { id: session_repo }).await?; |
| 168 | // A session in another workspace's repository names none of |
| 169 | // this one's work items. |
| 170 | if let Some(path) = path.filter(|path| path.namespace.eq_ignore_ascii_case(&opened.workspace)) { |
| 171 | let attached = self |
| 172 | .update_session(UpdateSessionArgs { |
| 173 | actor: a.actor.clone(), |
| 174 | repo: path, |
| 175 | id, |
| 176 | item: Some(opened.number), |
| 177 | pull: None, |
| 178 | branch: None, |
| 179 | head_commit: None, |
| 180 | title: None, |
| 181 | agent_type: None, |
| 182 | task: None, |
| 183 | prompt: None, |
| 184 | model: None, |
| 185 | result: None, |
| 186 | }) |
| 187 | .await?; |
| 188 | if let Outcome::Fail(failure) = attached { |
| 189 | worker::console_error!("#{}: the session was not attached: {}", opened.number, failure.message); |
| 190 | } |
| 191 | } |
| 192 | } |
| 193 | Ok(Outcome::Ok(Claimed { item: Some(opened), ..claimed })) |
| 194 | } |
| 195 | |
| 196 | async fn claim_number(&self, a: ClaimArgs) -> Result<Outcome<Claimed>> { |
| 197 | let repo = unwrap_or_fail!(self.claiming_repo(a.repo.as_ref(), &a.actor).await?); |
| 198 | let workspace = workspace_of(&a.workspace, a.repo.as_ref()); |
| 199 | let held = unwrap_or_fail!(self.item_for(&Some(a.actor.clone()), &workspace, a.number, Capability::Participate).await?); |
| 200 | if held.item.state == State::Closed { |
| 201 | return Ok(Outcome::fail(FailureCode::Conflict, format!("#{} is closed.", held.item.number))); |
| 202 | } |
| 203 | // The session it is made from, whose heartbeats renew it. |
| 204 | let session = match a.session_id.as_deref().map(str::trim).filter(|id| !id.is_empty()) { |
| 205 | Some(id) => match self.actor_session(repo.as_ref(), &a.actor, id).await? { |
| 206 | Some(row) => Some(row.at(now_ms())), |
| 207 | None => return Ok(Outcome::fail(FailureCode::Invalid, "session_id is not one of your sessions in this repository.")), |
| 208 | }, |
| 209 | None => None, |
| 210 | }; |
| 211 | if session.as_ref().is_some_and(|session| session.state == SessionState::Ended) { |
| 212 | return Ok(Outcome::fail(FailureCode::Invalid, "That session has ended. Start a new one.")); |
| 213 | } |
| 214 | let repo_id = repo.as_ref().map(|repo| repo.id.clone()).or_else(|| session.as_ref().map(|session| session.repo_id.clone())); |
| 215 | let minutes = claim_minutes(a.ttl_minutes); |
| 216 | let note = tidy(a.note.as_deref(), MAX_NOTE_CHARS); |
| 217 | let session_id = session.as_ref().map(|session| session.id.clone()); |
| 218 | let harness = session.as_ref().map(|session| session.harness.as_str().to_owned()); |
| 219 | let now = now_ms(); |
| 220 | let expires_at = after_minutes(now, minutes); |
| 221 | let standing = self.active_claims(&held.item.id).await?; |
| 222 | let (mine, others): (Vec<Claim>, Vec<Claim>) = standing |
| 223 | .into_iter() |
| 224 | .partition(|claim| claim_is_mine(&claim.actor.id, claim.session_id.as_deref(), &a.actor.id, session_id.as_deref())); |
| 225 | let blocked_by = self.blockers(&held.item.workspace, &held.item.blocked_by).await?; |
| 226 | let repos = claim_repos(&held, repo_id.as_deref()); |
| 227 | // Claiming what you hold renews it. |
| 228 | if let Some(held_claim) = mine.into_iter().next() { |
| 229 | let renewed = self |
| 230 | .db |
| 231 | .prepare(format!( |
| 232 | "UPDATE claims |
| 233 | SET expires_at = ?, ttl_minutes = ?, note = COALESCE(?, note), |
| 234 | session_id = COALESCE(?, session_id), harness = COALESCE(?, harness), |
| 235 | repo_id = COALESCE(?, repo_id) |
| 236 | WHERE id = ? AND state = 'active' |
| 237 | RETURNING {CLAIM_COLUMNS}" |
| 238 | )) |
| 239 | .bind(&[ |
| 240 | expires_at.as_str().into(), |
| 241 | minutes.into(), |
| 242 | text_or_null(¬e), |
| 243 | text_or_null(&session_id), |
| 244 | text_or_null(&harness), |
| 245 | text_or_null(&repo_id), |
| 246 | held_claim.id.as_str().into(), |
| 247 | ])? |
| 248 | .first::<ClaimRow>(None) |
| 249 | .await?; |
| 250 | if let Some(renewed) = renewed { |
| 251 | if let (Some(session_id), Some(repo_id)) = (&session_id, &repo_id) { |
| 252 | self.link_item(session_id, repo_id, &held.item.id, LinkVia::Claim).await; |
| 253 | } |
| 254 | return Ok(Outcome::Ok(Claimed { claim: renewed.into(), others, blocked_by, item: None })); |
| 255 | } |
| 256 | } |
| 257 | // Taking over what was handed off to you, or to anyone. |
| 258 | let handed: Option<ClaimRow> = self |
| 259 | .db |
| 260 | .prepare(format!( |
| 261 | "SELECT {CLAIM_COLUMNS} FROM claims |
| 262 | WHERE item_id = ? AND state <> 'active' ORDER BY ended_at DESC, id DESC LIMIT 1" |
| 263 | )) |
| 264 | .bind(&[held.item.id.as_str().into()])? |
| 265 | .first::<ClaimRow>(None) |
| 266 | .await?; |
| 267 | let took_over = handed |
| 268 | .map(Claim::from) |
| 269 | .filter(|last| last.state == ClaimState::HandedOff) |
| 270 | .filter(|last| last.handoff_to.as_deref().is_none_or(|to| to == a.actor.username)) |
| 271 | .map(|last| last.id); |
| 272 | let at = rfc3339(now); |
| 273 | let proposed_id = new_id("clm", now); |
| 274 | // A new claim is announced with itself; a renewal (the claim found |
| 275 | // is an older one of the caller's) is not. |
| 276 | let proposed = Claim { |
| 277 | id: proposed_id.clone(), |
| 278 | repo_id: repo_id.clone(), |
| 279 | workspace: held.item.workspace.clone(), |
| 280 | item: held.item.number, |
| 281 | session_id: session_id.clone(), |
| 282 | actor: Principal { id: a.actor.id.clone(), username: a.actor.username.clone() }, |
| 283 | harness: harness.as_deref().and_then(Harness::parse), |
| 284 | note: note.clone(), |
| 285 | state: ClaimState::Active, |
| 286 | created_at: at.clone(), |
| 287 | expires_at: expires_at.clone(), |
| 288 | ended_at: None, |
| 289 | end_reason: None, |
| 290 | handoff_to: None, |
| 291 | }; |
| 292 | let event = |claim: &Claim| { |
| 293 | g1t_contracts::events::marked( |
| 294 | ClaimEvent { others: Some(others.clone()), took_over: took_over.clone(), ..ClaimEvent::of(claim, &repos) }, |
| 295 | Some(&a.actor), |
| 296 | ) |
| 297 | }; |
| 298 | let mut statements = vec![self |
| 299 | .db |
| 300 | .prepare(upsert_claim_sql()) |
| 301 | .bind(&[ |
| 302 | proposed_id.as_str().into(), |
| 303 | held.item.id.as_str().into(), |
| 304 | text_or_null(&repo_id), |
| 305 | text_or_null(&session_id), |
| 306 | a.actor.id.as_str().into(), |
| 307 | a.actor.username.as_str().into(), |
| 308 | text_or_null(&harness), |
| 309 | text_or_null(¬e), |
| 310 | minutes.into(), |
| 311 | at.as_str().into(), |
| 312 | expires_at.as_str().into(), |
| 313 | ])?]; |
| 314 | let sent = push_delivery( |
| 315 | &mut statements, |
| 316 | self.delivery_to( |
| 317 | "work.claimed", |
| 318 | &repos, |
| 319 | event(&proposed), |
| 320 | &format!("{proposed_id}:claimed"), |
| 321 | "EXISTS (SELECT 1 FROM claims WHERE id = ?7)", |
| 322 | vec![proposed_id.as_str().into()], |
| 323 | )?, |
| 324 | ); |
| 325 | let created = match self.db.batch(statements).await?.into_iter().next() { |
| 326 | Some(result) => result.results::<ClaimRow>()?.into_iter().next(), |
| 327 | None => None, |
| 328 | }; |
| 329 | let Some(created) = created else { |
| 330 | return Err(worker::Error::RustError("the claim was not recorded".into())); |
| 331 | }; |
| 332 | let claim = Claim::from(created); |
| 333 | if let (Some(session_id), Some(repo_id)) = (&claim.session_id, &claim.repo_id) { |
| 334 | self.link_item(session_id, repo_id, &held.item.id, LinkVia::Claim).await; |
| 335 | } |
| 336 | if claim.id != proposed_id { |
| 337 | return Ok(Outcome::Ok(Claimed { claim, others, blocked_by, item: None })); |
| 338 | } |
| 339 | self.delivered(sent, "work.claimed", held.event_repo().unwrap_or_default(), Some(a.actor.id.clone()), event(&claim)).await?; |
| 340 | Ok(Outcome::Ok(Claimed { claim, others, blocked_by, item: None })) |
| 341 | } |
| 342 | |
| 343 | /// Ends the caller's active claim on a work item as `state`, announced |
| 344 | /// as `kind` with the event `event` makes of it, or refuses when they |
| 345 | /// hold none. |
| 346 | #[allow(clippy::too_many_arguments)] |
| 347 | async fn end_own_claim( |
| 348 | &self, |
| 349 | actor: &User, |
| 350 | workspace: &str, |
| 351 | repo: Option<&RepoPath>, |
| 352 | number: u32, |
| 353 | state: ClaimState, |
| 354 | reason: &str, |
| 355 | handoff: Option<(&Option<String>, &str)>, |
| 356 | session: Option<&str>, |
| 357 | kind: &'static str, |
| 358 | event: impl Fn(&Claim, &[String]) -> ClaimEvent, |
| 359 | ) -> Result<Outcome<Claim>> { |
| 360 | unwrap_or_fail!(self.claiming_repo(repo, actor).await?); |
| 361 | let workspace = workspace_of(workspace, repo); |
| 362 | let held = unwrap_or_fail!(self.item_for(&Some(actor.clone()), &workspace, number, Capability::Read).await?); |
| 363 | let Some(claim) = self.own_claim(&held.item.id, actor, session.map(str::trim).filter(|id| !id.is_empty())).await? else { |
| 364 | return Ok(Outcome::fail(FailureCode::NotFound, format!("You hold no claim on #{number}."))); |
| 365 | }; |
| 366 | let repos = claim_repos(&held, claim.repo_id.as_deref()); |
| 367 | let at = rfc3339(now_ms()); |
| 368 | let expected = Claim { |
| 369 | state, |
| 370 | ended_at: Some(at.clone()), |
| 371 | end_reason: Some(reason.to_owned()), |
| 372 | handoff_to: handoff.and_then(|(to, _)| to.clone()), |
| 373 | note: handoff.map(|(_, note)| note.to_owned()).or(claim.note.clone()), |
| 374 | ..claim.clone() |
| 375 | }; |
| 376 | let (handoff_to, note) = match handoff { |
| 377 | Some((to, note)) => (text_or_null(to), JsValue::from(note)), |
| 378 | None => (JsValue::NULL, JsValue::NULL), |
| 379 | }; |
| 380 | let mut statements = vec![self |
| 381 | .db |
| 382 | .prepare(format!( |
| 383 | "UPDATE claims |
| 384 | SET state = ?, ended_at = ?, end_reason = ?, handoff_to = ?, note = COALESCE(?, note) |
| 385 | WHERE id = ? AND state = 'active' |
| 386 | RETURNING {CLAIM_COLUMNS}" |
| 387 | )) |
| 388 | .bind(&[state.as_str().into(), at.as_str().into(), reason.into(), handoff_to, note, claim.id.as_str().into()])?]; |
| 389 | // A claim ends once: its id and how it ended name the change. |
| 390 | let sent = push_delivery( |
| 391 | &mut statements, |
| 392 | self.delivery_to( |
| 393 | kind, |
| 394 | &repos, |
| 395 | g1t_contracts::events::marked(event(&expected, &repos), Some(actor)), |
| 396 | &format!("{}:{}", claim.id, state.as_str()), |
| 397 | "EXISTS (SELECT 1 FROM claims WHERE id = ?7 AND state = ?8)", |
| 398 | vec![claim.id.as_str().into(), state.as_str().into()], |
| 399 | )?, |
| 400 | ); |
| 401 | let ended = match self.db.batch(statements).await?.into_iter().next() { |
| 402 | Some(result) => result.results::<ClaimRow>()?.into_iter().next(), |
| 403 | None => None, |
| 404 | }; |
| 405 | let Some(row) = ended else { |
| 406 | return Ok(Outcome::fail(FailureCode::NotFound, format!("You hold no claim on #{number}."))); |
| 407 | }; |
| 408 | let claim = Claim::from(row); |
| 409 | let data = g1t_contracts::events::marked(event(&claim, &repos), Some(actor)); |
| 410 | self.delivered(sent, kind, held.event_repo().unwrap_or_default(), Some(actor.id.clone()), data).await?; |
| 411 | Ok(Outcome::Ok(claim)) |
| 412 | } |
| 413 | |
| 414 | pub(crate) async fn release_claim(&self, a: ReleaseArgs) -> Result<Outcome<Claim>> { |
| 415 | let reason = tidy(a.reason.as_deref(), MAX_NOTE_CHARS).unwrap_or_else(|| "released".to_owned()); |
| 416 | let claim = unwrap_or_fail!( |
| 417 | self.end_own_claim( |
| 418 | &a.actor, |
| 419 | &a.workspace, |
| 420 | a.repo.as_ref(), |
| 421 | a.number, |
| 422 | ClaimState::Released, |
| 423 | &reason, |
| 424 | None, |
| 425 | a.session_id.as_deref(), |
| 426 | "work.released", |
| 427 | |claim, repos| ClaimEvent { reason: Some(reason.clone()), ..ClaimEvent::of(claim, repos) }, |
| 428 | ) |
| 429 | .await? |
| 430 | ); |
| 431 | Ok(Outcome::Ok(claim)) |
| 432 | } |
| 433 | |
| 434 | pub(crate) async fn hand_off_work_item(&self, a: HandOffArgs) -> Result<Outcome<Claim>> { |
| 435 | let Some(note) = tidy(Some(&a.note), MAX_NOTE_CHARS) else { |
| 436 | return Ok(Outcome::fail(FailureCode::Invalid, "Say where the work stands in note, for whoever picks it up.")); |
| 437 | }; |
| 438 | let to = tidy(a.to.as_deref().map(|to| to.trim().trim_start_matches('@')), 100).map(|to| to.to_lowercase()); |
| 439 | if let Some(name) = &to { |
| 440 | let account: Viewer = |
| 441 | g1t_kit::call(&self.identity, "user_by_username", &UsernameArgs { username: name.clone() }).await?; |
| 442 | if account.is_none() { |
| 443 | return Ok(Outcome::fail(FailureCode::Invalid, format!("There is no account named {name}."))); |
| 444 | } |
| 445 | } |
| 446 | let claim = unwrap_or_fail!( |
| 447 | self.end_own_claim( |
| 448 | &a.actor, |
| 449 | &a.workspace, |
| 450 | a.repo.as_ref(), |
| 451 | a.number, |
| 452 | ClaimState::HandedOff, |
| 453 | "handed_off", |
| 454 | Some((&to, ¬e)), |
| 455 | a.session_id.as_deref(), |
| 456 | "work.handoff_requested", |
| 457 | |claim, repos| ClaimEvent { |
| 458 | from: Some(a.actor.username.clone()), |
| 459 | to: to.clone(), |
| 460 | note: Some(note.clone()), |
| 461 | ..ClaimEvent::of(claim, repos) |
| 462 | }, |
| 463 | ) |
| 464 | .await? |
| 465 | ); |
| 466 | Ok(Outcome::Ok(claim)) |
| 467 | } |
| 468 | |
| 469 | /// A workspace's claims on the work items the viewer may see; with a |
| 470 | /// repository, those made from it or on work items linked to it. |
| 471 | pub(crate) async fn list_claims(&self, a: ListClaimsArgs) -> Result<Outcome<Vec<Claim>>> { |
| 472 | let workspace = workspace_of(&a.workspace, a.repo.as_ref()); |
| 473 | let repo = match &a.repo { |
| 474 | Some(path) => Some(unwrap_or_fail!(self.coordination_repo(path, &a.viewer).await?)), |
| 475 | None => None, |
| 476 | }; |
| 477 | let state = a.state.unwrap_or(ClaimState::Active); |
| 478 | let rows = self |
| 479 | .db |
| 480 | .prepare(format!( |
| 481 | "SELECT {CLAIM_COLUMNS} FROM claims |
| 482 | WHERE state = ?2 |
| 483 | AND item_id IN (SELECT id FROM work_items WHERE workspace = ?1 AND (?3 IS NULL OR number = ?3)) |
| 484 | AND (?4 IS NULL OR repo_id = ?4 OR item_id IN (SELECT item_id FROM work_item_repos WHERE repo_id = ?4)) |
| 485 | ORDER BY created_at DESC, id DESC LIMIT ?5" |
| 486 | )) |
| 487 | .bind(&[ |
| 488 | workspace.as_str().into(), |
| 489 | state.as_str().into(), |
| 490 | a.item.filter(|number| *number > 0).map_or(JsValue::NULL, JsValue::from), |
| 491 | text_or_null(&repo.as_ref().map(|repo| repo.id.clone())), |
| 492 | LIST_LIMIT.into(), |
| 493 | ])? |
| 494 | .all() |
| 495 | .await? |
| 496 | .results::<ClaimRow>()?; |
| 497 | let mut ids: Vec<String> = rows.iter().map(|row| row.item_id.clone()).collect(); |
| 498 | ids.sort(); |
| 499 | ids.dedup(); |
| 500 | let seen: Vec<String> = self |
| 501 | .items_by_ids(&ids) |
| 502 | .await? |
| 503 | .into_iter() |
| 504 | .filter(|held| held.may(a.viewer.as_ref(), Capability::Read)) |
| 505 | .map(|held| held.item.id) |
| 506 | .collect(); |
| 507 | Ok(Outcome::Ok(rows.into_iter().filter(|row| seen.contains(&row.item_id)).map(Claim::from).collect())) |
| 508 | } |
| 509 | |
| 510 | /// A work item closed: its active claims end as released |
| 511 | /// (`item_closed`), each announced. |
| 512 | pub(crate) async fn release_claims_on_close(&self, held: &Held, actor: Option<&User>) -> Result<()> { |
| 513 | let at = rfc3339(now_ms()); |
| 514 | let active = self.active_claims(&held.item.id).await?; |
| 515 | if active.is_empty() { |
| 516 | return Ok(()); |
| 517 | } |
| 518 | let event = |claim: &Claim| { |
| 519 | let repos = claim_repos(held, claim.repo_id.as_deref()); |
| 520 | g1t_contracts::events::marked(ClaimEvent { reason: Some("item_closed".to_owned()), ..ClaimEvent::of(claim, &repos) }, actor) |
| 521 | }; |
| 522 | // Each claim's release and its updates are one change, in one batch. |
| 523 | let mut statements = Vec::new(); |
| 524 | let mut sent = Vec::new(); |
| 525 | for claim in &active { |
| 526 | statements.push( |
| 527 | self.db |
| 528 | .prepare(format!( |
| 529 | "UPDATE claims SET state = 'released', ended_at = ?, end_reason = 'item_closed' |
| 530 | WHERE id = ? AND state = 'active' |
| 531 | RETURNING {CLAIM_COLUMNS}" |
| 532 | )) |
| 533 | .bind(&[at.as_str().into(), claim.id.as_str().into()])?, |
| 534 | ); |
| 535 | let expected = Claim { |
| 536 | state: ClaimState::Released, |
| 537 | ended_at: Some(at.clone()), |
| 538 | end_reason: Some("item_closed".to_owned()), |
| 539 | ..claim.clone() |
| 540 | }; |
| 541 | let delivery = self.delivery_to( |
| 542 | "work.released", |
| 543 | &claim_repos(held, claim.repo_id.as_deref()), |
| 544 | event(&expected), |
| 545 | &format!("{}:{}", claim.id, ClaimState::Released.as_str()), |
| 546 | "EXISTS (SELECT 1 FROM claims WHERE id = ?7 AND state = 'released')", |
| 547 | vec![claim.id.as_str().into()], |
| 548 | )?; |
| 549 | let index = statements.len() - 1; |
| 550 | sent.push((index, push_delivery(&mut statements, delivery))); |
| 551 | } |
| 552 | let results = self.db.batch(statements).await?; |
| 553 | for (index, sent) in sent { |
| 554 | let Some(row) = results.get(index).map(|result| result.results::<ClaimRow>()).transpose()?.and_then(|rows| rows.into_iter().next()) |
| 555 | else { |
| 556 | continue; |
| 557 | }; |
| 558 | let claim = Claim::from(row); |
| 559 | let repo = claim.repo_id.clone().or_else(|| held.event_repo().map(str::to_owned)).unwrap_or_default(); |
| 560 | self.delivered(sent, "work.released", &repo, actor.map(|actor| actor.id.clone()), event(&claim)).await?; |
| 561 | } |
| 562 | Ok(()) |
| 563 | } |
| 564 | } |