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
421 linesCodeBlameRaw
1//! Reading work items: one by number or id, as a viewer may see it, and
2//! `list_work_items` and `get_work_item`.
3
4use g1t_contracts::access::{self, Capability};
5use g1t_contracts::repos::{ReadableArgs, Repo};
6use g1t_contracts::work::*;
7use g1t_contracts::{Outcome, User, Viewer};
8use serde::Deserialize;
9use worker::Result;
10use worker::wasm_bindgen::JsValue;
11
12use super::rows::{Held, ITEM_COLUMNS, ItemRow, no_item, workspace_of};
13use crate::rows::{CommentRow, PULL_COLUMNS, PullRow};
14use crate::{LIST_PAGE, Work, optional, optional_number, state_name};
15
16/// What `list_work_items` reads: `?1` the workspace, then the filters, and
17/// `?8` whether the viewer is a member (else only work items filed in a
18/// public repository or one in `?9`, the repositories they were given a
19/// role on), at most `?10`. With `?11`, only those closed since then, the
20/// most recently closed first; with `?12`, only those numbered below it.
21/// With `?13`, only open ones on that horizon of the plan, in the plan's
22/// order (plan.rs).
23const LIST_SQL: &str = "SELECT {ITEM_COLUMNS} FROM work_items
24 WHERE workspace = ?1 AND (?2 IS NULL OR state = ?2)
25 AND (?3 IS NULL OR EXISTS (SELECT 1 FROM json_each(work_items.labels) WHERE json_each.value = ?3))
26 AND (?4 IS NULL OR milestone = ?4)
27 AND (?5 IS NULL OR EXISTS (SELECT 1 FROM json_each(work_items.assignees) WHERE json_each.value = ?5))
28 AND (?6 IS NULL OR EXISTS (SELECT 1 FROM work_item_repos r
29 WHERE r.item_id = work_items.id AND r.repo_id = ?6 AND (?7 IS NULL OR r.role = ?7)))
30 AND (?8 = 1 OR EXISTS (SELECT 1 FROM work_item_repos o JOIN repo_workspaces rw ON rw.repo_id = o.repo_id
31 WHERE o.item_id = work_items.id AND o.role = 'origin'
32 AND (rw.private = 0 OR o.repo_id IN (SELECT value FROM json_each(?9)))))
33 AND (?11 IS NULL OR closed_at >= ?11)
34 AND (?12 IS NULL OR number < ?12)
35 AND (?13 IS NULL OR (state = 'open' AND COALESCE(horizon, 'triage') = ?13))
36 ORDER BY CASE WHEN ?11 IS NULL THEN NULL ELSE closed_at END DESC,
37 CASE WHEN ?13 IS NULL THEN NULL ELSE urgent END DESC,
38 CASE WHEN ?13 IS NULL THEN NULL ELSE rank IS NULL END,
39 CASE WHEN ?13 IS NULL THEN NULL ELSE rank END,
40 number DESC LIMIT ?10";
41
42/// The work items matching `condition` (on `work_items`), with everything
43/// they are read with.
44pub(crate) fn select(condition: &str) -> String {
45 format!("SELECT {ITEM_COLUMNS} FROM work_items WHERE {condition}")
46}
47
48/// The work items a statement made by [`select`] read.
49pub(crate) fn held_rows(result: &worker::D1Result) -> Result<Vec<Held>> {
50 Ok(result.results::<ItemRow>()?.into_iter().filter_map(ItemRow::held).collect())
51}
52
53/// What a repository the reader may not read is called instead.
54const UNNAMED: &str = "a private repository";
55
56/// `text` with each of `hidden` (`owner/name`, any case) that stands as a
57/// name of its own, not inside a longer one, called [`UNNAMED`]; what
58/// follows it (`#5`, `@abc1234`) goes too.
59pub(crate) fn redact_paths(text: &str, hidden: &[String]) -> String {
60 let part = |c: char| c.is_alphanumeric() || matches!(c, '-' | '_' | '.' | '/');
61 let hidden: Vec<String> = hidden.iter().map(|path| path.to_lowercase()).collect();
62 let mut out = String::with_capacity(text.len());
63 let mut at = 0;
64 'scan: while let Some(c) = text[at..].chars().next() {
65 if text[..at].chars().next_back().is_none_or(|before| !part(before)) {
66 for path in &hidden {
67 let end = at + path.len();
68 let here = text.get(at..end).is_some_and(|found| found.eq_ignore_ascii_case(path));
69 if here && text[end..].chars().next().is_none_or(|after| !part(after)) {
70 out.push_str(UNNAMED);
71 // Its pull request or commit goes with it.
72 let mut rest = end;
73 if let Some(mark) = text[end..].chars().next().filter(|mark| matches!(mark, '#' | '@')) {
74 rest += mark.len_utf8();
75 rest += text[rest..].chars().take_while(char::is_ascii_alphanumeric).map(char::len_utf8).sum::<usize>();
76 }
77 at = rest;
78 continue 'scan;
79 }
80 }
81 }
82 out.push(c);
83 at += c.len_utf8();
84 }
85 out
86}
87
88impl Work {
89 /// A work item by workspace and number, whoever is asking.
90 pub(super) async fn item(&self, workspace: &str, number: u32) -> Result<Option<Held>> {
91 Ok(self
92 .db
93 .prepare(select("workspace = ? AND number = ?"))
94 .bind(&[workspace.to_lowercase().into(), number.into()])?
95 .first::<ItemRow>(None)
96 .await?
97 .and_then(ItemRow::held))
98 }
99
100 /// A work item by id, whoever is asking.
101 pub(super) async fn item_by_id(&self, id: &str) -> Result<Option<Held>> {
102 Ok(self
103 .db
104 .prepare(select("id = ?"))
105 .bind(&[id.into()])?
106 .first::<ItemRow>(None)
107 .await?
108 .and_then(ItemRow::held))
109 }
110
111 /// The work items of these ids, in no particular order.
112 pub(super) async fn items_by_ids(&self, ids: &[String]) -> Result<Vec<Held>> {
113 if ids.is_empty() {
114 return Ok(Vec::new());
115 }
116 held_rows(
117 &self
118 .db
119 .prepare(select("id IN (SELECT value FROM json_each(?))"))
120 .bind(&[serde_json::to_string(ids)?.into()])?
121 .all()
122 .await?,
123 )
124 }
125
126 /// Those of `ids` (repositories) that `viewer` may read, as
127 /// `repo_workspaces` records them: one not recorded is not.
128 pub(super) async fn readable_repo_ids(&self, viewer: Option<&User>, ids: &[String]) -> Result<std::collections::HashSet<String>> {
129 #[derive(Deserialize)]
130 struct Recorded {
131 repo_id: String,
132 workspace: String,
133 private: u8,
134 }
135 if ids.is_empty() {
136 return Ok(std::collections::HashSet::new());
137 }
138 let recorded = self
139 .db
140 .prepare("SELECT repo_id, workspace, private FROM repo_workspaces WHERE repo_id IN (SELECT value FROM json_each(?))")
141 .bind(&[serde_json::to_string(ids)?.into()])?
142 .all()
143 .await?
144 .results::<Recorded>()?;
145 Ok(recorded
146 .into_iter()
147 .filter(|row| super::rows::repo_readable(viewer, &row.repo_id, Some(&row.workspace), Some(row.private)))
148 .map(|row| row.repo_id)
149 .collect())
150 }
151
152 /// A work item, if `viewer` may do `capability` on it.
153 pub(super) async fn item_for(&self, viewer: &Viewer, workspace: &str, number: u32, capability: Capability) -> Result<Outcome<Held>> {
154 let Some(held) = self.item(workspace, number).await? else {
155 return Ok(no_item());
156 };
157 check!(held.allowed(viewer.as_ref(), capability));
158 Ok(Outcome::Ok(held))
159 }
160
161 /// A work item `actor` may change: theirs (`WorkItem::owner`), or one
162 /// they may triage.
163 pub(super) async fn manageable_item(&self, actor: &User, workspace: &str, number: u32) -> Result<Outcome<Held>> {
164 let held = check!(self.item_for(&Some(actor.clone()), workspace, number, Capability::Read).await?);
165 if held.item.owner().id != actor.id {
166 check!(held.allowed(Some(actor), Capability::Triage));
167 }
168 Ok(Outcome::Ok(held))
169 }
170
171 /// The ids of the work items of `workspace` with these numbers, in the
172 /// order given.
173 pub(super) async fn item_ids(&self, workspace: &str, numbers: &[u32]) -> Result<Vec<String>> {
174 #[derive(Deserialize)]
175 struct IdNumber {
176 id: String,
177 number: u32,
178 }
179 if numbers.is_empty() {
180 return Ok(Vec::new());
181 }
182 let rows = self
183 .db
184 .prepare("SELECT id, number FROM work_items WHERE workspace = ? AND number IN (SELECT value FROM json_each(?))")
185 .bind(&[workspace.into(), serde_json::to_string(numbers)?.into()])?
186 .all()
187 .await?
188 .results::<IdNumber>()?;
189 Ok(numbers.iter().filter_map(|number| rows.iter().find(|row| row.number == *number).map(|row| row.id.clone())).collect())
190 }
191
192 pub(crate) async fn list_work_items(&self, a: ListWorkItemsArgs) -> Result<Outcome<Vec<WorkItem>>> {
193 let workspace = workspace_of(&a.workspace, a.repo.as_ref());
194 let repo = match &a.repo {
195 Some(path) => {
196 let repo = check!(self.repo(path, &a.viewer).await?);
197 if !repo.namespace.eq_ignore_ascii_case(&workspace) {
198 return Ok(Outcome::Ok(Vec::new()));
199 }
200 Some(repo)
201 }
202 None => None,
203 };
204 let viewer = a.viewer.as_ref();
205 let member = viewer.and_then(|user| access::workspace_role(user, &workspace)).is_some();
206 let granted: Vec<&str> = viewer
207 .map(|user| user.grants.iter().filter(|grant| grant.workspace.eq_ignore_ascii_case(&workspace)).map(|grant| grant.repo_id.as_str()).collect())
208 .unwrap_or_default();
209 let label = a
210 .label
211 .map(|label| label.split_whitespace().collect::<Vec<_>>().join(" ").to_lowercase())
212 .filter(|label| !label.is_empty());
213 let assignee = a.assignee.map(|name| name.trim().trim_start_matches('@').to_lowercase()).filter(|name| !name.is_empty());
214 let state = state_name(a.state).map_or(JsValue::NULL, JsValue::from);
215 let query = self
216 .db
217 .prepare(LIST_SQL.replace("{ITEM_COLUMNS}", ITEM_COLUMNS))
218 .bind(&[
219 workspace.as_str().into(),
220 state,
221 optional(&label),
222 optional_number(a.milestone),
223 optional(&assignee),
224 optional(&repo.as_ref().map(|repo| repo.id.clone())),
225 a.role.map_or(JsValue::NULL, |role| role.as_str().into()),
226 u32::from(member).into(),
227 serde_json::to_string(&granted)?.into(),
228 a.limit.unwrap_or(LIST_PAGE).clamp(1, LIST_PAGE).into(),
229 optional(&a.closed_since.map(|at| at.trim().to_owned()).filter(|at| !at.is_empty())),
230 optional_number(if a.horizon.is_some() { None } else { a.before }),
231 a.horizon.map_or(JsValue::NULL, |horizon| horizon.as_str().into()),
232 ])?;
233 let rows = self.timing.db(1, query.all()).await?.results::<ItemRow>()?;
234 Ok(Outcome::Ok(
235 rows.into_iter()
236 .filter_map(ItemRow::held)
237 .filter(|held| held.may(viewer, Capability::Read))
238 .map(|held| held.shown(viewer))
239 .collect(),
240 ))
241 }
242
243 /// A work item, the pull requests for it the viewer may see, and its
244 /// comments and reactions.
245 pub(crate) async fn get_work_item(&self, a: WorkItemViewArgs) -> Result<Outcome<WorkItemDetail>> {
246 let held = check!(self.item_for(&a.viewer, &a.workspace, a.number, Capability::Read).await?);
247 let key = [JsValue::from(held.item.id.as_str())];
248 let statements = vec![
249 self.db.prepare(format!("SELECT {PULL_COLUMNS} FROM pulls WHERE item_id = ?1 ORDER BY created_at")).bind(&key)?,
250 self.db.prepare("SELECT * FROM comments WHERE item_id = ?1 ORDER BY id LIMIT 500").bind(&key)?,
251 self.db.prepare(crate::reactions::ITEM_THREAD_SQL).bind(&key)?,
252 ];
253 let results = self.timing.db(3, self.db.batch(statements)).await?;
254 let rows = |index: usize| results.get(index).ok_or_else(|| worker::Error::RustError("short batch".into()));
255 let pulls: Vec<Pull> = rows(0)?.results::<PullRow>()?.into_iter().map(Pull::from).collect();
256 let mut comments: Vec<Comment> = rows(1)?.results::<CommentRow>()?.into_iter().map(Comment::from).collect();
257 let reacted = rows(2)?.results::<crate::reactions::ReactionRow>()?;
258 let reactions = crate::reactions::attach(&reacted, &held.item.id, &mut comments);
259 let pulls = self.readable_pulls(&a.viewer, pulls).await?;
260 // What g1t noted names repositories ("added acme/api as a target",
261 // "closed this by merging acme/api#5"): those the viewer may not
262 // read are not named.
263 let hidden = self.hidden_paths(&held, a.viewer.as_ref()).await?;
264 if !hidden.is_empty() {
265 for comment in comments.iter_mut().filter(|comment| comment.kind == CommentKind::Event) {
266 comment.body = redact_paths(&comment.body, &hidden);
267 }
268 }
269 Ok(Outcome::Ok(WorkItemDetail { item: held.shown(a.viewer.as_ref()), pulls, comments, reactions }))
270 }
271
272 /// The paths (`owner/name`) of the repositories of `held`'s workspace,
273 /// and of those it is linked to, that `viewer` may not read.
274 async fn hidden_paths(&self, held: &Held, viewer: Option<&User>) -> Result<Vec<String>> {
275 #[derive(Deserialize)]
276 struct Recorded {
277 repo_id: String,
278 workspace: String,
279 path: String,
280 private: u8,
281 }
282 let recorded = self
283 .db
284 .prepare(
285 "SELECT repo_id, workspace, path, private FROM repo_workspaces
286 WHERE workspace = ?1 OR repo_id IN (SELECT repo_id FROM work_item_repos WHERE item_id = ?2)",
287 )
288 .bind(&[held.item.workspace.as_str().into(), held.item.id.as_str().into()])?
289 .all()
290 .await?
291 .results::<Recorded>()?;
292 Ok(recorded
293 .into_iter()
294 .filter(|row| !super::rows::repo_readable(viewer, &row.repo_id, Some(&row.workspace), Some(row.private)))
295 .map(|row| row.path)
296 .filter(|path| !path.is_empty())
297 .collect())
298 }
299
300 /// Those of `pulls` whose repositories `viewer` may read, each naming
301 /// the branch it merges into.
302 pub(super) async fn readable_pulls(&self, viewer: &Viewer, pulls: Vec<Pull>) -> Result<Vec<Pull>> {
303 if pulls.is_empty() {
304 return Ok(pulls);
305 }
306 let mut ids: Vec<String> = pulls.iter().map(|pull| pull.repo_id.clone()).collect();
307 ids.sort();
308 ids.dedup();
309 let readable: Vec<Repo> = g1t_kit::call(&self.repos, "readable", &ReadableArgs { ids, viewer: viewer.clone() }).await?;
310 Ok(pulls
311 .into_iter()
312 .filter_map(|mut pull| {
313 let repo = readable.iter().find(|repo| repo.id == pull.repo_id)?;
314 crate::fill_base(&mut pull, repo);
315 Some(pull)
316 })
317 .collect())
318 }
319}
320
321#[cfg(test)]
322mod tests {
323 use rusqlite::{Connection, ToSql};
324
325 use super::super::links::{MARK_DELETED, REMEMBER_REPO};
326 use super::super::rows::ITEM_COLUMNS;
327 use super::{LIST_SQL, redact_paths};
328 use crate::migrated::database;
329
330 /// acme's work items 1 to 5, linked to acme/web; 3, 4 and 5 closed,
331 /// in the order 4, 5, 3; 2 also to acme/hello.
332 fn acme() -> Connection {
333 let db = database();
334 db.execute_batch(
335 "INSERT INTO repo_workspaces (repo_id, workspace, path, private)
336 VALUES ('rep_web', 'acme', 'acme/web', 0), ('rep_hello', 'acme', 'acme/hello', 0);
337 INSERT INTO work_items (id, workspace, number, title, body, labels, author_id, author_name, created_at, updated_at,
338 state, closed_at)
339 VALUES ('wki_1', 'acme', 1, 'One', '', '[]', 'usr_1', 'ana', '', '', 'open', NULL),
340 ('wki_2', 'acme', 2, 'Two', '', '[]', 'usr_1', 'ana', '', '', 'open', NULL),
341 ('wki_3', 'acme', 3, 'Three', '', '[]', 'usr_1', 'ana', '', '', 'closed', '2026-10-17T00:00:00Z'),
342 ('wki_4', 'acme', 4, 'Four', '', '[]', 'usr_1', 'ana', '', '', 'closed', '2026-10-01T00:00:00Z'),
343 ('wki_5', 'acme', 5, 'Five', '', '[]', 'usr_1', 'ana', '', '', 'closed', '2026-10-10T00:00:00Z');
344 INSERT INTO work_item_repos (item_id, repo_id, role, path, added_at)
345 VALUES ('wki_1', 'rep_web', 'origin', 'acme/web', ''), ('wki_2', 'rep_hello', 'origin', 'acme/hello', ''),
346 ('wki_2', 'rep_web', 'target', 'acme/web', ''), ('wki_3', 'rep_web', 'origin', 'acme/web', ''),
347 ('wki_4', 'rep_web', 'origin', 'acme/web', ''), ('wki_5', 'rep_web', 'origin', 'acme/web', '');",
348 )
349 .unwrap();
350 db
351 }
352
353 /// The numbers `list_work_items` reads for acme as a member, in its order.
354 fn listed(db: &Connection, state: Option<&str>, limit: u32, closed_since: Option<&str>, before: Option<u32>) -> Vec<u32> {
355 let sql = LIST_SQL.replace("{ITEM_COLUMNS}", ITEM_COLUMNS);
356 let values: [&dyn ToSql; 13] = [&"acme", &state, &None::<String>, &None::<u32>, &None::<String>, &None::<String>, &None::<String>, &1, &"[]", &limit, &closed_since, &before, &None::<String>];
357 let mut statement = db.prepare(&sql).unwrap();
358 let numbers = statement.query_map(&values[..], |row| row.get::<_, u32>("number")).unwrap();
359 numbers.map(Result::unwrap).collect()
360 }
361
362 #[test]
363 fn work_items_page_on_by_number_and_closed_ones_come_by_when_they_closed() {
364 let db = acme();
365 assert_eq!(listed(&db, None, 100, None, None), [5, 4, 3, 2, 1]);
366 // A page, then the next from below its lowest number.
367 assert_eq!(listed(&db, None, 2, None, None), [5, 4]);
368 assert_eq!(listed(&db, None, 2, None, Some(4)), [3, 2]);
369 assert_eq!(listed(&db, None, 2, None, Some(2)), [1]);
370 // Closed since a time: the most recently closed first, older ones left out.
371 assert_eq!(listed(&db, Some("closed"), 100, Some("2026-10-04T00:00:00Z"), None), [3, 5]);
372 }
373
374 /// Whether each of work item 2's links says its repository is deleted, by path.
375 fn deleted_links(db: &Connection) -> Vec<(String, bool)> {
376 let links: String = db
377 .query_row(&format!("SELECT links FROM (SELECT {ITEM_COLUMNS} FROM work_items WHERE id = 'wki_2')"), [], |row| row.get(0))
378 .unwrap();
379 let mut links: Vec<(String, bool)> = serde_json::from_str::<Vec<serde_json::Value>>(&links)
380 .unwrap()
381 .into_iter()
382 .map(|link| (link["path"].as_str().unwrap().to_owned(), link["deleted"].as_i64() == Some(1)))
383 .collect();
384 links.sort();
385 links
386 }
387
388 #[test]
389 fn a_deleted_repository_is_marked_on_its_links_until_it_is_restored() {
390 let db = acme();
391 assert_eq!(deleted_links(&db), [("acme/hello".to_owned(), false), ("acme/web".to_owned(), false)]);
392 // repo.deleted: its path is kept, and its links say it is deleted.
393 db.execute(MARK_DELETED, rusqlite::params!["rep_hello", "2026-10-15T00:00:00Z"]).unwrap();
394 assert_eq!(deleted_links(&db), [("acme/hello".to_owned(), true), ("acme/web".to_owned(), false)]);
395 // Marked again (the job for those deleted before), it keeps when.
396 db.execute(MARK_DELETED, rusqlite::params!["rep_hello", "2026-10-18T00:00:00Z"]).unwrap();
397 let at: String = db.query_row("SELECT deleted_at FROM repo_workspaces WHERE repo_id = 'rep_hello'", [], |row| row.get(0)).unwrap();
398 assert_eq!(at, "2026-10-15T00:00:00Z");
399 // repo.restored records it where it is: no longer deleted.
400 db.execute(REMEMBER_REPO, rusqlite::params!["rep_hello", "acme", "acme/hello", 0]).unwrap();
401 assert_eq!(deleted_links(&db), [("acme/hello".to_owned(), false), ("acme/web".to_owned(), false)]);
402 // repo.purged drops its row: the link keeps its path, and is not marked.
403 db.execute(MARK_DELETED, rusqlite::params!["rep_hello", "2026-10-18T00:00:00Z"]).unwrap();
404 db.execute("DELETE FROM repo_workspaces WHERE repo_id = ?1", ["rep_hello"]).unwrap();
405 assert_eq!(deleted_links(&db), [("acme/hello".to_owned(), false), ("acme/web".to_owned(), false)]);
406 }
407
408 #[test]
409 fn notes_do_not_name_repositories_the_reader_may_not_read() {
410 let hidden = vec!["acme/secret".to_owned()];
411 assert_eq!(redact_paths("added acme/secret as a target", &hidden), "added a private repository as a target");
412 assert_eq!(redact_paths("closed this by merging Acme/Secret#5", &hidden), "closed this by merging a private repository");
413 assert_eq!(
414 redact_paths("landed acme/secret@abc1234; still to land in acme/web", &hidden),
415 "landed a private repository; still to land in acme/web"
416 );
417 // Only the name itself, never a longer one it starts or ends.
418 assert_eq!(redact_paths("added acme/secret-tools as a target", &hidden), "added acme/secret-tools as a target");
419 assert_eq!(redact_paths("added xacme/secret", &hidden), "added xacme/secret");
420 }
421}