| 1 | //! Claims: an advisory, expiring "I'm working on #N". |
| 2 | //! |
| 3 | //! A claim is never a lock: anyone may claim an issue 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 issue closes. |
| 7 | |
| 8 | use g1t_contracts::identity::UsernameArgs; |
| 9 | use g1t_contracts::time::rfc3339; |
| 10 | use g1t_contracts::work::*; |
| 11 | use g1t_contracts::{FailureCode, Outcome, User, Viewer, new_id}; |
| 12 | use g1t_kit::now_ms; |
| 13 | use worker::Result; |
| 14 | use worker::wasm_bindgen::JsValue; |
| 15 | |
| 16 | use super::{CLAIM_COLUMNS, ClaimRow, after_minutes, tidy, unwrap_or_fail}; |
| 17 | use crate::Work; |
| 18 | |
| 19 | /// The most claims one list returns. |
| 20 | const LIST_LIMIT: u32 = 100; |
| 21 | |
| 22 | fn text_or_null(value: &Option<String>) -> JsValue { |
| 23 | value.as_deref().map_or(JsValue::NULL, JsValue::from) |
| 24 | } |
| 25 | |
| 26 | impl Work { |
| 27 | /// The caller's active claim on an issue, if they hold one. |
| 28 | async fn own_claim(&self, repo_id: &str, issue: u32, actor: &User, session: Option<&str>) -> Result<Option<Claim>> { |
| 29 | Ok(self |
| 30 | .active_claims(repo_id, issue) |
| 31 | .await? |
| 32 | .into_iter() |
| 33 | .find(|claim| claim_is_mine(&claim.actor.id, claim.session_id.as_deref(), &actor.id, session))) |
| 34 | } |
| 35 | |
| 36 | pub(crate) async fn claim_issue(&self, a: ClaimArgs) -> Result<Outcome<Claimed>> { |
| 37 | let repo = unwrap_or_fail!(self.participating_repo(&a.repo, &a.actor).await?); |
| 38 | let Some(issue) = self.issue(&repo.id, a.number).await? else { |
| 39 | return Ok(Outcome::fail(FailureCode::NotFound, "Issue not found.")); |
| 40 | }; |
| 41 | if issue.state == State::Closed { |
| 42 | return Ok(Outcome::fail(FailureCode::Conflict, format!("#{} is closed.", issue.number))); |
| 43 | } |
| 44 | // The session it is made from, whose heartbeats renew it. |
| 45 | let session = match a.session_id.as_deref().map(str::trim).filter(|id| !id.is_empty()) { |
| 46 | Some(id) => match self.session_row(&repo.id, id).await? { |
| 47 | Some(row) if row.actor_id == a.actor.id => Some(row.at(now_ms())), |
| 48 | _ => return Ok(Outcome::fail(FailureCode::Invalid, "session_id is not one of your sessions in this repository.")), |
| 49 | }, |
| 50 | None => None, |
| 51 | }; |
| 52 | if session.as_ref().is_some_and(|session| session.state == SessionState::Ended) { |
| 53 | return Ok(Outcome::fail(FailureCode::Invalid, "That session has ended. Start a new one.")); |
| 54 | } |
| 55 | let minutes = claim_minutes(a.ttl_minutes); |
| 56 | let note = tidy(a.note.as_deref(), MAX_NOTE_CHARS); |
| 57 | let session_id = session.as_ref().map(|session| session.id.clone()); |
| 58 | let harness = session.as_ref().map(|session| session.harness.as_str().to_owned()); |
| 59 | let now = now_ms(); |
| 60 | let expires_at = after_minutes(now, minutes); |
| 61 | let standing = self.active_claims(&repo.id, issue.number).await?; |
| 62 | let (mine, others): (Vec<Claim>, Vec<Claim>) = standing |
| 63 | .into_iter() |
| 64 | .partition(|claim| claim_is_mine(&claim.actor.id, claim.session_id.as_deref(), &a.actor.id, session_id.as_deref())); |
| 65 | let blocked_by = self.blockers(&repo.id, &issue.blocked_by).await?; |
| 66 | // Claiming what you hold renews it. |
| 67 | if let Some(held) = mine.into_iter().next() { |
| 68 | let renewed = self |
| 69 | .db |
| 70 | .prepare(format!( |
| 71 | "UPDATE claims |
| 72 | SET expires_at = ?, ttl_minutes = ?, note = COALESCE(?, note), |
| 73 | session_id = COALESCE(?, session_id), harness = COALESCE(?, harness) |
| 74 | WHERE id = ? AND state = 'active' |
| 75 | RETURNING {CLAIM_COLUMNS}" |
| 76 | )) |
| 77 | .bind(&[ |
| 78 | expires_at.as_str().into(), |
| 79 | minutes.into(), |
| 80 | text_or_null(¬e), |
| 81 | text_or_null(&session_id), |
| 82 | text_or_null(&harness), |
| 83 | held.id.as_str().into(), |
| 84 | ])? |
| 85 | .first::<ClaimRow>(None) |
| 86 | .await?; |
| 87 | if let Some(renewed) = renewed { |
| 88 | if let Some(session_id) = &session_id { |
| 89 | self.link(session_id, &repo.id, LinkKind::Issue, issue.number, LinkVia::Claim).await; |
| 90 | } |
| 91 | return Ok(Outcome::Ok(Claimed { claim: renewed.into(), others, blocked_by })); |
| 92 | } |
| 93 | } |
| 94 | // Taking over what was handed off to you, or to anyone. |
| 95 | let handed: Option<ClaimRow> = self |
| 96 | .db |
| 97 | .prepare(format!( |
| 98 | "SELECT {CLAIM_COLUMNS} FROM claims |
| 99 | WHERE repo_id = ? AND issue = ? AND state <> 'active' ORDER BY ended_at DESC, id DESC LIMIT 1" |
| 100 | )) |
| 101 | .bind(&[repo.id.as_str().into(), issue.number.into()])? |
| 102 | .first::<ClaimRow>(None) |
| 103 | .await?; |
| 104 | let took_over = handed |
| 105 | .map(Claim::from) |
| 106 | .filter(|last| last.state == ClaimState::HandedOff) |
| 107 | .filter(|last| last.handoff_to.as_deref().is_none_or(|to| to == a.actor.username)) |
| 108 | .map(|last| last.id); |
| 109 | let at = rfc3339(now); |
| 110 | let created = self |
| 111 | .db |
| 112 | .prepare(format!( |
| 113 | "INSERT INTO claims |
| 114 | (id, repo_id, issue, session_id, actor_id, actor_name, harness, note, state, ttl_minutes, |
| 115 | created_at, expires_at) |
| 116 | VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'active', ?, ?, ?) |
| 117 | RETURNING {CLAIM_COLUMNS}" |
| 118 | )) |
| 119 | .bind(&[ |
| 120 | new_id("clm", now).into(), |
| 121 | repo.id.as_str().into(), |
| 122 | issue.number.into(), |
| 123 | text_or_null(&session_id), |
| 124 | a.actor.id.as_str().into(), |
| 125 | a.actor.username.as_str().into(), |
| 126 | text_or_null(&harness), |
| 127 | text_or_null(¬e), |
| 128 | minutes.into(), |
| 129 | at.as_str().into(), |
| 130 | expires_at.as_str().into(), |
| 131 | ])? |
| 132 | .first::<ClaimRow>(None) |
| 133 | .await?; |
| 134 | let Some(created) = created else { |
| 135 | return Err(worker::Error::RustError("the claim was not recorded".into())); |
| 136 | }; |
| 137 | let claim = Claim::from(created); |
| 138 | if let Some(session_id) = &claim.session_id { |
| 139 | self.link(session_id, &repo.id, LinkKind::Issue, claim.issue, LinkVia::Claim).await; |
| 140 | } |
| 141 | self.publish( |
| 142 | "work.claimed", |
| 143 | &repo.id, |
| 144 | &a.actor, |
| 145 | ClaimEvent { others: Some(others.clone()), took_over, ..ClaimEvent::of(&claim) }, |
| 146 | ) |
| 147 | .await?; |
| 148 | Ok(Outcome::Ok(Claimed { claim, others, blocked_by })) |
| 149 | } |
| 150 | |
| 151 | /// Ends the caller's active claim on an issue as `state`, or refuses |
| 152 | /// when they hold none. |
| 153 | async fn end_own_claim( |
| 154 | &self, |
| 155 | actor: &User, |
| 156 | repo: &g1t_contracts::repos::RepoPath, |
| 157 | number: u32, |
| 158 | state: ClaimState, |
| 159 | reason: &str, |
| 160 | handoff: Option<(&Option<String>, &str)>, |
| 161 | session: Option<&str>, |
| 162 | ) -> Result<Outcome<Claim>> { |
| 163 | let repo = unwrap_or_fail!(self.participating_repo(repo, actor).await?); |
| 164 | let Some(held) = self.own_claim(&repo.id, number, actor, session.map(str::trim).filter(|id| !id.is_empty())).await? else { |
| 165 | return Ok(Outcome::fail(FailureCode::NotFound, format!("You hold no claim on #{number}."))); |
| 166 | }; |
| 167 | let at = rfc3339(now_ms()); |
| 168 | let (handoff_to, note) = match handoff { |
| 169 | Some((to, note)) => (text_or_null(to), JsValue::from(note)), |
| 170 | None => (JsValue::NULL, JsValue::NULL), |
| 171 | }; |
| 172 | let ended = self |
| 173 | .db |
| 174 | .prepare(format!( |
| 175 | "UPDATE claims |
| 176 | SET state = ?, ended_at = ?, end_reason = ?, handoff_to = ?, note = COALESCE(?, note) |
| 177 | WHERE id = ? AND state = 'active' |
| 178 | RETURNING {CLAIM_COLUMNS}" |
| 179 | )) |
| 180 | .bind(&[state.as_str().into(), at.into(), reason.into(), handoff_to, note, held.id.as_str().into()])? |
| 181 | .first::<ClaimRow>(None) |
| 182 | .await?; |
| 183 | Ok(match ended { |
| 184 | Some(row) => Outcome::Ok(row.into()), |
| 185 | None => Outcome::fail(FailureCode::NotFound, format!("You hold no claim on #{number}.")), |
| 186 | }) |
| 187 | } |
| 188 | |
| 189 | pub(crate) async fn release_claim(&self, a: ReleaseArgs) -> Result<Outcome<Claim>> { |
| 190 | let reason = tidy(a.reason.as_deref(), MAX_NOTE_CHARS).unwrap_or_else(|| "released".to_owned()); |
| 191 | let claim = unwrap_or_fail!( |
| 192 | self.end_own_claim(&a.actor, &a.repo, a.number, ClaimState::Released, &reason, None, a.session_id.as_deref()).await? |
| 193 | ); |
| 194 | self.publish( |
| 195 | "work.released", |
| 196 | &claim.repo_id.clone(), |
| 197 | &a.actor, |
| 198 | ClaimEvent { reason: Some(reason), ..ClaimEvent::of(&claim) }, |
| 199 | ) |
| 200 | .await?; |
| 201 | Ok(Outcome::Ok(claim)) |
| 202 | } |
| 203 | |
| 204 | pub(crate) async fn hand_off_issue(&self, a: HandOffArgs) -> Result<Outcome<Claim>> { |
| 205 | let Some(note) = tidy(Some(&a.note), MAX_NOTE_CHARS) else { |
| 206 | return Ok(Outcome::fail(FailureCode::Invalid, "Say where the work stands in note, for whoever picks it up.")); |
| 207 | }; |
| 208 | let to = tidy(a.to.as_deref().map(|to| to.trim().trim_start_matches('@')), 100).map(|to| to.to_lowercase()); |
| 209 | if let Some(name) = &to { |
| 210 | let account: Viewer = |
| 211 | g1t_kit::call(&self.identity, "user_by_username", &UsernameArgs { username: name.clone() }).await?; |
| 212 | if account.is_none() { |
| 213 | return Ok(Outcome::fail(FailureCode::Invalid, format!("There is no account named {name}."))); |
| 214 | } |
| 215 | } |
| 216 | let claim = unwrap_or_fail!( |
| 217 | self.end_own_claim(&a.actor, &a.repo, a.number, ClaimState::HandedOff, "handed_off", Some((&to, ¬e)), a.session_id.as_deref()) |
| 218 | .await? |
| 219 | ); |
| 220 | self.publish( |
| 221 | "work.handoff_requested", |
| 222 | &claim.repo_id.clone(), |
| 223 | &a.actor, |
| 224 | ClaimEvent { from: Some(a.actor.username.clone()), to, note: Some(note), ..ClaimEvent::of(&claim) }, |
| 225 | ) |
| 226 | .await?; |
| 227 | Ok(Outcome::Ok(claim)) |
| 228 | } |
| 229 | |
| 230 | pub(crate) async fn list_claims(&self, a: ListClaimsArgs) -> Result<Outcome<Vec<Claim>>> { |
| 231 | let repo = unwrap_or_fail!(self.coordination_repo(&a.repo, &a.viewer).await?); |
| 232 | let state = a.state.unwrap_or(ClaimState::Active); |
| 233 | let rows = self |
| 234 | .db |
| 235 | .prepare(format!( |
| 236 | "SELECT {CLAIM_COLUMNS} FROM claims |
| 237 | WHERE repo_id = ?1 AND state = ?2 AND (?3 IS NULL OR issue = ?3) |
| 238 | ORDER BY created_at DESC, id DESC LIMIT ?4" |
| 239 | )) |
| 240 | .bind(&[ |
| 241 | repo.id.as_str().into(), |
| 242 | state.as_str().into(), |
| 243 | a.issue.filter(|number| *number > 0).map_or(JsValue::NULL, JsValue::from), |
| 244 | LIST_LIMIT.into(), |
| 245 | ])? |
| 246 | .all() |
| 247 | .await? |
| 248 | .results::<ClaimRow>()?; |
| 249 | Ok(Outcome::Ok(rows.into_iter().map(Claim::from).collect())) |
| 250 | } |
| 251 | |
| 252 | /// An issue closed: its active claims end as released (`issue_closed`), |
| 253 | /// each announced. |
| 254 | pub(crate) async fn release_claims_on_close(&self, repo_id: &str, issue: u32, actor: Option<&User>) -> Result<()> { |
| 255 | let at = rfc3339(now_ms()); |
| 256 | let released = self |
| 257 | .db |
| 258 | .prepare(format!( |
| 259 | "UPDATE claims SET state = 'released', ended_at = ?, end_reason = 'issue_closed' |
| 260 | WHERE repo_id = ? AND issue = ? AND state = 'active' |
| 261 | RETURNING {CLAIM_COLUMNS}" |
| 262 | )) |
| 263 | .bind(&[at.into(), repo_id.into(), issue.into()])? |
| 264 | .all() |
| 265 | .await? |
| 266 | .results::<ClaimRow>()?; |
| 267 | for row in released { |
| 268 | let claim = Claim::from(row); |
| 269 | let data = ClaimEvent { reason: Some("issue_closed".to_owned()), ..ClaimEvent::of(&claim) }; |
| 270 | match actor { |
| 271 | Some(actor) => self.publish("work.released", repo_id, actor, data).await?, |
| 272 | None => self.publish_as("work.released", repo_id, None, data).await?, |
| 273 | } |
| 274 | } |
| 275 | Ok(()) |
| 276 | } |
| 277 | } |