Sign in

flagon-io/g1t

public

g1t/services/work/src/authored.rs

427 lines15,276 bytesCodeBlame
1//! A person's work, for their profile: the issues and pull requests they
2//! opened, and those g1t opened for them (`requested_by`, see Pull::owner),
3//! newest first, a page at a time.
4//!
5//! Only work on repositories the viewer may read is ever shown, counted or
6//! named. Which those are is the repos service's decision (`readable`, the
7//! same check as opening the repository), made once per request over every
8//! repository the person has worked in; every query here is then confined
9//! to that set. A private repository's titles, numbers and even its
10//! existence never reach anyone who could not open it.
11
12use std::collections::HashMap;
13
14use g1t_contracts::identity::UsernameArgs;
15use g1t_contracts::repos::{ReadableArgs, Repo, RepoPath, MAX_READABLE};
16use g1t_contracts::work::*;
17use g1t_contracts::{FailureCode, Outcome, Viewer};
18use serde::Deserialize;
19use worker::Result;
20use worker::wasm_bindgen::JsValue;
21
22use crate::Work;
23
24#[derive(Deserialize)]
25struct ItemRow {
26 kind: AuthoredKind,
27 id: String,
28 repo_id: String,
29 number: u32,
30 title: String,
31 state: State,
32 reason: Option<IssueReason>,
33 status: Option<PullStatus>,
34 created_at: String,
35 updated_at: String,
36 merged_at: Option<String>,
37}
38
39#[derive(Deserialize)]
40struct RepoCount {
41 repo_id: String,
42 n: u32,
43}
44
45/// Where a page ends: the sort key and id of its last item.
46fn cursor(sort: AuthoredSort, row: &ItemRow) -> String {
47 let key = match sort {
48 AuthoredSort::Updated => &row.updated_at,
49 AuthoredSort::Created | AuthoredSort::Oldest => &row.created_at,
50 };
51 format!("{key}|{}", row.id)
52}
53
54/// A cursor read back, or none if it is not one.
55fn parse_cursor(value: &str) -> Option<(&str, &str)> {
56 let (key, id) = value.split_once('|')?;
57 (!key.is_empty() && !id.is_empty() && !id.contains('|')).then_some((key, id))
58}
59
60/// One bound value; made a `JsValue` only when bound, so the SQL can be
61/// built and tested outside a Worker.
62enum Param {
63 Text(String),
64 Number(u32),
65}
66
67impl From<&str> for Param {
68 fn from(value: &str) -> Self {
69 Param::Text(value.to_owned())
70 }
71}
72
73impl From<u32> for Param {
74 fn from(value: u32) -> Self {
75 Param::Number(value)
76 }
77}
78
79/// Parameters bound by number, so each can be used more than once.
80struct Params(Vec<Param>);
81
82impl Params {
83 fn push(&mut self, value: impl Into<Param>) -> String {
84 self.0.push(value.into());
85 format!("?{}", self.0.len())
86 }
87
88 fn values(&self) -> Vec<JsValue> {
89 self.0
90 .iter()
91 .map(|param| match param {
92 Param::Text(text) => JsValue::from(text.as_str()),
93 Param::Number(n) => JsValue::from(*n),
94 })
95 .collect()
96 }
97}
98
99/// The work SQL for one page: the person's issues and pull requests in
100/// `visible` as one list, filtered, ordered and cut to `limit + 1` rows.
101fn page_sql(a: &ByAuthorArgs, params: &mut Params, author: &str, visible: &str, limit: u32) -> String {
102 let author = params.push(author);
103 let visible = params.push(visible);
104 let mut conditions = Vec::new();
105 if let Some(kind) = a.kind {
106 conditions.push(format!(
107 "kind = {}",
108 params.push(match kind {
109 AuthoredKind::Issue => "issue",
110 AuthoredKind::Pull => "pull",
111 })
112 ));
113 }
114 match a.state {
115 Some(AuthoredState::Open) => conditions.push("state = 'open'".to_owned()),
116 Some(AuthoredState::Closed) => conditions.push("state = 'closed'".to_owned()),
117 Some(AuthoredState::Merged) => conditions.push("status = 'merged'".to_owned()),
118 None => {}
119 }
120 let (key, direction, beyond) = match a.sort {
121 AuthoredSort::Created => ("created_at", "DESC", "<"),
122 AuthoredSort::Updated => ("updated_at", "DESC", "<"),
123 AuthoredSort::Oldest => ("created_at", "ASC", ">"),
124 };
125 if let Some((after_key, after_id)) = a.before.as_deref().and_then(parse_cursor) {
126 let after_key = params.push(after_key);
127 let after_id = params.push(after_id);
128 conditions.push(format!(
129 "({key} {beyond} {after_key} OR ({key} = {after_key} AND id {beyond} {after_id}))"
130 ));
131 }
132 let filter = if conditions.is_empty() {
133 String::new()
134 } else {
135 format!("WHERE {}", conditions.join(" AND "))
136 };
137 let limit = params.push(limit + 1);
138 format!(
139 "SELECT * FROM (
140 SELECT 'issue' AS kind, id, repo_id, number, title, state, reason,
141 NULL AS status, created_at, updated_at, NULL AS merged_at
142 FROM issues
143 WHERE COALESCE(requested_by_id, author_id) = {author} AND repo_id IN (SELECT value FROM json_each({visible}))
144 UNION ALL
145 SELECT 'pull' AS kind, id, repo_id, number, title,
146 CASE WHEN status IN ('draft', 'open') THEN 'open' ELSE 'closed' END AS state,
147 NULL AS reason, status, created_at, updated_at, merged_at
148 FROM pulls
149 WHERE COALESCE(requested_by_id, author_id) = {author} AND repo_id IN (SELECT value FROM json_each({visible}))
150 ) AS work
151 {filter}
152 ORDER BY {key} {direction}, id {direction}
153 LIMIT {limit}"
154 )
155}
156
157const COUNTS_SQL: &str = "SELECT
158 (SELECT count(*) FROM pulls WHERE COALESCE(requested_by_id, author_id) = ?1
159 AND repo_id IN (SELECT value FROM json_each(?2)) AND status = 'merged') AS pulls_merged,
160 (SELECT count(*) FROM pulls WHERE COALESCE(requested_by_id, author_id) = ?1
161 AND repo_id IN (SELECT value FROM json_each(?2)) AND status IN ('draft', 'open')) AS pulls_open,
162 (SELECT count(*) FROM pulls WHERE COALESCE(requested_by_id, author_id) = ?1
163 AND repo_id IN (SELECT value FROM json_each(?2))) AS pulls,
164 (SELECT count(*) FROM issues WHERE COALESCE(requested_by_id, author_id) = ?1
165 AND repo_id IN (SELECT value FROM json_each(?2))) AS issues,
166 (SELECT count(*) FROM issues WHERE COALESCE(requested_by_id, author_id) = ?1
167 AND repo_id IN (SELECT value FROM json_each(?2)) AND state = 'open') AS issues_open";
168
169#[derive(Deserialize)]
170#[serde(rename_all = "snake_case")]
171struct CountsRow {
172 pulls_merged: u32,
173 pulls_open: u32,
174 pulls: u32,
175 issues: u32,
176 issues_open: u32,
177}
178
179/// Whether `repo` is the one named `namespace/name`.
180fn is_named(repo: &Repo, name: &str) -> bool {
181 name.split_once('/').is_some_and(|(namespace, name)| {
182 repo.namespace.eq_ignore_ascii_case(namespace) && repo.name.eq_ignore_ascii_case(name)
183 })
184}
185
186impl Work {
187 pub(crate) async fn by_author(&self, a: ByAuthorArgs) -> Result<Outcome<Authored>> {
188 let person: Viewer = g1t_kit::call(
189 &self.identity,
190 "user_by_username",
191 &UsernameArgs {
192 username: a.username.trim().to_lowercase(),
193 },
194 )
195 .await?;
196 let Some(person) = person else {
197 return Ok(Outcome::fail(FailureCode::NotFound, "There is no such account."));
198 };
199
200 // Every repository they have opened work in, busiest first...
201 let touched = self
202 .db
203 .prepare(
204 "SELECT repo_id, count(*) AS n FROM (
205 SELECT repo_id FROM issues WHERE COALESCE(requested_by_id, author_id) = ?1
206 UNION ALL SELECT repo_id FROM pulls WHERE COALESCE(requested_by_id, author_id) = ?1
207 ) GROUP BY repo_id ORDER BY n DESC LIMIT ?2",
208 )
209 .bind(&[person.id.as_str().into(), (MAX_READABLE as u32).into()])?
210 .all()
211 .await?
212 .results::<RepoCount>()?;
213 if touched.is_empty() {
214 return Ok(Outcome::Ok(Authored::default()));
215 }
216 // ...and of those, the ones the viewer may read.
217 let readable: Vec<Repo> = g1t_kit::call(
218 &self.repos,
219 "readable",
220 &ReadableArgs {
221 ids: touched.iter().map(|row| row.repo_id.clone()).collect(),
222 viewer: a.viewer.clone(),
223 },
224 )
225 .await?;
226 let by_id: HashMap<&str, &Repo> = readable.iter().map(|repo| (repo.id.as_str(), repo)).collect();
227 let repos: Vec<AuthoredRepo> = touched
228 .iter()
229 .filter_map(|row| {
230 by_id.get(row.repo_id.as_str()).map(|repo| AuthoredRepo {
231 repo: RepoPath {
232 namespace: repo.namespace.clone(),
233 name: repo.name.clone(),
234 },
235 count: row.n,
236 })
237 })
238 .collect();
239 if readable.is_empty() {
240 return Ok(Outcome::Ok(Authored::default()));
241 }
242
243 let all_ids: Vec<&str> = readable.iter().map(|repo| repo.id.as_str()).collect();
244 let all = serde_json::to_string(&all_ids)?;
245 let counts = self
246 .db
247 .prepare(COUNTS_SQL)
248 .bind(&[person.id.as_str().into(), all.as_str().into()])?
249 .first::<CountsRow>(None)
250 .await?
251 .map(|row| AuthoredCounts {
252 pulls_merged: row.pulls_merged,
253 pulls_open: row.pulls_open,
254 pulls: row.pulls,
255 issues: row.issues,
256 issues_open: row.issues_open,
257 })
258 .unwrap_or_default();
259
260 // A repository filter narrows the set; one the viewer cannot read,
261 // or the person never worked in, leaves nothing.
262 let shown = match a.repo.as_deref().map(str::trim).filter(|name| !name.is_empty()) {
263 Some(name) => serde_json::to_string(
264 &readable
265 .iter()
266 .filter(|repo| is_named(repo, name))
267 .map(|repo| repo.id.as_str())
268 .collect::<Vec<_>>(),
269 )?,
270 None => all,
271 };
272 let limit = a.limit.unwrap_or(AUTHORED_PAGE).clamp(1, AUTHORED_PAGE);
273 let mut params = Params(Vec::new());
274 let sql = page_sql(&a, &mut params, &person.id, &shown, limit);
275 let mut rows = self
276 .db
277 .prepare(sql)
278 .bind(&params.values())?
279 .all()
280 .await?
281 .results::<ItemRow>()?;
282 let next = if rows.len() > limit as usize {
283 rows.truncate(limit as usize);
284 rows.last().map(|row| cursor(a.sort, row))
285 } else {
286 None
287 };
288 let items = rows
289 .into_iter()
290 .filter_map(|row| {
291 let repo = by_id.get(row.repo_id.as_str())?;
292 Some(AuthoredItem {
293 kind: row.kind,
294 repo: RepoPath {
295 namespace: repo.namespace.clone(),
296 name: repo.name.clone(),
297 },
298 number: row.number,
299 title: row.title,
300 state: row.state,
301 draft: row.status == Some(PullStatus::Draft),
302 merged: row.status == Some(PullStatus::Merged),
303 status: row.status,
304 reason: row.reason,
305 created_at: row.created_at,
306 updated_at: row.updated_at,
307 merged_at: row.merged_at,
308 })
309 })
310 .collect();
311 Ok(Outcome::Ok(Authored {
312 items,
313 next,
314 counts,
315 repos,
316 }))
317 }
318}
319
320#[cfg(test)]
321mod tests {
322 use super::*;
323
324 fn args() -> ByAuthorArgs {
325 ByAuthorArgs {
326 username: "ada".into(),
327 viewer: None,
328 kind: None,
329 state: None,
330 repo: None,
331 sort: AuthoredSort::Created,
332 before: None,
333 limit: None,
334 }
335 }
336
337 fn row(created: &str, updated: &str, id: &str) -> ItemRow {
338 ItemRow {
339 kind: AuthoredKind::Pull,
340 id: id.into(),
341 repo_id: "rep_1".into(),
342 number: 1,
343 title: "t".into(),
344 state: State::Open,
345 reason: None,
346 status: Some(PullStatus::Open),
347 created_at: created.into(),
348 updated_at: updated.into(),
349 merged_at: None,
350 }
351 }
352
353 #[test]
354 fn every_query_is_confined_to_the_readable_set() {
355 let mut params = Params(Vec::new());
356 let sql = page_sql(&args(), &mut params, "usr_1", "[]", 25);
357 assert_eq!(sql.matches("json_each(?2)").count(), 2);
358 assert_eq!(sql.matches("COALESCE(requested_by_id, author_id) = ?1").count(), 2);
359 assert_eq!(COUNTS_SQL.matches("json_each(?2)").count(), 5);
360 assert!(sql.contains("LIMIT ?3"));
361 assert!(sql.contains("ORDER BY created_at DESC, id DESC"));
362 }
363
364 #[test]
365 fn filters_by_kind_state_and_page() {
366 let mut params = Params(Vec::new());
367 let a = ByAuthorArgs {
368 kind: Some(AuthoredKind::Pull),
369 state: Some(AuthoredState::Merged),
370 sort: AuthoredSort::Oldest,
371 before: Some("2026-10-01T00:00:00.000Z|pul_9".into()),
372 ..args()
373 };
374 let sql = page_sql(&a, &mut params, "usr_1", "[]", 10);
375 assert!(sql.contains("kind = ?3"));
376 assert!(sql.contains("status = 'merged'"));
377 assert!(sql.contains("(created_at > ?4 OR (created_at = ?4 AND id > ?5))"));
378 assert!(sql.contains("ORDER BY created_at ASC, id ASC"));
379 assert!(sql.contains("LIMIT ?6"));
380 assert_eq!(params.0.len(), 6);
381 }
382
383 #[test]
384 fn ignores_a_cursor_that_is_not_one() {
385 let mut params = Params(Vec::new());
386 let a = ByAuthorArgs {
387 before: Some("nonsense".into()),
388 ..args()
389 };
390 let sql = page_sql(&a, &mut params, "usr_1", "[]", 10);
391 assert!(!sql.contains(" OR ("));
392 assert_eq!(parse_cursor("a|b|c"), None);
393 assert_eq!(parse_cursor("|b"), None);
394 assert_eq!(parse_cursor("k|i"), Some(("k", "i")));
395 }
396
397 #[test]
398 fn a_cursor_names_the_sort_key() {
399 let r = row("2026-01-01T00:00:00.000Z", "2026-02-01T00:00:00.000Z", "pul_1");
400 assert_eq!(cursor(AuthoredSort::Created, &r), "2026-01-01T00:00:00.000Z|pul_1");
401 assert_eq!(cursor(AuthoredSort::Updated, &r), "2026-02-01T00:00:00.000Z|pul_1");
402 assert_eq!(parse_cursor(&cursor(AuthoredSort::Oldest, &r)), Some(("2026-01-01T00:00:00.000Z", "pul_1")));
403 }
404
405 #[test]
406 fn names_a_repository_by_its_path() {
407 let repo = Repo {
408 id: "rep_1".into(),
409 namespace: "acme".into(),
410 name: "Rocket".into(),
411 description: None,
412 is_private: false,
413 owner_id: "usr_1".into(),
414 default_branch: "main".into(),
415 fork_of: None,
416 protected: false,
417 created_at: String::new(),
418 topics: Vec::new(),
419 website: None,
420 archived_at: None,
421 };
422 assert!(is_named(&repo, "acme/rocket"));
423 assert!(is_named(&repo, "ACME/Rocket"));
424 assert!(!is_named(&repo, "acme"));
425 assert!(!is_named(&repo, "other/rocket"));
426 }
427}