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
613 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 })
217 .await?;
218 if let Outcome::Fail(failure) = attached {
219 worker::console_error!("#{}: the session was not attached: {}", opened.number, failure.message);
220 }
221 }
222 }
223 Ok(Outcome::Ok(Claimed { item: Some(opened), ..claimed }))
224 }
225
226 async fn claim_number(&self, a: ClaimArgs) -> Result<Outcome<Claimed>> {
227 let repo = unwrap_or_fail!(self.claiming_repo(a.repo.as_ref(), &a.actor).await?);
228 let workspace = workspace_of(&a.workspace, a.repo.as_ref());
229 let held = unwrap_or_fail!(items::for_viewer(self, &Some(a.actor.clone()), &workspace, a.number, Capability::Participate).await?);
230 if held.item.state == State::Closed {
231 return Ok(Outcome::fail(FailureCode::Conflict, format!("#{} is closed.", held.item.number)));
232 }
233 // The session it is made from, whose heartbeats renew it.
234 let session = match a.session_id.as_deref().map(str::trim).filter(|id| !id.is_empty()) {
235 Some(id) => match self.actor_session(repo.as_ref(), &a.actor, id).await? {
236 Some(row) => Some(row.at(now_ms())),
237 None => return Ok(Outcome::fail(FailureCode::Invalid, "session_id is not one of your sessions in this repository.")),
238 },
239 None => None,
240 };
241 if session.as_ref().is_some_and(|session| session.state == SessionState::Ended) {
242 return Ok(Outcome::fail(FailureCode::Invalid, "That session has ended. Start a new one."));
243 }
244 // Only a repository, or a session in one, of the work item's own
245 // workspace is tied to it.
246 let session_workspace = match &session {
247 Some(session) => Some(items::repo_workspace(self, &session.repo_id).await?),
248 None => None,
249 };
250 if let Some(refusal) = outside_workspace(
251 &held.item.workspace,
252 repo.as_ref().map(|repo| repo.namespace.as_str()),
253 session_workspace.as_ref().map(Option::as_deref),
254 ) {
255 return Ok(Outcome::fail(FailureCode::Invalid, refusal));
256 }
257 let repo_id = repo.as_ref().map(|repo| repo.id.clone()).or_else(|| session.as_ref().map(|session| session.repo_id.clone()));
258 let minutes = claim_minutes(a.ttl_minutes);
259 let note = tidy(a.note.as_deref(), MAX_NOTE_CHARS);
260 let session_id = session.as_ref().map(|session| session.id.clone());
261 let harness = session.as_ref().map(|session| session.harness.as_str().to_owned());
262 let now = now_ms();
263 let expires_at = after_minutes(now, minutes);
264 let standing = self.active_claims(&held.item.id).await?;
265 let (mine, others): (Vec<Claim>, Vec<Claim>) = standing
266 .into_iter()
267 .partition(|claim| claim_is_mine(&claim.actor.id, claim.session_id.as_deref(), &a.actor.id, session_id.as_deref()));
268 // The others, as far as the actor may read where they are.
269 let claimed_in: Vec<String> = others.iter().filter_map(|claim| claim.repo_id.clone()).collect();
270 let readable = items::readable_repo_ids(self, Some(&a.actor), &claimed_in).await?;
271 let others: Vec<Claim> =
272 others.into_iter().filter(|claim| claim.repo_id.as_ref().is_none_or(|id| readable.contains(id))).collect();
273 let blocked_by = self.blockers(Some(&a.actor), &held.item.workspace, &held.item.blocked_by).await?;
274 let repos = claim_repos(&held, repo_id.as_deref());
275 // Claiming what you hold renews it.
276 if let Some(held_claim) = mine.into_iter().next() {
277 let renewed = self
278 .db
279 .prepare(format!(
280 "UPDATE claims
281 SET expires_at = ?, ttl_minutes = ?, note = COALESCE(?, note),
282 session_id = COALESCE(?, session_id), harness = COALESCE(?, harness),
283 repo_id = COALESCE(?, repo_id)
284 WHERE id = ? AND state = 'active'
285 RETURNING {CLAIM_COLUMNS}"
286 ))
287 .bind(&[
288 expires_at.as_str().into(),
289 minutes.into(),
290 text_or_null(&note),
291 text_or_null(&session_id),
292 text_or_null(&harness),
293 text_or_null(&repo_id),
294 held_claim.id.as_str().into(),
295 ])?
296 .first::<ClaimRow>(None)
297 .await?;
298 if let Some(renewed) = renewed {
299 if let (Some(session_id), Some(repo_id)) = (&session_id, &repo_id) {
300 self.link_item(session_id, repo_id, &held.item.id, LinkVia::Claim).await;
301 }
302 return Ok(Outcome::Ok(Claimed { claim: renewed.into(), others, blocked_by, item: None }));
303 }
304 }
305 // Taking over what was handed off to you, or to anyone.
306 let handed: Option<ClaimRow> = self
307 .db
308 .prepare(format!(
309 "SELECT {CLAIM_COLUMNS} FROM claims
310 WHERE item_id = ? AND state <> 'active' ORDER BY ended_at DESC, id DESC LIMIT 1"
311 ))
312 .bind(&[held.item.id.as_str().into()])?
313 .first::<ClaimRow>(None)
314 .await?;
315 let took_over = handed
316 .map(Claim::from)
317 .filter(|last| last.state == ClaimState::HandedOff)
318 .filter(|last| last.handoff_to.as_deref().is_none_or(|to| to == a.actor.username))
319 .map(|last| last.id);
320 let at = rfc3339(now);
321 let proposed_id = new_id("clm", now);
322 // A new claim is announced with itself; a renewal (the claim found
323 // is an older one of the caller's) is not.
324 let proposed = Claim {
325 id: proposed_id.clone(),
326 repo_id: repo_id.clone(),
327 workspace: held.item.workspace.clone(),
328 item: held.item.number,
329 session_id: session_id.clone(),
330 actor: Principal { id: a.actor.id.clone(), username: a.actor.username.clone() },
331 harness: harness.as_deref().and_then(Harness::parse),
332 note: note.clone(),
333 state: ClaimState::Active,
334 created_at: at.clone(),
335 expires_at: expires_at.clone(),
336 ended_at: None,
337 end_reason: None,
338 handoff_to: None,
339 };
340 let event = |claim: &Claim| {
341 g1t_contracts::events::marked(
342 ClaimEvent { others: Some(others.clone()), took_over: took_over.clone(), ..ClaimEvent::of(claim, &repos) },
343 Some(&a.actor),
344 )
345 };
346 let mut statements = vec![self
347 .db
348 .prepare(upsert_claim_sql())
349 .bind(&[
350 proposed_id.as_str().into(),
351 held.item.id.as_str().into(),
352 text_or_null(&repo_id),
353 text_or_null(&session_id),
354 a.actor.id.as_str().into(),
355 a.actor.username.as_str().into(),
356 text_or_null(&harness),
357 text_or_null(&note),
358 minutes.into(),
359 at.as_str().into(),
360 expires_at.as_str().into(),
361 ])?];
362 let sent = push_delivery(
363 &mut statements,
364 self.delivery_to(
365 "work.claimed",
366 &repos,
367 event(&proposed),
368 &format!("{proposed_id}:claimed"),
369 "EXISTS (SELECT 1 FROM claims WHERE id = ?7)",
370 vec![proposed_id.as_str().into()],
371 )?,
372 );
373 let created = match self.db.batch(statements).await?.into_iter().next() {
374 Some(result) => result.results::<ClaimRow>()?.into_iter().next(),
375 None => None,
376 };
377 let Some(created) = created else {
378 return Err(worker::Error::RustError("the claim was not recorded".into()));
379 };
380 let claim = Claim::from(created);
381 if let (Some(session_id), Some(repo_id)) = (&claim.session_id, &claim.repo_id) {
382 self.link_item(session_id, repo_id, &held.item.id, LinkVia::Claim).await;
383 }
384 if claim.id != proposed_id {
385 return Ok(Outcome::Ok(Claimed { claim, others, blocked_by, item: None }));
386 }
387 self.delivered(sent, "work.claimed", held.event_repo().unwrap_or_default(), Some(a.actor.id.clone()), event(&claim)).await?;
388 Ok(Outcome::Ok(Claimed { claim, others, blocked_by, item: None }))
389 }
390
391 /// Ends the caller's active claim on a work item as `state`, announced
392 /// as `kind` with the event `event` makes of it, or refuses when they
393 /// hold none.
394 #[allow(clippy::too_many_arguments)]
395 async fn end_own_claim(
396 &self,
397 actor: &User,
398 workspace: &str,
399 repo: Option<&RepoPath>,
400 number: u32,
401 state: ClaimState,
402 reason: &str,
403 handoff: Option<(&Option<String>, &str)>,
404 session: Option<&str>,
405 kind: &'static str,
406 event: impl Fn(&Claim, &[String]) -> ClaimEvent,
407 ) -> Result<Outcome<Claim>> {
408 unwrap_or_fail!(self.claiming_repo(repo, actor).await?);
409 let workspace = workspace_of(workspace, repo);
410 let held = unwrap_or_fail!(items::for_viewer(self, &Some(actor.clone()), &workspace, number, Capability::Read).await?);
411 let Some(claim) = self.own_claim(&held.item.id, actor, session.map(str::trim).filter(|id| !id.is_empty())).await? else {
412 return Ok(Outcome::fail(FailureCode::NotFound, format!("You hold no claim on #{number}.")));
413 };
414 let repos = claim_repos(&held, claim.repo_id.as_deref());
415 let at = rfc3339(now_ms());
416 let expected = Claim {
417 state,
418 ended_at: Some(at.clone()),
419 end_reason: Some(reason.to_owned()),
420 handoff_to: handoff.and_then(|(to, _)| to.clone()),
421 note: handoff.map(|(_, note)| note.to_owned()).or(claim.note.clone()),
422 ..claim.clone()
423 };
424 let (handoff_to, note) = match handoff {
425 Some((to, note)) => (text_or_null(to), JsValue::from(note)),
426 None => (JsValue::NULL, JsValue::NULL),
427 };
428 let mut statements = vec![self
429 .db
430 .prepare(format!(
431 "UPDATE claims
432 SET state = ?, ended_at = ?, end_reason = ?, handoff_to = ?, note = COALESCE(?, note)
433 WHERE id = ? AND state = 'active'
434 RETURNING {CLAIM_COLUMNS}"
435 ))
436 .bind(&[state.as_str().into(), at.as_str().into(), reason.into(), handoff_to, note, claim.id.as_str().into()])?];
437 // A claim ends once: its id and how it ended name the change.
438 let sent = push_delivery(
439 &mut statements,
440 self.delivery_to(
441 kind,
442 &repos,
443 g1t_contracts::events::marked(event(&expected, &repos), Some(actor)),
444 &format!("{}:{}", claim.id, state.as_str()),
445 "EXISTS (SELECT 1 FROM claims WHERE id = ?7 AND state = ?8)",
446 vec![claim.id.as_str().into(), state.as_str().into()],
447 )?,
448 );
449 let ended = match self.db.batch(statements).await?.into_iter().next() {
450 Some(result) => result.results::<ClaimRow>()?.into_iter().next(),
451 None => None,
452 };
453 let Some(row) = ended else {
454 return Ok(Outcome::fail(FailureCode::NotFound, format!("You hold no claim on #{number}.")));
455 };
456 let claim = Claim::from(row);
457 let data = g1t_contracts::events::marked(event(&claim, &repos), Some(actor));
458 self.delivered(sent, kind, held.event_repo().unwrap_or_default(), Some(actor.id.clone()), data).await?;
459 Ok(Outcome::Ok(claim))
460 }
461
462 pub(crate) async fn release_claim(&self, a: ReleaseArgs) -> Result<Outcome<Claim>> {
463 let reason = tidy(a.reason.as_deref(), MAX_NOTE_CHARS).unwrap_or_else(|| "released".to_owned());
464 let claim = unwrap_or_fail!(
465 self.end_own_claim(
466 &a.actor,
467 &a.workspace,
468 a.repo.as_ref(),
469 a.number,
470 ClaimState::Released,
471 &reason,
472 None,
473 a.session_id.as_deref(),
474 "work.released",
475 |claim, repos| ClaimEvent { reason: Some(reason.clone()), ..ClaimEvent::of(claim, repos) },
476 )
477 .await?
478 );
479 Ok(Outcome::Ok(claim))
480 }
481
482 pub(crate) async fn hand_off_work_item(&self, a: HandOffArgs) -> Result<Outcome<Claim>> {
483 let Some(note) = tidy(Some(&a.note), MAX_NOTE_CHARS) else {
484 return Ok(Outcome::fail(FailureCode::Invalid, "Say where the work stands in note, for whoever picks it up."));
485 };
486 let to = tidy(a.to.as_deref().map(|to| to.trim().trim_start_matches('@')), 100).map(|to| to.to_lowercase());
487 if let Some(name) = &to {
488 let account: Viewer =
489 g1t_kit::call(&self.identity, "user_by_username", &UsernameArgs { username: name.clone() }).await?;
490 if account.is_none() {
491 return Ok(Outcome::fail(FailureCode::Invalid, format!("There is no account named {name}.")));
492 }
493 }
494 let claim = unwrap_or_fail!(
495 self.end_own_claim(
496 &a.actor,
497 &a.workspace,
498 a.repo.as_ref(),
499 a.number,
500 ClaimState::HandedOff,
501 "handed_off",
502 Some((&to, &note)),
503 a.session_id.as_deref(),
504 "work.handoff_requested",
505 |claim, repos| ClaimEvent {
506 from: Some(a.actor.username.clone()),
507 to: to.clone(),
508 note: Some(note.clone()),
509 ..ClaimEvent::of(claim, repos)
510 },
511 )
512 .await?
513 );
514 Ok(Outcome::Ok(claim))
515 }
516
517 /// A workspace's claims on the work items the viewer may see; with a
518 /// repository, those made from it or on work items linked to it.
519 pub(crate) async fn list_claims(&self, a: ListClaimsArgs) -> Result<Outcome<Vec<Claim>>> {
520 let workspace = workspace_of(&a.workspace, a.repo.as_ref());
521 let repo = match &a.repo {
522 Some(path) => Some(unwrap_or_fail!(self.coordination_repo(path, &a.viewer).await?)),
523 None => None,
524 };
525 let state = a.state.unwrap_or(ClaimState::Active);
526 let rows = self
527 .db
528 .prepare(format!(
529 "SELECT {CLAIM_COLUMNS} FROM claims
530 WHERE state = ?2 AND {workspace} = ?1 AND (?3 IS NULL OR {number} = ?3)
531 AND (?4 IS NULL OR repo_id = ?4 OR item_id IN {linked})
532 ORDER BY created_at DESC, id DESC LIMIT ?5",
533 workspace = items::sql::item_workspace!("claims.item_id"),
534 number = items::sql::item_number!("claims.item_id"),
535 linked = items::sql::items_linked_to!("?4"),
536 ))
537 .bind(&[
538 workspace.as_str().into(),
539 state.as_str().into(),
540 a.item.filter(|number| *number > 0).map_or(JsValue::NULL, JsValue::from),
541 text_or_null(&repo.as_ref().map(|repo| repo.id.clone())),
542 LIST_LIMIT.into(),
543 ])?
544 .all()
545 .await?
546 .results::<ClaimRow>()?;
547 let mut ids: Vec<String> = rows.iter().map(|row| row.item_id.clone()).collect();
548 ids.sort();
549 ids.dedup();
550 let seen: Vec<String> = items::by_ids(self, &ids)
551 .await?
552 .into_iter()
553 .filter(|held| held.may(a.viewer.as_ref(), Capability::Read))
554 .map(|held| held.item.id)
555 .collect();
556 Ok(Outcome::Ok(rows.into_iter().filter(|row| seen.contains(&row.item_id)).map(Claim::from).collect()))
557 }
558
559 /// A work item closed: its active claims end as released
560 /// (`item_closed`), each announced.
561 pub(crate) async fn release_claims_on_close(&self, held: &Held, actor: Option<&User>) -> Result<()> {
562 let at = rfc3339(now_ms());
563 let active = self.active_claims(&held.item.id).await?;
564 if active.is_empty() {
565 return Ok(());
566 }
567 let event = |claim: &Claim| {
568 let repos = claim_repos(held, claim.repo_id.as_deref());
569 g1t_contracts::events::marked(ClaimEvent { reason: Some("item_closed".to_owned()), ..ClaimEvent::of(claim, &repos) }, actor)
570 };
571 // Each claim's release and its updates are one change, in one batch.
572 let mut statements = Vec::new();
573 let mut sent = Vec::new();
574 for claim in &active {
575 statements.push(
576 self.db
577 .prepare(format!(
578 "UPDATE claims SET state = 'released', ended_at = ?, end_reason = 'item_closed'
579 WHERE id = ? AND state = 'active'
580 RETURNING {CLAIM_COLUMNS}"
581 ))
582 .bind(&[at.as_str().into(), claim.id.as_str().into()])?,
583 );
584 let expected = Claim {
585 state: ClaimState::Released,
586 ended_at: Some(at.clone()),
587 end_reason: Some("item_closed".to_owned()),
588 ..claim.clone()
589 };
590 let delivery = self.delivery_to(
591 "work.released",
592 &claim_repos(held, claim.repo_id.as_deref()),
593 event(&expected),
594 &format!("{}:{}", claim.id, ClaimState::Released.as_str()),
595 "EXISTS (SELECT 1 FROM claims WHERE id = ?7 AND state = 'released')",
596 vec![claim.id.as_str().into()],
597 )?;
598 let index = statements.len() - 1;
599 sent.push((index, push_delivery(&mut statements, delivery)));
600 }
601 let results = self.db.batch(statements).await?;
602 for (index, sent) in sent {
603 let Some(row) = results.get(index).map(|result| result.results::<ClaimRow>()).transpose()?.and_then(|rows| rows.into_iter().next())
604 else {
605 continue;
606 };
607 let claim = Claim::from(row);
608 let repo = claim.repo_id.clone().or_else(|| held.event_repo().map(str::to_owned)).unwrap_or_default();
609 self.delivered(sent, "work.released", &repo, actor.map(|actor| actor.id.clone()), event(&claim)).await?;
610 }
611 Ok(())
612 }
613}