Skip to content

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

479 linesCodeBlameRaw
1//! Notifications: a person's inbox, its threads, their subscriptions to
2//! work items and pull requests, and how they watch repositories. The events
3//! service keeps all of it (`g1t_contracts::inbox`); this is its public
4//! shape, which follows the inbox's own: a thread is one item about one
5//! thing, brought back to the top as things happen to it.
6//!
7//! Every operation here is the person's own: a personal access token or a
8//! session, never a workspace's token or a run's token (which acts as
9//! g1t, and g1t is never told anything).
10
11use g1t_contracts::inbox::*;
12use g1t_contracts::repos::{GetArgs, Repo, RepoPath};
13use g1t_contracts::time::{parse_rfc3339, rfc3339};
14use g1t_contracts::work::{WorkItemDetail, WorkItemViewArgs};
15use g1t_contracts::{FailureCode, Outcome, PrincipalKind, User, Viewer};
16use serde_json::{Value, json};
17use worker::Result;
18
19use crate::operations::{Op, Services};
20
21fn failed(code: FailureCode, message: &str) -> Result<Outcome<Value>> {
22 Ok(Outcome::fail(code, message))
23}
24
25fn ok<T: serde::Serialize>(value: &T) -> Result<Outcome<Value>> {
26 Ok(Outcome::Ok(serde_json::to_value(value)?))
27}
28
29const NO_THREAD: &str = "No such notification thread.";
30const NO_SUBJECT: &str = "Name a thread by id, a pull request by repo and number, or a work item by workspace and number.";
31
32fn text(input: &Value, key: &str) -> Option<String> {
33 input[key].as_str().map(str::trim).filter(|value| !value.is_empty()).map(str::to_owned)
34}
35
36/// A yes or no, given as a boolean or as `true`/`false` (a query string).
37fn flag(input: &Value, key: &str) -> Option<bool> {
38 match &input[key] {
39 Value::Bool(value) => Some(*value),
40 Value::String(text) => match text.trim() {
41 "true" | "1" => Some(true),
42 "false" | "0" => Some(false),
43 _ => None,
44 },
45 _ => None,
46 }
47}
48
49fn whole(input: &Value, key: &str) -> Option<u32> {
50 match &input[key] {
51 Value::Number(number) => number.as_u64().and_then(|n| u32::try_from(n).ok()),
52 Value::String(digits) => digits.trim().parse().ok(),
53 _ => None,
54 }
55}
56
57/// A time given as RFC 3339, as the inbox stores times (to the
58/// millisecond, in UTC), or why it is not one.
59pub(crate) fn instant(input: &Value, key: &str) -> std::result::Result<Option<String>, String> {
60 match text(input, key) {
61 None => Ok(None),
62 Some(given) => parse_rfc3339(&given)
63 .map(|ms| Some(rfc3339(ms)))
64 .ok_or_else(|| format!("{key} is a time like 2026-10-07T12:00:00Z, not {given}.")),
65 }
66}
67
68/// The repository `repo` names, if the viewer may read it.
69async fn readable_repo(services: &Services, viewer: &Viewer, input: &Value) -> Result<Option<Outcome<Repo>>> {
70 let Some(path) = crate::operations::repo_path(input) else {
71 return Ok(None);
72 };
73 let found: Outcome<Repo> = g1t_kit::call(
74 &services.repos,
75 "get",
76 &GetArgs {
77 path: RepoPath {
78 namespace: path.namespace,
79 name: path.name,
80 },
81 viewer: viewer.clone(),
82 },
83 )
84 .await?;
85 Ok(Some(found))
86}
87
88/// What `list_notifications` reads from its input.
89pub(crate) fn list_args(viewer: &Viewer, input: &Value) -> std::result::Result<ListInboxArgs, String> {
90 let view = match text(input, "view") {
91 None => InboxView::Inbox,
92 Some(view) => InboxView::parse(&view).ok_or_else(|| format!("view is inbox, saved or done, not {view}."))?,
93 };
94 let reason = match text(input, "reason") {
95 None => None,
96 Some(reason) => Some(Reason::parse(&reason).ok_or_else(|| {
97 format!(
98 "{reason} is not a reason. Give one of {}.",
99 Reason::ALL.map(Reason::as_str).join(", ")
100 )
101 })?),
102 };
103 let severity = match text(input, "severity") {
104 None => None,
105 Some(severity) => Some(
106 Severity::parse(&severity).ok_or_else(|| format!("severity is error, warning, success or info, not {severity}."))?,
107 ),
108 };
109 // As it is elsewhere: what is unread, unless all is asked for. Saved
110 // and done are lists of their own, read or not.
111 let all = flag(input, "all").unwrap_or(false);
112 let unread = flag(input, "unread").unwrap_or(view == InboxView::Inbox && !all);
113 Ok(ListInboxArgs {
114 viewer: viewer.clone(),
115 view,
116 severity,
117 reason,
118 participating: flag(input, "participating").unwrap_or(false),
119 repo_id: None,
120 unread,
121 since: instant(input, "since")?,
122 updated_before: instant(input, "before")?,
123 before: text(input, "cursor"),
124 limit: whole(input, "per_page").map(|n| n.clamp(1, MAX_INBOX_PAGE)),
125 })
126}
127
128/// How a person watches a repository, as the API shows it: its level and
129/// kinds, and the plain answers to whether they get its activity and
130/// whether they ignore it.
131pub(crate) fn watching_json(watching: &Watching, repo: Option<&str>) -> Value {
132 json!({
133 "repo": repo.map(str::to_owned).or_else(|| watching.repo.clone()),
134 "level": watching.level,
135 "events": watching.events,
136 "subscribed": matches!(watching.level, WatchLevel::All | WatchLevel::Custom),
137 "ignored": watching.level == WatchLevel::Ignore,
138 "updated_at": watching.updated_at,
139 })
140}
141
142/// The level `set_repo_subscription` asks for: `level` (with `events`), or
143/// the yes-or-no of `subscribed` and `ignored`.
144pub(crate) fn watch_level(input: &Value) -> std::result::Result<(WatchLevel, Vec<String>), String> {
145 let events: Vec<String> = input["events"]
146 .as_array()
147 .map(|events| events.iter().filter_map(|event| event.as_str().map(str::to_owned)).collect())
148 .unwrap_or_default();
149 if let Some(level) = text(input, "level") {
150 let level = WatchLevel::parse(&level)
151 .ok_or_else(|| format!("level is participating, all, ignore or custom, not {level}."))?;
152 if level == WatchLevel::Custom {
153 let unknown: Vec<&String> = events
154 .iter()
155 .filter(|event| !WATCH_EVENTS.contains(&event.trim().to_lowercase().as_str()))
156 .collect();
157 if let Some(event) = unknown.first() {
158 return Err(format!("{event} is not something to watch. Give some of {}.", WATCH_EVENTS.join(", ")));
159 }
160 if events.is_empty() {
161 return Err(format!("A custom watch needs events: some of {}.", WATCH_EVENTS.join(", ")));
162 }
163 }
164 return Ok((level, events));
165 }
166 Ok(match (flag(input, "ignored"), flag(input, "subscribed")) {
167 (Some(true), _) => (WatchLevel::Ignore, Vec::new()),
168 (_, Some(false)) => (WatchLevel::Participating, Vec::new()),
169 _ => (WatchLevel::All, Vec::new()),
170 })
171}
172
173/// Which work item or pull request a subscription call names: a thread's
174/// id, a repository and a pull request's number, or a workspace and a work
175/// item's number. Checks the viewer can read the repository, or see the
176/// work item.
177async fn subscription_args(services: &Services, viewer: &Viewer, input: &Value) -> Result<Outcome<SubscriptionArgs>> {
178 if let Some(id) = text(input, "id") {
179 return Ok(Outcome::Ok(SubscriptionArgs {
180 viewer: viewer.clone(),
181 id: Some(id),
182 ..SubscriptionArgs::default()
183 }));
184 }
185 if let (Some(workspace), None, Some(number)) = (text(input, "workspace"), text(input, "repo"), whole(input, "number")) {
186 let asked = WorkItemViewArgs { workspace: workspace.to_lowercase(), number, viewer: viewer.clone() };
187 let found: Outcome<WorkItemDetail> = g1t_kit::call(&services.work, "get_work_item", &asked).await?;
188 return Ok(match found {
189 Outcome::Ok(detail) => Outcome::Ok(SubscriptionArgs {
190 viewer: viewer.clone(),
191 item_id: Some(detail.item.id),
192 number: Some(number),
193 ..SubscriptionArgs::default()
194 }),
195 Outcome::Fail(failure) => Outcome::Fail(failure),
196 });
197 }
198 let (Some(found), Some(number)) = (readable_repo(services, viewer, input).await?, whole(input, "number")) else {
199 return Ok(Outcome::fail(FailureCode::Invalid, NO_SUBJECT));
200 };
201 Ok(match found {
202 Outcome::Ok(repo) => Outcome::Ok(SubscriptionArgs {
203 viewer: viewer.clone(),
204 id: None,
205 repo_id: Some(repo.id),
206 number: Some(number),
207 item_id: None,
208 }),
209 Outcome::Fail(failure) => Outcome::Fail(failure),
210 })
211}
212
213/// The viewer as a person: notifications are nobody else's.
214fn person(viewer: &Viewer) -> std::result::Result<&User, &'static str> {
215 match viewer {
216 Some(user) if user.kind == PrincipalKind::User => Ok(user),
217 Some(_) => Err("Notifications are a person's own: use a personal access token, not a workspace's or an agent's."),
218 None => Err("This needs a g1t access token."),
219 }
220}
221
222/// One thread, after a change, or not found.
223async fn thread_after(services: &Services, viewer: &Viewer, id: String) -> Result<Outcome<Value>> {
224 let thread: Option<InboxThread> = g1t_kit::call(&services.events, "inbox_thread", &ThreadArgs { viewer: viewer.clone(), id }).await?;
225 match thread {
226 Some(thread) => ok(&thread),
227 None => failed(FailureCode::NotFound, NO_THREAD),
228 }
229}
230
231/// Marks one of the person's threads, then returns it as it is now.
232async fn mark_one(services: &Services, viewer: &Viewer, user: &User, input: &Value, mark: InboxMark, until: Option<String>) -> Result<Outcome<Value>> {
233 let Some(id) = text(input, "id") else {
234 return failed(FailureCode::Invalid, "Give the thread's id.");
235 };
236 // Only a thread the person can still see; one about a repository they
237 // lost is gone.
238 if let Outcome::Fail(failure) = thread_after(services, viewer, id.clone()).await? {
239 return Ok(Outcome::Fail(failure));
240 }
241 let _: u32 = g1t_kit::call(
242 &services.events,
243 "inbox_mark",
244 &MarkInboxArgs {
245 username: user.username.clone(),
246 mark,
247 ids: vec![id.clone()],
248 all: false,
249 severity: None,
250 repo_id: None,
251 last_read_at: None,
252 until,
253 },
254 )
255 .await?;
256 thread_after(services, viewer, id).await
257}
258
259/// Runs one of the notification operations.
260pub async fn run(op: Op, services: &Services, viewer: &Viewer, input: &Value) -> Result<Outcome<Value>> {
261 let user = match person(viewer) {
262 Ok(user) => user,
263 Err(message) => return failed(FailureCode::Forbidden, message),
264 };
265 let username = user.username.to_lowercase();
266 match op {
267 Op::ListNotifications => {
268 let mut args = match list_args(viewer, input) {
269 Ok(args) => args,
270 Err(message) => return failed(FailureCode::Invalid, &message),
271 };
272 if let Some(found) = readable_repo(services, viewer, input).await? {
273 match found {
274 Outcome::Ok(repo) => args.repo_id = Some(repo.id),
275 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
276 }
277 }
278 let page: InboxPage = g1t_kit::call(&services.events, "inbox_list", &args).await?;
279 ok(&page)
280 }
281 Op::MarkNotificationsRead => {
282 let last_read_at = match instant(input, "last_read_at") {
283 Ok(at) => at.unwrap_or_else(|| rfc3339(g1t_kit::now_ms())),
284 Err(message) => return failed(FailureCode::Invalid, &message),
285 };
286 let repo_id = match readable_repo(services, viewer, input).await? {
287 Some(Outcome::Ok(repo)) => Some(repo.id),
288 Some(Outcome::Fail(failure)) => return Ok(Outcome::Fail(failure)),
289 None => None,
290 };
291 let mark = if flag(input, "read") == Some(false) { InboxMark::Unread } else { InboxMark::Read };
292 let marked: u32 = g1t_kit::call(
293 &services.events,
294 "inbox_mark",
295 &MarkInboxArgs {
296 username,
297 mark,
298 ids: Vec::new(),
299 all: true,
300 severity: None,
301 repo_id,
302 last_read_at: Some(last_read_at.clone()),
303 until: None,
304 },
305 )
306 .await?;
307 ok(&json!({ "marked": marked, "last_read_at": last_read_at }))
308 }
309 Op::GetNotificationThread => match text(input, "id") {
310 Some(id) => thread_after(services, viewer, id).await,
311 None => failed(FailureCode::Invalid, "Give the thread's id."),
312 },
313 Op::MarkThreadRead => {
314 let mark = if flag(input, "read") == Some(false) { InboxMark::Unread } else { InboxMark::Read };
315 mark_one(services, viewer, user, input, mark, None).await
316 }
317 Op::MarkThreadDone => {
318 let mark = if flag(input, "done") == Some(false) { InboxMark::Undone } else { InboxMark::Done };
319 mark_one(services, viewer, user, input, mark, None).await
320 }
321 Op::SaveThread => {
322 let mark = if flag(input, "saved") == Some(false) { InboxMark::Unsave } else { InboxMark::Save };
323 mark_one(services, viewer, user, input, mark, None).await
324 }
325 Op::SnoozeThread => {
326 let until = match instant(input, "until") {
327 Ok(until) => until,
328 Err(message) => return failed(FailureCode::Invalid, &message),
329 };
330 match until {
331 Some(until) if until.as_str() <= rfc3339(g1t_kit::now_ms()).as_str() => {
332 failed(FailureCode::Invalid, "until is a time to come; leave it out to bring the thread back now.")
333 }
334 Some(until) => mark_one(services, viewer, user, input, InboxMark::Snooze, Some(until)).await,
335 None => mark_one(services, viewer, user, input, InboxMark::Unsnooze, None).await,
336 }
337 }
338 Op::GetThreadSubscription | Op::SetThreadSubscription | Op::DeleteThreadSubscription => {
339 let on = match subscription_args(services, viewer, input).await? {
340 Outcome::Ok(on) => on,
341 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
342 };
343 let found: Option<ThreadSubscription> = match op {
344 Op::GetThreadSubscription => g1t_kit::call(&services.events, "inbox_subscription", &on).await?,
345 _ => {
346 let (subscribed, ignored) = match op {
347 Op::SetThreadSubscription => (Some(flag(input, "subscribed").unwrap_or(true)), flag(input, "ignored").unwrap_or(false)),
348 _ => (Some(false), false),
349 };
350 g1t_kit::call(&services.events, "inbox_subscribe", &SubscribeArgs { on, subscribed, ignored }).await?
351 }
352 };
353 match found {
354 Some(subscription) => ok(&subscription),
355 None => failed(FailureCode::NotFound, "No such work item or pull request, or it is a thread nobody subscribes to."),
356 }
357 }
358 Op::GetRepoSubscription | Op::SetRepoSubscription | Op::DeleteRepoSubscription => {
359 let repo = match readable_repo(services, viewer, input).await? {
360 Some(Outcome::Ok(repo)) => repo,
361 Some(Outcome::Fail(failure)) => return Ok(Outcome::Fail(failure)),
362 None => return failed(FailureCode::Invalid, "Give the repository as \"owner/name\"."),
363 };
364 let path = format!("{}/{}", repo.namespace, repo.name);
365 let watching: Watching = match op {
366 Op::GetRepoSubscription => {
367 g1t_kit::call(&services.events, "inbox_watching", &WatchingArgs { username, repo_id: repo.id }).await?
368 }
369 _ => {
370 let (level, events) = match op {
371 Op::SetRepoSubscription => match watch_level(input) {
372 Ok((level, events)) => (Some(level), events),
373 Err(message) => return failed(FailureCode::Invalid, &message),
374 },
375 _ => (None, Vec::new()),
376 };
377 g1t_kit::call(
378 &services.events,
379 "inbox_watch",
380 &WatchArgs {
381 username,
382 repo_id: repo.id,
383 repo: Some(path.clone()),
384 level,
385 events,
386 },
387 )
388 .await?
389 }
390 };
391 Ok(Outcome::Ok(watching_json(&watching, Some(&path))))
392 }
393 Op::ListWatchedRepos => {
394 let watched: Vec<Watching> = g1t_kit::call(&services.events, "inbox_watched", &InboxCountsArgs { username }).await?;
395 Ok(Outcome::Ok(Value::Array(watched.iter().map(|watching| watching_json(watching, None)).collect())))
396 }
397 _ => failed(FailureCode::NotFound, "No such endpoint."),
398 }
399}
400
401#[cfg(test)]
402mod tests {
403 use super::*;
404
405 fn viewer() -> Viewer {
406 Some(User {
407 id: "usr_1".into(),
408 username: "ana".into(),
409 ..User::default()
410 })
411 }
412
413 #[test]
414 fn a_list_shows_what_is_unread_unless_all_is_asked_for() {
415 let args = list_args(&viewer(), &json!({})).unwrap();
416 assert!(args.unread);
417 assert_eq!(args.view, InboxView::Inbox);
418 let args = list_args(&viewer(), &json!({ "all": "true" })).unwrap();
419 assert!(!args.unread);
420 // Saved and done show everything in them.
421 assert!(!list_args(&viewer(), &json!({ "view": "done" })).unwrap().unread);
422 let args = list_args(
423 &viewer(),
424 &json!({ "reason": "review_requested", "participating": true, "since": "2026-10-01T00:00:00Z", "before": "2026-10-07T00:00:00Z", "cursor": "ntf_9", "per_page": "500" }),
425 )
426 .unwrap();
427 assert_eq!(args.reason, Some(Reason::ReviewRequested));
428 assert!(args.participating);
429 assert_eq!(args.since.as_deref(), Some("2026-10-01T00:00:00.000Z"));
430 assert_eq!(args.updated_before.as_deref(), Some("2026-10-07T00:00:00.000Z"));
431 assert_eq!(args.before.as_deref(), Some("ntf_9"));
432 assert_eq!(args.limit, Some(MAX_INBOX_PAGE));
433 }
434
435 #[test]
436 fn a_list_names_what_it_cannot_read() {
437 assert!(list_args(&viewer(), &json!({ "reason": "gossip" })).unwrap_err().contains("not a reason"));
438 assert!(list_args(&viewer(), &json!({ "view": "archive" })).unwrap_err().contains("inbox, saved or done"));
439 assert!(list_args(&viewer(), &json!({ "since": "yesterday" })).unwrap_err().contains("since is a time"));
440 }
441
442 #[test]
443 fn watching_is_a_level_or_a_yes_or_no() {
444 assert_eq!(watch_level(&json!({ "level": "all" })).unwrap().0, WatchLevel::All);
445 assert_eq!(watch_level(&json!({ "ignored": true })).unwrap().0, WatchLevel::Ignore);
446 assert_eq!(watch_level(&json!({ "subscribed": false })).unwrap().0, WatchLevel::Participating);
447 assert_eq!(watch_level(&json!({})).unwrap().0, WatchLevel::All);
448 let (level, events) = watch_level(&json!({ "level": "custom", "events": ["pulls", "deployments"] })).unwrap();
449 assert_eq!((level, events.len()), (WatchLevel::Custom, 2));
450 assert!(watch_level(&json!({ "level": "custom" })).unwrap_err().contains("needs events"));
451 assert!(watch_level(&json!({ "level": "custom", "events": ["releases"] })).unwrap_err().contains("releases"));
452 assert!(watch_level(&json!({ "level": "loud" })).is_err());
453 }
454
455 #[test]
456 fn watching_says_plainly_whether_activity_comes() {
457 let custom = Watching {
458 repo_id: "rep_1".into(),
459 repo: None,
460 level: WatchLevel::Custom,
461 events: vec!["pulls".into()],
462 updated_at: None,
463 };
464 let shown = watching_json(&custom, Some("acme/rocket"));
465 assert_eq!(shown["repo"], "acme/rocket");
466 assert_eq!(shown["level"], "custom");
467 assert_eq!(shown["subscribed"], true);
468 assert_eq!(shown["ignored"], false);
469 assert!(shown.get("repo_id").is_none());
470 }
471
472 #[test]
473 fn notifications_are_a_persons_own() {
474 assert!(person(&viewer()).is_ok());
475 let workspace = Some(User { kind: PrincipalKind::Workspace, ..User::default() });
476 assert!(person(&workspace).unwrap_err().contains("person's own"));
477 assert!(person(&None).is_err());
478 }
479}