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
525 linesCodeBlameRaw
1//! What a push to a branch settles, as a merge from the site would.
2//!
3//! An open pull request into the branch whose head the push brought in was
4//! merged by it: it is recorded as merged by whoever pushed, through the
5//! same bookkeeping as any merge (`Work::record_merge`). A push that landed
6//! a pull request names it (`GitPush::landed_pull`): whoever landed it
7//! records that one, and the merge queue's entry being landed is left to
8//! the queue.
9//!
10//! A commit pushed to the default branch whose message names a work item
11//! after a closing keyword (`Fixes #12`, closing.rs) lands it there, where
12//! the repository closes work items that way, and the work item says which
13//! commit and who pushed it; `Part of #12` links it.
14
15use g1t_contracts::events::Event;
16use g1t_contracts::repos::{BranchDriftArgs, BranchDrifts, Commit, GetByIdArgs, Landed, LogArgs, Pushed, PushedArgs, Repo, RepoPath};
17use g1t_contracts::rules::MergeMethod;
18use g1t_contracts::work::*;
19use g1t_contracts::{FailureCode, Outcome, User, Viewer};
20use serde::Deserialize;
21use worker::Result;
22use worker::wasm_bindgen::JsValue;
23
24use crate::items::Held;
25use crate::{Work, closing};
26
27/// How many open pull requests into a branch are looked at after a push.
28const MAX_CANDIDATES: u32 = 100;
29
30/// An open pull request into the pushed branch, and the head it is at.
31#[derive(Deserialize)]
32struct CandidateRow {
33 id: String,
34 head_commit: String,
35}
36
37/// The open pull requests into a branch (`None`: the default branch) that
38/// a push may have merged: those with a head, but not the one the push
39/// landed, nor one the merge queue is landing.
40const CANDIDATES: &str = "SELECT id, head_commit FROM pulls
41 WHERE repo_id = ?1 AND status IN ('draft', 'open') AND base_branch IS ?2 AND head_commit IS NOT NULL
42 AND number IS NOT ?3
43 AND NOT EXISTS (SELECT 1 FROM queue_entries q
44 WHERE q.pull_id = pulls.id AND q.state = 'passed' AND q.landing_commit IS NOT NULL)
45 ORDER BY updated_at DESC LIMIT ?4";
46
47/// A session (`?1`) of a repository (`?2`).
48const SESSION_IN_REPO: &str = "SELECT id AS value FROM sessions WHERE id = ?1 AND repo_id = ?2";
49
50/// The session, still going, of the caller (`?2`, or the person it acts
51/// for, `?3`) that works on work item `?1`: attached to it or tied to it
52/// since, in any repository; the one seen last.
53const CALLER_SESSION: &str = "SELECT id AS value FROM sessions
54 WHERE state = 'active' AND actor_id IN (?2, ?3)
55 AND (item_id = ?1 OR EXISTS (SELECT 1 FROM session_links l
56 WHERE l.session_id = sessions.id AND l.kind = 'item' AND l.item_id = ?1))
57 ORDER BY last_seen_at DESC LIMIT 1";
58
59/// A note of what closed a work item, and the session it was closed in.
60pub(crate) fn closed_note(note: &str, session: Option<&str>) -> String {
61 match session {
62 Some(session) => format!("{note} via session {session}"),
63 None => note.to_owned(),
64 }
65}
66
67/// The work items a commit (`?2`) pushed to repository `?1` resolved.
68const CLOSED_BY_COMMIT: &str = "SELECT i.workspace, i.number, i.title, i.state = 'closed' AS closed
69 FROM work_item_repos r JOIN work_items i ON i.id = r.item_id
70 WHERE r.repo_id = ?1 AND i.number IS NOT NULL
71 AND EXISTS (SELECT 1 FROM json_each(i.resolutions) j
72 WHERE json_extract(j.value, '$.repo_id') = ?1 AND json_extract(j.value, '$.commit') = ?2)
73 ORDER BY i.number LIMIT 50";
74
75#[derive(Deserialize)]
76struct ClosedRow {
77 workspace: String,
78 number: u32,
79 title: String,
80 closed: u8,
81}
82
83/// References, each with the commit that named it.
84pub(crate) type CommitRefs = Vec<(refs::Ref, String)>;
85
86/// The references a push's commits name after a closing keyword, and after
87/// `Part of`, each with the first commit, oldest first, that names it: in
88/// the order named, each once, no more than one pull request may close.
89pub(crate) fn commit_refs(commits: &[g1t_contracts::repos::PushedCommit]) -> (CommitRefs, CommitRefs) {
90 let mut closes: CommitRefs = Vec::new();
91 let mut parts: CommitRefs = Vec::new();
92 for commit in commits {
93 let named = refs::named(&commit.message);
94 for reference in named.closes {
95 if !closes.iter().any(|(known, _)| *known == reference) {
96 closes.push((reference, commit.hash.clone()));
97 }
98 }
99 for reference in named.parts {
100 if !parts.iter().any(|(known, _)| *known == reference) {
101 parts.push((reference, commit.hash.clone()));
102 }
103 }
104 }
105 parts.retain(|(part, _)| !closes.iter().any(|(closed, _)| closed == part));
106 closes.truncate(closing::MAX_CLOSES);
107 parts.truncate(closing::MAX_CLOSES);
108 (closes, parts)
109}
110
111impl Work {
112 /// After a push moved `branch` of a repository: records the pull
113 /// requests into it that the push merged, and, on the default branch,
114 /// lands the work items its commits name. Returns the pull requests it
115 /// recorded as merged. Never fails the push's handling: a problem is
116 /// logged.
117 pub(crate) async fn after_branch_push(&self, event: &Event, repo_id: &str, branch: &str, after: &str) -> Vec<String> {
118 match self.settle_push(event, repo_id, branch, after).await {
119 Ok(merged) => merged,
120 Err(error) => {
121 worker::console_error!("push to {branch} of {repo_id}: merges and closes not recorded: {error}");
122 Vec::new()
123 }
124 }
125 }
126
127 async fn settle_push(&self, event: &Event, repo_id: &str, branch: &str, after: &str) -> Result<Vec<String>> {
128 // Who pushed is who merged; a push nobody made (an import) settles nothing.
129 let Some(actor_id) = event.actor.as_deref() else { return Ok(Vec::new()) };
130 let default = event.data["defaultBranch"].as_bool() == Some(true);
131 let landed_pull = event.data["landedPull"].as_u64().and_then(|n| u32::try_from(n).ok());
132 let base = if default { JsValue::NULL } else { branch.into() };
133 let candidates = self
134 .db
135 .prepare(CANDIDATES)
136 .bind(&[repo_id.into(), base, landed_pull.map_or(JsValue::NULL, Into::into), MAX_CANDIDATES.into()])?
137 .all()
138 .await?
139 .results::<CandidateRow>()?;
140 let closes_items = default && self.settings(repo_id).await?.auto_close_items;
141 if candidates.is_empty() && !closes_items {
142 return Ok(Vec::new());
143 }
144 let pushed: Pushed = g1t_kit::call(
145 &self.repos,
146 "pushed",
147 &PushedArgs {
148 repo_id: repo_id.to_owned(),
149 before: event.data["before"].as_str().map(str::to_owned),
150 after: after.to_owned(),
151 heads: candidates.iter().map(|row| row.head_commit.clone()).collect(),
152 messages: closes_items,
153 // No `before` on the default branch: the repository's first push.
154 first: default && event.data["before"].as_str().is_none(),
155 },
156 )
157 .await?;
158 if pushed.landed.is_empty() && pushed.commits.is_empty() {
159 return Ok(Vec::new());
160 }
161 let actor: Viewer =
162 g1t_kit::call(&self.identity, "user_by_id", &g1t_contracts::identity::UserIdArgs { id: actor_id.to_owned() }).await?;
163 let Some(actor) = actor else { return Ok(Vec::new()) };
164 let repo: Outcome<Repo> =
165 g1t_kit::call(&self.repos, "get_by_id", &GetByIdArgs { id: repo_id.to_owned(), viewer: Some(actor.clone()) }).await?;
166 let Outcome::Ok(repo) = repo else { return Ok(Vec::new()) };
167
168 let mut merged = Vec::new();
169 for row in candidates.iter().filter(|row| pushed.landed.contains(&row.head_commit)) {
170 let Some(pull) = self.pull_by_id(&row.id).await? else { continue };
171 let landed = Landed {
172 commit: after.to_owned(),
173 previous: event.data["before"].as_str().map(str::to_owned),
174 method: if row.head_commit == after { MergeMethod::FastForward } else { MergeMethod::Merge },
175 rebased: Vec::new(),
176 };
177 let recorded = self.record_merge(&repo, pull, &actor, false, landed, Some(branch)).await?;
178 if recorded.status != PullStatus::Merged {
179 continue;
180 }
181 // Waiting in the merge queue: it has nothing left to land.
182 if self.leave(repo_id, &recorded, QueueState::Removed, Some("A push merged it.")).await? {
183 self.publish_as(
184 "queue.changed",
185 repo_id,
186 None,
187 g1t_contracts::events::QueueChanged { repo_id: repo_id.to_owned() },
188 )
189 .await?;
190 }
191 merged.push(recorded.id);
192 }
193
194 // A session g1t started pushed it: what it closes is g1t's, for the pusher.
195 let agent = event.data["agent"].as_str().is_some();
196 if closes_items {
197 let (closes, parts) = commit_refs(&pushed.commits);
198 for (reference, commit) in closes.iter().map(|(r, c)| (r, c)).chain(parts.iter().map(|(r, c)| (r, c))) {
199 let Some(id) = self.ref_item(&repo, reference).await? else { continue };
200 let closing = closes.iter().any(|(known, _)| known == reference);
201 let message = pushed.commits.iter().find(|pushed| pushed.hash == *commit).map(|pushed| pushed.message.as_str());
202 let session = self.trailer_session(&repo.id, message.unwrap_or_default()).await?;
203 let landing =
204 closing::Landing { pull: None, commit: Some(commit.clone()), pushed_by: Some(actor.username.clone()) };
205 let (closes, parts) = if closing { (vec![id], Vec::new()) } else { (Vec::new(), vec![id]) };
206 self.land(&repo, &landing, &actor, &closes, &parts, session.as_deref(), agent).await?;
207 }
208 }
209 Ok(merged)
210 }
211
212 /// `close_work_item` with a commit: the commit, on the default branch
213 /// of `repo`, whose message names work item `held` after a closing
214 /// keyword, with the session it names, if it names one of this
215 /// repository's. Anything else is refused, saying why.
216 pub(crate) async fn closing_commit(
217 &self,
218 repo: &Repo,
219 held: &Held,
220 commit: &str,
221 actor: &User,
222 ) -> Result<Outcome<(String, Option<String>)>> {
223 let number = held.item.number;
224 let commit = commit.trim().to_lowercase();
225 if !(4..=64).contains(&commit.len()) || !commit.bytes().all(|b| b.is_ascii_hexdigit()) {
226 return Ok(Outcome::fail(FailureCode::Invalid, "commit is a commit's hash."));
227 }
228 let path = RepoPath { namespace: repo.namespace.clone(), name: repo.name.clone() };
229 let log = |at: Option<String>| LogArgs { path: path.clone(), viewer: Some(actor.clone()), git_ref: at, limit: 1 };
230 let (named, tip) = (log(Some(commit.clone())), log(None));
231 let (found, head) = futures_util::future::join(
232 g1t_kit::call::<_, Outcome<Vec<Commit>>>(&self.repos, "log", &named),
233 g1t_kit::call::<_, Outcome<Vec<Commit>>>(&self.repos, "log", &tip),
234 )
235 .await;
236 let found = match found? {
237 Outcome::Ok(commits) => commits.into_iter().next().filter(|found| found.hash.starts_with(&commit)),
238 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
239 };
240 let Some(found) = found else {
241 return Ok(Outcome::fail(
242 FailureCode::NotFound,
243 format!("There is no commit {commit} in {}/{}; give its full hash.", repo.namespace, repo.name),
244 ));
245 };
246 let short = &found.hash[..7.min(found.hash.len())];
247 let mut names = false;
248 for reference in refs::named(&found.message).closes {
249 if self.ref_item(repo, &reference).await?.as_deref() == Some(held.item.id.as_str()) {
250 names = true;
251 break;
252 }
253 }
254 if !names {
255 return Ok(Outcome::fail(
256 FailureCode::Invalid,
257 format!("Commit {short} does not close #{number}: its message names it after no closing keyword (Fixes #{number})."),
258 ));
259 }
260 let head = match head? {
261 Outcome::Ok(commits) => commits.into_iter().next(),
262 Outcome::Fail(_) => None,
263 };
264 let Some(head) = head else {
265 return Ok(Outcome::fail(FailureCode::Invalid, format!("Commit {short} is not on {}.", repo.default_branch)));
266 };
267 if head.hash != found.hash {
268 let drifts: Outcome<BranchDrifts> = g1t_kit::call(
269 &self.repos,
270 "branch_drift",
271 &BranchDriftArgs { path: path.clone(), viewer: Some(actor.clone()), base: head.hash, heads: vec![found.hash.clone()] },
272 )
273 .await?;
274 let drift = match drifts {
275 Outcome::Ok(drifts) => drifts.branches.into_iter().next().and_then(|branch| branch.drift),
276 Outcome::Fail(_) => None,
277 };
278 match drift {
279 Some(drift) if drift.ahead == 0 => {}
280 Some(_) => {
281 return Ok(Outcome::fail(
282 FailureCode::Invalid,
283 format!("Commit {short} is not on {}, the default branch: only a commit there closes a work item.", repo.default_branch),
284 ));
285 }
286 None => {
287 return Ok(Outcome::fail(
288 FailureCode::Invalid,
289 format!("Could not tell whether commit {short} is on {}: it is too far back.", repo.default_branch),
290 ));
291 }
292 }
293 }
294 let session = self.trailer_session(&repo.id, &found.message).await?;
295 Ok(Outcome::Ok((found.hash, session)))
296 }
297
298 /// The session a commit message names in a `G1t-Session` trailer, when
299 /// it is one of this repository's.
300 pub(crate) async fn trailer_session(&self, repo_id: &str, message: &str) -> Result<Option<String>> {
301 let Some(id) = session_trailers(message).into_iter().next() else { return Ok(None) };
302 Ok(self
303 .db
304 .prepare(SESSION_IN_REPO)
305 .bind(&[id.as_str().into(), repo_id.into()])?
306 .first::<crate::rows::ValueRow>(None)
307 .await?
308 .map(|row| row.value))
309 }
310
311 /// The session closing a work item with a token: the caller's own
312 /// session on it that is still going, the one seen last. A person
313 /// signed in on the site closes it in none.
314 pub(crate) async fn caller_session(&self, held: &Held, actor: &User) -> Result<Option<String>> {
315 if actor.token.is_none() && actor.acting.is_none() {
316 return Ok(None);
317 }
318 let person = actor.acting.as_ref().map_or(actor.id.as_str(), |acting| acting.on_behalf_of.id.as_str());
319 Ok(self
320 .db
321 .prepare(CALLER_SESSION)
322 .bind(&[held.item.id.as_str().into(), actor.id.as_str().into(), person.into()])?
323 .first::<crate::rows::ValueRow>(None)
324 .await?
325 .map(|row| row.value))
326 }
327
328 /// `closed_by_commit`: the work items a commit pushed to a repository
329 /// resolved, for anyone who may read the repository.
330 pub(crate) async fn closed_by_commit(&self, a: ClosedByCommitArgs) -> Result<Outcome<Vec<ClosedItem>>> {
331 let Some(commit) = crate::commit_comments::valid_commit(&a.commit).filter(|commit| commit.len() >= 40) else {
332 return Ok(Outcome::Ok(Vec::new()));
333 };
334 let repo = match self.repo(&a.repo, &a.viewer).await? {
335 Outcome::Ok(repo) => repo,
336 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
337 };
338 Ok(Outcome::Ok(
339 self.db
340 .prepare(CLOSED_BY_COMMIT)
341 .bind(&[repo.id.as_str().into(), commit.into()])?
342 .all()
343 .await?
344 .results::<ClosedRow>()?
345 .into_iter()
346 .map(|row| ClosedItem { workspace: row.workspace, number: row.number, title: row.title, closed: row.closed != 0 })
347 .collect(),
348 ))
349 }
350}
351
352#[cfg(test)]
353mod tests {
354 use g1t_contracts::repos::PushedCommit;
355 use rusqlite::{Connection, OptionalExtension, params};
356
357 use super::*;
358
359 /// The work database as the migrations make it.
360 fn database() -> Connection {
361 crate::migrated::database()
362 }
363
364 fn pull(db: &Connection, id: &str, number: u32, status: &str, base: Option<&str>, head: Option<&str>) {
365 db.execute(
366 "INSERT INTO pulls (id, repo_id, number, title, agent, runtime, author_id, author_name, status,
367 base_branch, head_commit, created_at, updated_at)
368 VALUES (?1, 'repo', ?2, 'Change', 'codex', 'external', 'user', 'user', ?3, ?4, ?5, 'now', 'now')",
369 params![id, number, status, base, head],
370 )
371 .unwrap();
372 }
373
374 fn candidates(db: &Connection, base: Option<&str>, landed: Option<u32>) -> Vec<String> {
375 let mut statement = db.prepare(CANDIDATES).unwrap();
376 let mut ids: Vec<String> = statement
377 .query_map(params!["repo", base, landed, MAX_CANDIDATES], |row| row.get(0))
378 .unwrap()
379 .map(Result::unwrap)
380 .collect();
381 ids.sort();
382 ids
383 }
384
385 #[test]
386 fn the_open_pull_requests_into_the_branch_are_looked_at() {
387 let db = database();
388 pull(&db, "open", 1, "open", None, Some("h1"));
389 pull(&db, "draft", 2, "draft", None, Some("h2"));
390 pull(&db, "closed", 3, "closed", None, Some("h3"));
391 pull(&db, "merged", 4, "merged", None, Some("h4"));
392 pull(&db, "empty", 5, "open", None, None);
393 pull(&db, "release", 6, "open", Some("release"), Some("h6"));
394 assert_eq!(candidates(&db, None, None), vec!["draft", "open"]);
395 assert_eq!(candidates(&db, Some("release"), None), vec!["release"]);
396 // The pull request the push landed is recorded by whoever landed it.
397 assert_eq!(candidates(&db, None, Some(1)), vec!["draft"]);
398 }
399
400 #[test]
401 fn the_entry_the_merge_queue_is_landing_is_left_to_it() {
402 let db = database();
403 pull(&db, "landing", 1, "open", None, Some("h1"));
404 pull(&db, "waiting", 2, "open", None, Some("h2"));
405 db.execute_batch(
406 "INSERT INTO queue_entries (id, state, repo_id, pull_id, number, enqueued_by, created_at, landing_commit)
407 VALUES ('e1', 'passed', 'repo', 'landing', 1, '{}', 'now', 'c1');
408 INSERT INTO queue_entries (id, state, repo_id, pull_id, number, enqueued_by, created_at)
409 VALUES ('e2', 'waiting', 'repo', 'waiting', 2, '{}', 'now');",
410 )
411 .unwrap();
412 assert_eq!(candidates(&db, None, None), vec!["waiting"]);
413 }
414
415 #[test]
416 fn a_commit_names_what_it_resolved_in_its_repository() {
417 let db = database();
418 let resolved = |repo: &str| format!(r#"[{{"repo_id":"{repo}","pull":null,"commit":"c1","pushed_by":"ana"}}]"#);
419 let item = |id: &str, number: u32, title: &str, state: &str, repo: &str| {
420 db.execute(
421 "INSERT INTO work_items (id, workspace, number, title, body, state, resolutions, author_id, author_name, created_at, updated_at)
422 VALUES (?1, 'acme', ?2, ?3, '', ?4, ?5, 'user', 'user', 'now', 'now')",
423 params![id, number, title, state, resolved(repo)],
424 )
425 .unwrap();
426 };
427 item("wki_1", 7, "Broken header", "closed", "repo");
428 item("wki_2", 8, "Footer", "open", "repo");
429 item("wki_3", 9, "Elsewhere", "closed", "other");
430 db.execute_batch(
431 "INSERT INTO work_item_repos (item_id, repo_id, role, added_at) VALUES
432 ('wki_1', 'repo', 'origin', 'now'), ('wki_2', 'repo', 'touched', 'now'), ('wki_3', 'other', 'origin', 'now');",
433 )
434 .unwrap();
435 let closed: Vec<(String, u32, String, u8)> = db
436 .prepare(CLOSED_BY_COMMIT)
437 .unwrap()
438 .query_map(params!["repo", "c1"], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)))
439 .unwrap()
440 .map(Result::unwrap)
441 .collect();
442 assert_eq!(closed, vec![("acme".to_owned(), 7, "Broken header".to_owned(), 1), ("acme".to_owned(), 8, "Footer".to_owned(), 0)]);
443 }
444
445 fn commit(hash: &str, message: &str) -> PushedCommit {
446 PushedCommit { hash: hash.into(), message: message.into() }
447 }
448
449 fn here(number: u32) -> refs::Ref {
450 refs::Ref { scope: refs::RefScope::Here, number }
451 }
452
453 #[test]
454 fn each_reference_is_landed_by_the_first_commit_that_names_it() {
455 let commits = [
456 commit("a", "Fix the header\n\nFixes #12, closes #3"),
457 commit("b", "Tidy up (#12)\n\nResolves #12 and resolves acme/web#7. Part of #20"),
458 commit("c", "Mention #9 without a keyword; see `fixes #10` in code"),
459 commit("d", "Refs #3, part of #21"),
460 ];
461 let (closes, parts) = commit_refs(&commits);
462 let web = refs::Ref { scope: refs::RefScope::Repo(RepoPath { namespace: "acme".into(), name: "web".into() }), number: 7 };
463 assert_eq!(closes, vec![(here(12), "a".into()), (here(3), "a".into()), (web, "b".into())]);
464 // What a commit closes is not also only part of it.
465 assert_eq!(parts, vec![(here(20), "b".into()), (here(21), "d".into())]);
466 }
467
468 #[test]
469 fn no_more_than_a_pull_request_closes() {
470 let message: String = (1..=80).map(|n| format!("Fixes #{n}\n")).collect();
471 assert_eq!(commit_refs(&[commit("a", &message)]).0.len(), closing::MAX_CLOSES);
472 }
473
474 #[test]
475 fn a_close_names_the_session_it_was_made_in() {
476 assert_eq!(closed_note("closed this in abc", Some("ses_1")), "closed this in abc via session ses_1");
477 assert_eq!(closed_note("closed this as completed", None), "closed this as completed");
478 }
479
480 #[test]
481 fn the_callers_session_on_the_work_item_is_the_one_still_going_seen_last() {
482 let db = database();
483 db.execute_batch(
484 "INSERT INTO work_items (id, workspace, number, title, body, author_id, author_name, created_at, updated_at)
485 VALUES ('wki_7', 'acme', 7, 't', '', 'u', 'u', '', ''), ('wki_8', 'acme', 8, 't', '', 'u', 'u', '', ''),
486 ('wki_9', 'acme', 9, 't', '', 'u', 'u', '', '');",
487 )
488 .unwrap();
489 let session = |id: &str, repo: &str, actor: &str, item: Option<&str>, state: &str, seen: &str| {
490 db.execute(
491 "INSERT INTO sessions (id, repo_id, item_id, actor_id, actor_name, harness, state, started_at, last_seen_at)
492 VALUES (?1, ?2, ?3, ?4, ?4, 'claude_code', ?5, 'then', ?6)",
493 params![id, repo, item, actor, state, seen],
494 )
495 .unwrap();
496 };
497 session("ses_old", "repo", "ana", Some("wki_7"), "active", "1");
498 // In another repository of the workspace, on the same work item.
499 session("ses_new", "other", "ana", Some("wki_7"), "active", "2");
500 session("ses_ended", "repo", "ana", Some("wki_7"), "ended", "3");
501 session("ses_other", "repo", "bo", Some("wki_7"), "active", "4");
502 session("ses_elsewhere", "repo", "ana", Some("wki_8"), "active", "5");
503 // Tied to work item 9 since it started, by a claim.
504 session("ses_claimed", "repo", "ana", None, "active", "6");
505 db.execute(
506 "INSERT INTO session_links (session_id, repo_id, kind, number, item_id, via, at)
507 VALUES ('ses_claimed', 'repo', 'item', 0, 'wki_9', 'claim', 'now')",
508 [],
509 )
510 .unwrap();
511 let caller = |item: &str, actor: &str, person: &str| -> Option<String> {
512 db.query_row(CALLER_SESSION, params![item, actor, person], |row| row.get(0)).optional().unwrap()
513 };
514 assert_eq!(caller("wki_7", "ana", "ana").as_deref(), Some("ses_new"));
515 // g1t acting for ana: ana's session.
516 assert_eq!(caller("wki_7", "g1t", "ana").as_deref(), Some("ses_new"));
517 assert_eq!(caller("wki_9", "ana", "ana").as_deref(), Some("ses_claimed"));
518 assert_eq!(caller("wki_7", "cy", "cy"), None);
519 let in_repo = |id: &str| -> Option<String> {
520 db.query_row(SESSION_IN_REPO, params![id, "repo"], |row| row.get(0)).optional().unwrap()
521 };
522 assert_eq!(in_repo("ses_old").as_deref(), Some("ses_old"));
523 assert_eq!(in_repo("ses_nowhere"), None);
524 }
525}