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