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