flagon-io/g1t

public

1	//! The inbox, kept beside the event log. See `g1t_contracts::inbox`.
2	//!
3	//! As each batch arrives from the bus, the events that tell someone are
4	//! read against what they name (the work service's `inbox_subject`), who
5	//! subscribes to it and who watches its repository (subscriptions.rs), and
6	//! each person told gets their thread about it brought to the top, unread,
7	//! with a line added to its history. Who is told is worked out in
8	//! [`notices`], from the event, its subject and that audience alone:
9	//!
10	//! \| Event \| Who \| Reason \| Severity \|
11	//! \| --- \| --- \| --- \| --- \|
12	//! \| `agent.asked`, `pull.stalled` \| the pull request's owner, and its issue's owner and assignees \| agent \| warning \|
13	//! \| `pull.review_requested` \| the reviewers asked, and the people each team asked tells \| review_requested \| warning \|
14	//! \| `issue.assigned`, `pull.assigned` \| the people newly assigned \| assign \| info \|
15	//! \| `checks.completed`, failed or errored \| the pull request's owner \| ci_activity \| error \|
16	//! \| `workflow.completed`, failed \| the pull request's owner, or whoever pushed \| ci_activity \| error \|
17	//! \| `deployment.failed` \| the pull request's owner, or whoever pushed; watchers \| ci_activity \| error \|
18	//! \| `deployment.succeeded` after a failure \| the same \| ci_activity \| success \|
19	//! \| `review.completed` by g1t \| the pull request's owner \| author \| success, or info for changes asked \|
20	//! \| `pull.ready` for a change g1t made \| whoever asked g1t for it \| author \| success \|
21	//! \| `pull.merged`, `pull.closed`, `issue.closed`, `issue.reopened` \| everyone subscribed \| state_change \| success for a merge, else info \|
22	//! \| `comment.created` \| everyone mentioned, the people of teams mentioned, then everyone subscribed \| mention, team_mention, or why they are subscribed \| info (success for an approval) \|
23	//! \| `issue.opened`, `pull.opened` \| whoever was assigned, asked to review or mentioned in its description (people and teams); watchers \| assign, review_requested, mention, team_mention, subscribed \| info \|
24	//!
25	//! Watchers of a repository at `all` (or `custom`, for the kinds they
26	//! chose) hear of every issue and pull request opened, commented on,
27	//! closed, reopened or merged, and of deployments. Anyone who ignores the
28	//! thread or the repository hears of nothing on it; anyone who
29	//! unsubscribed hears only of what is asked of them.
30	//!
31	//! Nobody is told of what they did themselves, though an outcome they set
32	//! off (checks, a workflow, a deployment, g1t's work) is theirs to hear of.
33	//! g1t is never told. A failure here is logged and the batch goes on: the
34	//! bus never waits on the inbox, so an item can be missed, but nothing else
35	//! is held up.
36
37	use std::collections::{HashMap, HashSet};
38
39	use g1t_contracts::credentials::Principal;
40	use g1t_contracts::events::{Event, WorkspaceRenamed};
41	use g1t_contracts::identity::{AGENT_ID, UsernamesArgs};
42	use g1t_contracts::inbox::*;
43	use g1t_contracts::repos::{PathByIdArgs, ReadableArgs, Repo, RepoPath};
44	use g1t_contracts::time::rfc3339;
45	use g1t_contracts::{new_id, system};
46	use g1t_kit::now_ms;
47	use serde::Deserialize;
48	use worker::wasm_bindgen::JsValue;
49	use worker::{D1Database, D1PreparedStatement, Fetcher, Result};
50
51	use crate::subscriptions::{self, Audience};
52
53	/// Items marked done are kept this long, then removed.
54	pub const DONE_DAYS: u32 = 30;
55	/// No item is kept longer than this, unless it was saved.
56	pub const MAX_DAYS: u32 = 180;
57	/// Unread warnings shown ahead of everything else on the first page.
58	const MAX_RANKED: u32 = 20;
59	const MAX_TITLE: usize = 200;
60	const MAX_BODY: usize = 300;
61
62	/// What the inbox asks about an event before deciding who is told.
63	#[derive(Debug, PartialEq, Eq)]
64	pub struct Wanted {
65	pub repo_id: String,
66	/// The issue or pull request, when the event names one.
67	pub number: Option<u32>,
68	pub comment_id: Option<String>,
69	}
70
71	/// One person to tell, and what.
72	#[derive(Debug, PartialEq, Eq)]
73	pub struct Notice {
74	pub username: String,
75	pub reason: Reason,
76	pub severity: Severity,
77	pub title: String,
78	pub body: String,
79	}
80
81	/// Who did it, as far as is known.
82	#[derive(Debug, Default)]
83	pub struct Actor {
84	pub id: Option<String>,
85	pub username: Option<String>,
86	}
87
88	impl Actor {
89	fn is(&self, person: &Principal) -> bool {
90	self.id.as_deref().is_some_and(\|id\| id == person.id) \|\| self.is_named(&person.username)
91	}
92
93	fn is_named(&self, username: &str) -> bool {
94	self.username.as_deref().is_some_and(\|name\| name.eq_ignore_ascii_case(username))
95	}
96
97	/// A person's name, for a title: never g1t's ids.
98	fn name(&self) -> Option<&str> {
99	self.username.as_deref().filter(\|name\| !is_g1t(name))
100	}
101	}
102
103	pub(crate) fn is_g1t(username: &str) -> bool {
104	username.eq_ignore_ascii_case(system::USERNAME) \|\| username.eq_ignore_ascii_case("g1t-agent")
105	}
106
107	pub(crate) fn is_g1t_id(id: &str) -> bool {
108	system::is_system_id(id) \|\| id == AGENT_ID
109	}
110
111	/// Events that end what an agent was waiting on a person for: it picked
112	/// back up, its head moved, a merge was asked for, or it is over.
113	const RESUMES: [&str; 5] = ["pull.resumed", "pull.updated", "pull.merge_requested", "pull.merged", "pull.closed"];
114
115	/// The issue or pull request an event names, and what to read for it.
116	/// None for events the inbox does not tell anyone of.
117	pub fn wants(event: &Event) -> Option<Wanted> {
118	let data = &event.data;
119	let text = \|key: &str\| data[key].as_str().filter(\|value\| !value.is_empty()).map(str::to_owned);
120	let number = \|key: &str\| data[key].as_u64().and_then(\|n\| u32::try_from(n).ok());
121	let repo_id = text("repoId").or_else(\|\| event.repo_id.clone())?;
122	let on = \|number: Option<u32>, comment_id: Option<String>\| {
123	Some(Wanted {
124	repo_id: repo_id.clone(),
125	number,
126	comment_id,
127	})
128	};
129	match event.kind.as_str() {
130	"agent.asked" \| "pull.stalled" \| "pull.merged" \| "pull.closed" \| "pull.ready" \| "pull.opened"
131	\| "pull.review_requested" \| "pull.assigned" \| "issue.opened" \| "issue.closed" \| "issue.reopened"
132	\| "issue.assigned" => on(Some(number("number")?), None),
133	"checks.completed" => match data["status"].as_str() {
134	Some("failed" \| "errored") => on(Some(number("number")?), None),
135	_ => None,
136	},
137	"review.completed" => match data["verdict"].as_str() {
138	Some("approve" \| "request_changes") => on(Some(number("number")?), None),
139	_ => None,
140	},
141	"workflow.completed" => match data["conclusion"].as_str() {
142	Some("failure") => on(number("pull"), None),
143	_ => None,
144	},
145	"deployment.failed" \| "deployment.succeeded" => on(number("number"), None),
146	// A workflow run's jobs wait for their environment's reviewers.
147	"deployment.review_requested" => on(None, None),
148	"comment.created" => on(Some(number("number")?), Some(text("commentId")?)),
149	// Security alerts are threads of their own, not of an issue.
150	kind if SECURITY_EVENTS.contains(&kind) => on(None, None),
151	_ => None,
152	}
153	}
154
155	/// The security service's events the inbox tells people of: new alerts,
156	/// and push protection bypasses asked for and decided. Fixes and
157	/// dismissals are on the Security page and in webhooks.
158	pub const SECURITY_EVENTS: [&str; 5] = [
159	"secret_scanning_alert.created",
160	"code_scanning_alert.created",
161	"vulnerability_alert.created",
162	"secret_scanning.bypass_requested",
163	"secret_scanning.bypass_reviewed",
164	];
165
166	/// Where an event's items go: which thread, what it is, and where it is.
167	#[derive(Clone, Debug, PartialEq, Eq)]
168	pub struct Thread {
169	pub key: String,
170	pub kind: Option<SubjectKind>,
171	pub number: Option<u32>,
172	pub run_id: Option<String>,
173	/// A path, for a subject with a page of its own.
174	pub link: Option<String>,
175	}
176
177	/// The thread key of an issue or pull request.
178	pub fn numbered_thread(repo_id: &str, number: u32) -> String {
179	format!("{repo_id}#{number}")
180	}
181
182	/// The thread an event's items go to.
183	pub fn thread_of(event: &Event, wanted: &Wanted, subject: Option<&InboxSubject>) -> Thread {
184	let data = &event.data;
185	let text = \|key: &str\| data[key].as_str().filter(\|value\| !value.is_empty()).map(str::to_owned);
186	match event.kind.as_str() {
187	// A project's production, or one pull request's preview.
188	"deployment.failed" \| "deployment.succeeded" => {
189	let which = wanted.number.map_or_else(\|\| "production".to_owned(), \|number\| number.to_string());
190	Thread {
191	key: format!("{}/deploy/{}/{which}", wanted.repo_id, text("projectId").unwrap_or_default()),
192	kind: Some(SubjectKind::Deploy),
193	number: wanted.number,
194	run_id: text("deploymentId"),
195	link: text("path"),
196	}
197	}
198	// One run's deployment to one environment, waiting for review.
199	"deployment.review_requested" => Thread {
200	key: format!(
201	"{}/review/{}/{}",
202	wanted.repo_id,
203	text("runId").unwrap_or_default(),
204	text("environment").unwrap_or_default()
205	),
206	kind: Some(SubjectKind::Run),
207	number: None,
208	run_id: text("runId"),
209	link: text("link"),
210	},
211	// A workflow on a branch: its next failure bumps the same thread.
212	"workflow.completed" if subject.is_none() => {
213	let branch = text("ref").unwrap_or_default();
214	let branch = branch.strip_prefix("refs/heads/").unwrap_or(&branch).to_owned();
215	let workflow = text("path").or_else(\|\| text("workflow")).unwrap_or_default();
216	Thread {
217	key: format!("{}/run/{workflow}@{branch}", wanted.repo_id),
218	kind: Some(SubjectKind::Run),
219	number: None,
220	run_id: text("runId"),
221	link: None,
222	}
223	}
224	// One alert, or one bypass request: its page is the link.
225	kind if SECURITY_EVENTS.contains(&kind) => {
226	let which = text("requestId").or_else(\|\| text("alertId")).unwrap_or_else(\|\| event.id.clone());
227	Thread {
228	key: format!("{}/security/{which}", wanted.repo_id),
229	kind: None,
230	number: None,
231	run_id: None,
232	link: text("link"),
233	}
234	}
235	_ => match wanted.number {
236	Some(number) => Thread {
237	key: numbered_thread(&wanted.repo_id, number),
238	kind: subject.and_then(\|subject\| subject.kind),
239	number: Some(number),
240	run_id: None,
241	link: None,
242	},
243	None => Thread {
244	key: format!("event/{}", event.id),
245	kind: None,
246	number: None,
247	run_id: None,
248	link: None,
249	},
250	},
251	}
252	}
253
254	/// The issue or pull request whose agent threads an event closes: what an
255	/// agent was waiting on a person for is over.
256	pub fn resolves(event: &Event) -> Option<String> {
257	if !RESUMES.contains(&event.kind.as_str()) {
258	return None;
259	}
260	let repo_id = event.data["repoId"].as_str().map(str::to_owned).or_else(\|\| event.repo_id.clone())?;
261	let number = event.data["number"].as_u64().and_then(\|n\| u32::try_from(n).ok())?;
262	Some(numbered_thread(&repo_id, number))
263	}
264
265	/// Collects who is told, each once with the most specific reason, never
266	/// the actor, never g1t, and never anyone ignoring the thread.
267	struct Told<'a> {
268	actor: &'a Actor,
269	audience: &'a Audience,
270	notices: Vec<Notice>,
271	}
272
273	impl Told<'_> {
274	/// `asked`: something asked of the person directly, told even when
275	/// they unsubscribed from the thread.
276	fn tell(&mut self, username: &str, reason: Reason, severity: Severity, title: &str, body: &str, asked: bool) {
277	let username = username.trim().trim_start_matches('@').to_lowercase();
278	if username.is_empty()
279	\|\| is_g1t(&username)
280	\|\| self.actor.is_named(&username)
281	\|\| self.audience.ignores(&username)
282	\|\| (!asked && self.audience.unsubscribed(&username))
283	{
284	return;
285	}
286	let notice = Notice {
287	username,
288	reason,
289	severity,
290	title: clip(title, MAX_TITLE),
291	body: clip(body, MAX_BODY),
292	};
293	match self.notices.iter_mut().find(\|told\| told.username == notice.username) {
294	Some(told) if notice.reason.rank() < told.reason.rank() => *told = notice,
295	Some(_) => {}
296	None => self.notices.push(notice),
297	}
298	}
299
300	/// A person known by id as well as name, such as an author.
301	fn tell_person(&mut self, person: &Principal, reason: Reason, severity: Severity, title: &str, body: &str, asked: bool) {
302	if is_g1t_id(&person.id) \|\| self.actor.is(person) {
303	return;
304	}
305	self.tell(&person.username, reason, severity, title, body, asked);
306	}
307
308	/// Everyone subscribed to an issue or pull request: its owner and
309	/// author, its assignees and reviewers, and whoever subscribed by
310	/// commenting, being mentioned or by hand. `reason` overrides why
311	/// each is told, as a state change does.
312	fn tell_subscribed(&mut self, subject: &InboxSubject, reason: Option<Reason>, severity: Severity, title: &str, body: &str) {
313	let why = \|own: Reason\| reason.unwrap_or(own);
314	self.tell_person(subject.owner(), why(Reason::Author), severity, title, body, false);
315	self.tell_person(&subject.author, why(Reason::Author), severity, title, body, false);
316	for name in &subject.assignees {
317	self.tell(name, why(Reason::Assign), severity, title, body, false);
318	}
319	for name in &subject.reviewers {
320	self.tell(name, why(Reason::ReviewRequested), severity, title, body, false);
321	}
322	let subscribed: Vec<(String, Reason)> = self.audience.subscribed().collect();
323	for (name, own) in subscribed {
324	self.tell(&name, why(own), severity, title, body, false);
325	}
326	}
327
328	/// Everyone watching the repository for this kind of activity.
329	fn tell_watchers(&mut self, kind: &str, severity: Severity, title: &str, body: &str) {
330	let watching: Vec<String> = self.audience.watching(kind).collect();
331	for name in watching {
332	self.tell(&name, Reason::Subscribed, severity, title, body, false);
333	}
334	}
335	}
336
337	fn clip(text: &str, max: usize) -> String {
338	let text = text.trim();
339	if text.chars().count() <= max {
340	return text.to_owned();
341	}
342	let cut: String = text.chars().take(max - 1).collect();
343	format!("{}…", cut.trim_end())
344	}
345
346	/// The kind of activity a watcher chooses: `issues` or `pulls`.
347	fn activity_of(subject: &InboxSubject) -> &'static str {
348	match subject.kind {
349	Some(SubjectKind::Pull) => "pulls",
350	_ => "issues",
351	}
352	}
353
354	/// The names in an event's list field, such as the reviewers just asked.
355	/// The teams an event asked to review: each `workspace/slug` with the
356	/// people it tells.
357	fn teams_asked(data: &serde_json::Value) -> Vec<(String, Vec<String>)> {
358	data["teams"]
359	.as_array()
360	.map(\|teams\| {
361	teams
362	.iter()
363	.filter_map(\|team\| Some((team["team"].as_str()?.to_owned(), names(team, "notified"))))
364	.collect()
365	})
366	.unwrap_or_default()
367	}
368
369	fn names(data: &serde_json::Value, key: &str) -> Vec<String> {
370	data[key]
371	.as_array()
372	.map(\|names\| names.iter().filter_map(\|name\| name.as_str().map(str::to_owned)).collect())
373	.unwrap_or_default()
374	}
375
376	/// Who is told of `event`, in `repo` (`owner/name`), given what it names
377	/// and who follows it. `actor` is who caused it; outcomes nobody chose
378	/// (checks, workflows, deployments, a review, an agent finishing or
379	/// stopping) are told whoever caused them.
380	pub fn notices(event: &Event, repo: &str, actor: &Actor, subject: Option<&InboxSubject>, audience: &Audience) -> Vec<Notice> {
381	let nobody = Actor::default();
382	let data = &event.data;
383	// The issue or pull request, as titles name it: `acme/rocket#12`.
384	let at = match data["number"].as_u64().or(data["pull"].as_u64()) {
385	Some(number) if subject.is_some() => format!("{repo}#{number}"),
386	_ => repo.to_owned(),
387	};
388	let outcome = matches!(
389	event.kind.as_str(),
390	"checks.completed"
391	\| "workflow.completed"
392	\| "review.completed"
393	\| "pull.ready"
394	\| "agent.asked"
395	\| "pull.stalled"
396	\| "deployment.failed"
397	\| "deployment.succeeded"
398	\| "deployment.review_requested"
399	) \|\| SECURITY_EVENTS.contains(&event.kind.as_str());
400	let mut told = Told {
401	actor: if outcome { &nobody } else { actor },
402	audience,
403	notices: Vec::new(),
404	};
405	// Who did it, as titles name them.
406	let who = actor.name().unwrap_or("g1t");
407
408	match (event.kind.as_str(), subject) {
409	("agent.asked" \| "pull.stalled", Some(pull)) => {
410	let (title, body) = if event.kind == "agent.asked" {
411	(format!("An agent is waiting on {at}"), pull.title.clone())
412	} else {
413	let detail = data["detail"].as_str().map(str::trim).filter(\|detail\| !detail.is_empty());
414	(format!("g1t stopped on {at} and needs you"), detail.unwrap_or(&pull.title).to_owned())
415	};
416	told.tell_person(pull.owner(), Reason::Agent, Severity::Warning, &title, &body, true);
417	if let Some(issue) = &pull.issue {
418	told.tell_person(issue.owner(), Reason::Agent, Severity::Warning, &title, &body, true);
419	for name in &issue.assignees {
420	told.tell(name, Reason::Agent, Severity::Warning, &title, &body, true);
421	}
422	}
423	}
424	("pull.review_requested", Some(pull)) => {
425	// Asked because the CODEOWNERS file says they own what changed.
426	let owned = data["codeOwners"].as_bool() == Some(true);
427	let title = if owned {
428	format!("{at} changes files you own")
429	} else {
430	format!("{who} asked you to review {at}")
431	};
432	for name in names(data, "reviewers") {
433	told.tell(&name, Reason::ReviewRequested, Severity::Warning, &title, &pull.title, true);
434	}
435	for (team, people) in teams_asked(data) {
436	let title = if owned {
437	format!("{at} changes files @{team} owns")
438	} else {
439	format!("{who} asked @{team} to review {at}")
440	};
441	for name in people {
442	told.tell(&name, Reason::ReviewRequested, Severity::Warning, &title, &pull.title, true);
443	}
444	}
445	}
446	("issue.assigned" \| "pull.assigned", Some(on)) => {
447	let title = format!("{who} assigned you to {at}");
448	for name in names(data, "added") {
449	told.tell(&name, Reason::Assign, Severity::Info, &title, &on.title, true);
450	}
451	}
452	("issue.opened" \| "pull.opened", Some(on)) => {
453	let title = format!("{who} opened {at}");
454	for name in &on.assignees {
455	told.tell(name, Reason::Assign, Severity::Info, &format!("{who} assigned you to {at}"), &on.title, true);
456	}
457	for name in &on.reviewers {
458	told.tell(name, Reason::ReviewRequested, Severity::Warning, &format!("{who} asked you to review {at}"), &on.title, true);
459	}
460	for name in &on.mentions {
461	told.tell(name, Reason::Mention, Severity::Info, &format!("{who} mentioned you on {at}"), &on.title, true);
462	}
463	for team in &on.team_mentions {
464	let title = format!("{who} mentioned @{} on {at}", team.team);
465	for name in &team.members {
466	told.tell(name, Reason::TeamMention, Severity::Info, &title, &on.title, true);
467	}
468	}
469	told.tell_watchers(activity_of(on), Severity::Info, &title, &on.title);
470	}
471	("checks.completed", Some(pull)) => {
472	let title = match data["status"].as_str() {
473	Some("errored") => format!("Checks could not run on {at}"),
474	_ => format!("Checks failed on {at}"),
475	};
476	told.tell_person(pull.owner(), Reason::CiActivity, Severity::Error, &title, &pull.title, true);
477	}
478	("workflow.completed", subject) => {
479	let workflow = data["workflow"].as_str().filter(\|name\| !name.is_empty()).unwrap_or("A workflow");
480	match subject {
481	Some(pull) => {
482	let title = format!("{workflow} failed on {at}");
483	told.tell_person(pull.owner(), Reason::CiActivity, Severity::Error, &title, &pull.title, true);
484	}
485	None => {
486	// Not on a pull request: whoever pushed the commit it ran on.
487	let branch = data["ref"].as_str().unwrap_or_default();
488	let branch = branch.strip_prefix("refs/heads/").unwrap_or(branch);
489	let title = format!("{workflow} failed on {branch} in {repo}");
490	let body = format!("Run {} at {}", data["number"], short(data["sha"].as_str().unwrap_or_default()));
491	if let Some(id) = &actor.id
492	&& !is_g1t_id(id)
493	&& let Some(name) = &actor.username
494	{
495	told.tell(name, Reason::CiActivity, Severity::Error, &title, &body, true);
496	}
497	}
498	}
499	}
500	("deployment.failed" \| "deployment.succeeded", subject) => {
501	let failed = event.kind == "deployment.failed";
502	let project = data["project"].as_str().filter(\|name\| !name.is_empty()).unwrap_or(repo);
503	let what = match subject {
504	Some(_) => format!("The preview of {at}"),
505	None => format!("Production of {project}"),
506	};
507	let (title, severity) = if failed {
508	(format!("{what} failed to deploy"), Severity::Error)
509	} else {
510	(format!("{what} is live"), Severity::Success)
511	};
512	let body = match (failed, data["error"].as_str().map(str::trim).filter(\|error\| !error.is_empty())) {
513	(true, Some(error)) => error.to_owned(),
514	_ => match subject {
515	Some(pull) => pull.title.clone(),
516	None => format!("Commit {}", short(data["commit"].as_str().unwrap_or_default())),
517	},
518	};
519	// A success is news to whoever answers for it only after a failure.
520	if failed \|\| data["recovered"].as_bool() == Some(true) {
521	let title = if failed { title.clone() } else { format!("{what} is live again") };
522	match subject {
523	Some(pull) => told.tell_person(pull.owner(), Reason::CiActivity, severity, &title, &body, true),
524	None => {
525	let pusher = actor
526	.id
527	.as_deref()
528	.filter(\|id\| !is_g1t_id(id))
529	.and(actor.username.as_deref())
530	.or_else(\|\| data["triggeredBy"].as_str());
531	if let Some(name) = pusher {
532	told.tell(name, Reason::CiActivity, severity, &title, &body, true);
533	}
534	}
535	}
536	}
537	told.tell_watchers("deployments", severity, &title, &body);
538	}
539	("review.completed", Some(pull)) => {
540	let (title, severity) = match data["verdict"].as_str() {
541	Some("approve") => (format!("g1t approved {at}"), Severity::Success),
542	_ => (format!("g1t asked for changes on {at}"), Severity::Info),
543	};
544	told.tell_person(pull.owner(), Reason::Author, severity, &title, &pull.title, true);
545	}
546	("pull.ready", Some(pull)) => {
547	// A change g1t made is ready: the agent's run is over.
548	if let Some(owner) = pull.requested_by.as_ref().filter(\|_\| is_g1t_id(&pull.author.id) \|\| is_g1t(&pull.author.username)) {
549	let title = format!("g1t finished {at}");
550	told.tell_person(owner, Reason::Author, Severity::Success, &title, &pull.title, true);
551	}
552	}
553	("pull.merged" \| "pull.closed" \| "issue.closed" \| "issue.reopened", Some(on)) => {
554	let (verb, severity) = match event.kind.as_str() {
555	"pull.merged" => ("merged", Severity::Success),
556	"issue.reopened" => ("reopened", Severity::Info),
557	_ => ("closed", Severity::Info),
558	};
559	let title = match (actor.name(), data["resolvedBy"].as_u64()) {
560	(_, Some(pull)) if event.kind == "issue.closed" => format!("{at} was closed by #{pull}"),
561	(Some(name), _) => format!("{name} {verb} {at}"),
562	(None, _) => format!("{at} was {verb}"),
563	};
564	told.tell_subscribed(on, Some(Reason::StateChange), severity, &title, &on.title);
565	told.tell_watchers(activity_of(on), severity, &title, &on.title);
566	}
567	("deployment.review_requested", _) => {
568	let environment = data["environment"].as_str().unwrap_or("an environment");
569	let workflow = data["workflow"].as_str().unwrap_or("A workflow");
570	let title = format!("{workflow} is waiting for your review to deploy to {environment} in {repo}");
571	let body = data["title"].as_str().unwrap_or_default();
572	// Those the actions service names: the environment's reviewers.
573	for name in names(data, "notify") {
574	told.tell(&name, Reason::ReviewRequested, Severity::Warning, &title, body, true);
575	}
576	}
577	(kind, None) if SECURITY_EVENTS.contains(&kind) => {
578	let severity = match data["severity"].as_str() {
579	_ if kind == "secret_scanning.bypass_requested" => Severity::Warning,
580	_ if kind == "secret_scanning.bypass_reviewed" => Severity::Info,
581	Some("critical" \| "high") => Severity::Error,
582	_ => Severity::Warning,
583	};
584	let what = data["title"].as_str().unwrap_or("A security alert");
585	let title = match kind {
586	"secret_scanning.bypass_requested" => format!("{who} asked to bypass push protection in {repo}"),
587	"secret_scanning.bypass_reviewed" => {
588	let verdict = data["state"].as_str().unwrap_or("reviewed");
589	format!("Your request to bypass push protection in {repo} was {verdict}")
590	}
591	_ if data["pusher"].is_string() => format!("A push to {repo} was blocked: it adds a secret"),
592	_ => format!("New security alert in {repo}"),
593	};
594	// Whoever pushed a blocked secret, and those the service names
595	// (the workspace's owners, or whoever asked for a bypass).
596	if let Some(pusher) = data["pusher"].as_str() {
597	told.tell(pusher, Reason::SecurityAlert, Severity::Error, &title, what, true);
598	}
599	for name in names(data, "notify") {
600	told.tell(&name, Reason::SecurityAlert, severity, &title, what, true);
601	}
602	// Watchers who chose security alerts, if they may see findings:
603	// the service lists who may (`members`), as findings are never
604	// shown to someone who cannot change the code.
605	if !kind.starts_with("secret_scanning.bypass") {
606	let members = names(data, "members");
607	let watching: Vec<String> = told.audience.watching("security").collect();
608	for name in watching.iter().filter(\|name\| members.iter().any(\|member\| member.eq_ignore_ascii_case(name))) {
609	told.tell(name, Reason::SecurityAlert, severity, &title, what, false);
610	}
611	}
612	}
613	("comment.created", Some(on)) => {
614	let Some(comment) = on.comment.as_ref().filter(\|comment\| !comment.event) else {
615	return Vec::new();
616	};
617	// Whoever wrote it is the actor, whatever the event says.
618	let writer = Actor {
619	id: Some(comment.author.id.clone()),
620	username: Some(comment.author.username.clone()),
621	};
622	told.actor = &writer;
623	let who = &comment.author.username;
624	let body = if comment.excerpt.is_empty() { &on.title } else { &comment.excerpt };
625	for name in &comment.mentions {
626	let title = format!("{who} mentioned you on {at}");
627	told.tell(name, Reason::Mention, Severity::Info, &title, body, true);
628	}
629	for team in &comment.team_mentions {
630	let title = format!("{who} mentioned @{} on {at}", team.team);
631	for name in &team.members {
632	told.tell(name, Reason::TeamMention, Severity::Info, &title, body, true);
633	}
634	}
635	let (severity, title) = match comment.verdict.as_deref() {
636	Some("approve") => (Severity::Success, format!("{who} approved {at}")),
637	Some("request_changes") => (Severity::Info, format!("{who} asked for changes on {at}")),
638	_ => (Severity::Info, format!("{who} commented on {at}")),
639	};
640	// An approval or a request for changes is the owner's to hear
641	// of whatever they chose; the rest, as they subscribed.
642	if comment.verdict.is_some() {
643	told.tell_person(on.owner(), Reason::Author, severity, &title, body, true);
644	}
645	told.tell_subscribed(on, None, severity, &title, body);
646	told.tell_watchers(activity_of(on), severity, &title, body);
647	return told.notices;
648	}
649	_ => {}
650	}
651	told.notices
652	}
653
654	/// Who an event subscribes to its issue or pull request without asking,
655	/// and why: whoever commented, whoever they mentioned, the people assigned
656	/// and the reviewers asked. Never g1t.
657	pub fn subscribes(event: &Event, subject: Option<&InboxSubject>) -> Vec<(String, Reason)> {
658	let mut people: Vec<(String, Reason)> = Vec::new();
659	let mut add = \|name: &str, reason: Reason\| {
660	let name = name.trim().trim_start_matches('@').to_lowercase();
661	if !name.is_empty() && !is_g1t(&name) && !people.iter().any(\|(had, _)\| *had == name) {
662	people.push((name, reason));
663	}
664	};
665	match event.kind.as_str() {
666	"comment.created" => {
667	if let Some(comment) = subject.and_then(\|subject\| subject.comment.as_ref()).filter(\|comment\| !comment.event) {
668	if !is_g1t_id(&comment.author.id) {
669	add(&comment.author.username, Reason::Comment);
670	}
671	for name in &comment.mentions {
672	add(name, Reason::Mention);
673	}
674	for team in &comment.team_mentions {
675	for name in &team.members {
676	add(name, Reason::TeamMention);
677	}
678	}
679	}
680	}
681	"issue.opened" \| "pull.opened" => {
682	if let Some(on) = subject {
683	for name in &on.mentions {
684	add(name, Reason::Mention);
685	}
686	for team in &on.team_mentions {
687	for name in &team.members {
688	add(name, Reason::TeamMention);
689	}
690	}
691	}
692	}
693	"issue.assigned" \| "pull.assigned" => {
694	for name in names(&event.data, "added") {
695	add(&name, Reason::Assign);
696	}
697	}
698	"pull.review_requested" => {
699	for name in names(&event.data, "reviewers") {
700	add(&name, Reason::ReviewRequested);
701	}
702	for (_, people) in teams_asked(&event.data) {
703	for name in people {
704	add(&name, Reason::ReviewRequested);
705	}
706	}
707	}
708	_ => {}
709	}
710	people
711	}
712
713	fn short(sha: &str) -> &str {
714	sha.get(..7).unwrap_or(sha)
715	}
716
717	/// Where an item is on g1t.sh.
718	pub fn url(repo: Option<&str>, subject: Option<SubjectKind>, number: Option<u32>, run_id: Option<&str>, link: Option<&str>) -> String {
719	if let Some(link) = link.filter(\|link\| link.starts_with('/')) {
720	return link.to_owned();
721	}
722	let Some(repo) = repo else {
723	return "/inbox".to_owned();
724	};
725	match (subject, number, run_id) {
726	(Some(SubjectKind::Pull), Some(number), _) => format!("/{repo}/pull/{number}"),
727	(Some(SubjectKind::Issue), Some(number), _) => format!("/{repo}/issues/{number}"),
728	(Some(SubjectKind::Run), _, Some(run)) => format!("/{repo}/actions/runs/{run}"),
729	(Some(SubjectKind::Deploy), Some(number), _) => format!("/{repo}/pull/{number}"),
730	_ => format!("/{repo}"),
731	}
732	}
733
734	// --- Writing ---------------------------------------------------------------
735
736	/// The services the inbox reads from as events arrive.
737	pub struct Sources<'a> {
738	pub work: &'a Fetcher,
739	pub repos: &'a Fetcher,
740	pub identity: &'a Fetcher,
741	}
742
743	/// One notice written, and whether it was news (not a redelivery): what
744	/// may be emailed.
745	struct Written {
746	notice: Notice,
747	repo_id: String,
748	url: String,
749	}
750
751	/// Writes the items a batch from the bus calls for. Never fails the batch:
752	/// what cannot be worked out is logged and left.
753	pub async fn deliver(db: &D1Database, sources: &Sources<'_>, events: &[Event]) {
754	if let Err(error) = resolve(db, events).await {
755	worker::console_error!("inbox: agent threads not closed: {error}");
756	}
757	let wanted: Vec<(&Event, Wanted)> = events
758	.iter()
759	.filter_map(\|event\| wants(event).map(\|wanted\| (event, wanted)))
760	.collect();
761	let created: Vec<&Event> = events.iter().filter(\|event\| event.kind == "repo.created").collect();
762	if wanted.is_empty() && created.is_empty() {
763	return;
764	}
765	// Everyone who caused one, named in one call.
766	let ids: Vec<String> = wanted
767	.iter()
768	.map(\|(event, _)\| *event)
769	.chain(created.iter().copied())
770	.filter_map(\|event\| event.actor.clone())
771	.filter(\|id\| !is_g1t_id(id))
772	.collect::<HashSet<_>>()
773	.into_iter()
774	.collect();
775	let names: HashMap<String, String> = if ids.is_empty() {
776	HashMap::new()
777	} else {
778	g1t_kit::call(sources.identity, "usernames", &UsernamesArgs { ids })
779	.await
780	.unwrap_or_else(\|error\| {
781	worker::console_error!("inbox: could not name who acted: {error}");
782	HashMap::new()
783	})
784	};
785	for event in created {
786	if let Err(error) = subscriptions::watch_created(db, event, &names).await {
787	worker::console_error!("inbox: {} {} not watched: {error}", event.kind, event.id);
788	}
789	}
790	let mut paths: HashMap<String, Option<RepoPath>> = HashMap::new();
791	let mut watchers: HashMap<String, Vec<subscriptions::Watcher>> = HashMap::new();
792	let mut written: Vec<Written> = Vec::new();
793	for (event, wanted) in wanted {
794	match deliver_one(db, sources, &names, &mut paths, &mut watchers, event, wanted).await {
795	Ok(mut news) => written.append(&mut news),
796	Err(error) => worker::console_error!("inbox: {} {} not delivered: {error}", event.kind, event.id),
797	}
798	}
799	if let Err(error) = email(db, sources.identity, written).await {
800	worker::console_error!("inbox: emails not sent: {error}");
801	}
802	}
803
804	/// Closes what an agent was waiting on a person for once it is over.
805	async fn resolve(db: &D1Database, events: &[Event]) -> Result<()> {
806	let now = rfc3339(now_ms());
807	let mut statements = Vec::new();
808	for thread in events.iter().filter_map(resolves) {
809	statements.push(
810	db.prepare(
811	"UPDATE inbox_items SET done_at = ?1, read_at = COALESCE(read_at, ?1)
812	WHERE thread = ?2 AND reason = 'agent' AND done_at IS NULL",
813	)
814	.bind(&[now.as_str().into(), thread.into()])?,
815	);
816	}
817	// Reviews no longer asked for are no longer waiting.
818	for event in events.iter().filter(\|event\| event.kind == "pull.review_request_removed") {
819	let (Some(repo_id), Some(number)) = (
820	event.data["repoId"].as_str().map(str::to_owned).or_else(\|\| event.repo_id.clone()),
821	event.data["number"].as_u64(),
822	) else {
823	continue;
824	};
825	for name in names(&event.data, "reviewers") {
826	statements.push(
827	db.prepare(
828	"UPDATE inbox_items SET done_at = ?1, read_at = COALESCE(read_at, ?1)
829	WHERE thread = ?2 AND username = ?3 AND reason = 'review_requested' AND done_at IS NULL",
830	)
831	.bind(&[
832	now.as_str().into(),
833	format!("{repo_id}#{number}").into(),
834	name.to_lowercase().into(),
835	])?,
836	);
837	}
838	}
839	if !statements.is_empty() {
840	db.batch(statements).await?;
841	}
842	Ok(())
843	}
844
845	async fn deliver_one(
846	db: &D1Database,
847	sources: &Sources<'_>,
848	names: &HashMap<String, String>,
849	paths: &mut HashMap<String, Option<RepoPath>>,
850	watchers: &mut HashMap<String, Vec<subscriptions::Watcher>>,
851	event: &Event,
852	wanted: Wanted,
853	) -> Result<Vec<Written>> {
854	if !paths.contains_key(&wanted.repo_id) {
855	let path: Option<RepoPath> = g1t_kit::call(
856	sources.repos,
857	"path_by_id",
858	&PathByIdArgs {
859	id: wanted.repo_id.clone(),
860	},
861	)
862	.await?;
863	paths.insert(wanted.repo_id.clone(), path);
864	}
865	// A repository that is gone tells nobody.
866	let Some(path) = paths.get(&wanted.repo_id).cloned().flatten() else {
867	return Ok(Vec::new());
868	};
869	let subject: Option<InboxSubject> = match wanted.number {
870	Some(number) => {
871	let found: Option<InboxSubject> = g1t_kit::call(
872	sources.work,
873	"inbox_subject",
874	&InboxSubjectArgs {
875	repo_id: wanted.repo_id.clone(),
876	number,
877	comment_id: wanted.comment_id.clone(),
878	},
879	)
880	.await?;
881	if found.is_none() {
882	return Ok(Vec::new());
883	}
884	found
885	}
886	None => None,
887	};
888	let actor = Actor {
889	id: event.actor.clone(),
890	username: match event.actor.as_deref() {
891	Some(id) if is_g1t_id(id) => Some(system::USERNAME.to_owned()),
892	Some(id) => names.get(id).map(\|name\| name.to_lowercase()),
893	None => None,
894	},
895	};
896	let thread = thread_of(event, &wanted, subject.as_ref());
897	if !watchers.contains_key(&wanted.repo_id) {
898	watchers.insert(wanted.repo_id.clone(), subscriptions::watchers(db, &wanted.repo_id).await?);
899	}
900	let audience = Audience {
901	subscriptions: match thread.kind {
902	Some(SubjectKind::Issue \| SubjectKind::Pull) => subscriptions::of_thread(db, &thread.key).await?,
903	_ => Vec::new(),
904	},
905	watchers: watchers.get(&wanted.repo_id).cloned().unwrap_or_default(),
906	};
907	let repo = format!("{}/{}", path.namespace, path.name).to_lowercase();
908	let told = notices(event, &repo, &actor, subject.as_ref(), &audience);
909	let mut statements = subscriptions::auto_subscribe(db, &thread.key, &wanted.repo_id, &subscribes(event, subject.as_ref()), &event.time)?;
910	if told.is_empty() {
911	if !statements.is_empty() {
912	db.batch(statements).await?;
913	}
914	return Ok(Vec::new());
915	}
916	let now = now_ms();
917	let link = thread.link.as_deref();
918	let item_url = url(Some(&repo), thread.kind, thread.number, thread.run_id.as_deref(), link);
919	// Three statements a notice: its thread brought up (unless this event
920	// was told before), a line of history, and the history kept short.
921	let first = statements.len();
922	for notice in &told {
923	let username = notice.username.as_str();
924	let values: Vec<JsValue> = vec![
925	new_id("ntf", now).into(),
926	username.into(),
927	thread.key.as_str().into(),
928	event.id.as_str().into(),
929	event.kind.as_str().into(),
930	notice.reason.as_str().into(),
931	notice.severity.as_str().into(),
932	notice.title.as_str().into(),
933	notice.body.as_str().into(),
934	path.namespace.to_lowercase().into(),
935	wanted.repo_id.as_str().into(),
936	repo.as_str().into(),
937	thread.kind.map_or(JsValue::NULL, \|kind\| kind.as_str().into()),
938	thread.number.map_or(JsValue::NULL, JsValue::from),
939	thread.run_id.as_deref().map_or(JsValue::NULL, JsValue::from),
940	link.map_or(JsValue::NULL, JsValue::from),
941	actor.username.as_deref().map_or(JsValue::NULL, JsValue::from),
942	event.time.as_str().into(),
943	// Same prefix as item ids, so old and new sort by time together.
944	new_id("ntf", now).into(),
945	];
946	statements.push(db.prepare(BUMP).bind(&values)?);
947	statements.push(
948	db.prepare(
949	"INSERT OR IGNORE INTO inbox_activity (id, item_id, username, event_id, event, reason, severity, title, body, actor, created_at)
950	SELECT ?1, id, ?2, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11 FROM inbox_items WHERE username = ?2 AND thread = ?3
951	RETURNING username",
952	)
953	.bind(&[
954	new_id("ntf", now).into(),
955	username.into(),
956	thread.key.as_str().into(),
957	event.id.as_str().into(),
958	event.kind.as_str().into(),
959	notice.reason.as_str().into(),
960	notice.severity.as_str().into(),
961	notice.title.as_str().into(),
962	notice.body.as_str().into(),
963	actor.username.as_deref().map_or(JsValue::NULL, JsValue::from),
964	event.time.as_str().into(),
965	])?,
966	);
967	statements.push(
968	db.prepare(
969	"DELETE FROM inbox_activity
970	WHERE item_id = (SELECT id FROM inbox_items WHERE username = ?1 AND thread = ?2)
971	AND id NOT IN (
972	SELECT a.id FROM inbox_activity a
973	WHERE a.item_id = (SELECT id FROM inbox_items WHERE username = ?1 AND thread = ?2)
974	ORDER BY a.id DESC LIMIT ?3)",
975	)
976	.bind(&[username.into(), thread.key.as_str().into(), MAX_ACTIVITY.into()])?,
977	);
978	}
979	let results = db.batch(statements).await?;
980	#[derive(Deserialize)]
981	struct Told {
982	username: String,
983	}
984	// A line of history written means the event is news to that person.
985	let mut news = HashSet::new();
986	for (at, _) in told.iter().enumerate() {
987	if let Some(result) = results.get(first + at * 3 + 1)
988	&& let Ok(rows) = result.results::<Told>()
989	{
990	news.extend(rows.into_iter().map(\|row\| row.username));
991	}
992	}
993	Ok(told
994	.into_iter()
995	.filter(\|notice\| news.contains(&notice.username))
996	.map(\|notice\| Written {
997	notice,
998	repo_id: wanted.repo_id.clone(),
999	url: item_url.clone(),
1000	})
1001	.collect())
1002	}
1003
1004	/// Brings a person's thread up with new activity, or starts it. `?1` id,
1005	/// `?2` username, `?3` thread, `?4` event id, `?5` event type, `?6` reason,
1006	/// `?7` severity, `?8` title, `?9` body, `?10` workspace, `?11` repo id,
1007	/// `?12` repo, `?13` subject, `?14` number, `?15` run id, `?16` link, `?17`
1008	/// actor, `?18` time, `?19` the time-sortable id lists order by. Nothing
1009	/// happens when the person was already told of this event. While the
1010	/// thread is unread its most urgent severity is kept; done or not, it
1011	/// comes back to the inbox. A snooze stands.
1012	const BUMP: &str = "INSERT INTO inbox_items (id, username, thread, event_id, event, reason, severity, title, body,
1013	workspace, repo_id, repo, subject, number, run_id, link, actor, created_at, updated_at, bumped, activity)
1014	SELECT ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?18, ?19, 1
1015	WHERE NOT EXISTS (SELECT 1 FROM inbox_activity WHERE event_id = ?4 AND username = ?2)
1016	ON CONFLICT (username, thread) DO UPDATE SET
1017	event_id = excluded.event_id,
1018	event = excluded.event,
1019	reason = excluded.reason,
1020	severity = CASE
1021	WHEN inbox_items.read_at IS NULL AND inbox_items.done_at IS NULL
1022	AND (CASE inbox_items.severity WHEN 'warning' THEN 0 WHEN 'error' THEN 1 WHEN 'success' THEN 2 ELSE 3 END)
1023	< (CASE excluded.severity WHEN 'warning' THEN 0 WHEN 'error' THEN 1 WHEN 'success' THEN 2 ELSE 3 END)
1024	THEN inbox_items.severity ELSE excluded.severity END,
1025	title = excluded.title,
1026	body = excluded.body,
1027	workspace = excluded.workspace,
1028	repo = excluded.repo,
1029	subject = COALESCE(excluded.subject, inbox_items.subject),
1030	run_id = COALESCE(excluded.run_id, inbox_items.run_id),
1031	link = COALESCE(excluded.link, inbox_items.link),
1032	actor = excluded.actor,
1033	updated_at = excluded.updated_at,
1034	bumped = excluded.bumped,
1035	activity = inbox_items.activity + 1,
1036	read_at = NULL,
1037	done_at = NULL";
1038
1039	/// Emails what was news to people who asked to be emailed for its reason.
1040	/// Identity sends each, only to a confirmed address and only if the person
1041	/// can still read the repository.
1042	async fn email(db: &D1Database, identity: &Fetcher, written: Vec<Written>) -> Result<()> {
1043	if written.is_empty() {
1044	return Ok(());
1045	}
1046	let people: Vec<String> = written
1047	.iter()
1048	.map(\|item\| item.notice.username.clone())
1049	.collect::<HashSet<_>>()
1050	.into_iter()
1051	.collect();
1052	let settings = subscriptions::settings_of(db, &people).await?;
1053	for item in written {
1054	let wants = settings.get(&item.notice.username).map_or(&DEFAULT_EMAIL[..], \|settings\| &settings.email[..]);
1055	if !wants.contains(&item.notice.reason) {
1056	continue;
1057	}
1058	let sent: Result<bool> = g1t_kit::call(
1059	identity,
1060	"notify_by_email",
1061	&NotifyByEmailArgs {
1062	username: item.notice.username.clone(),
1063	repo_id: item.repo_id.clone(),
1064	subject: item.notice.title.clone(),
1065	intro: item.notice.body.clone(),
1066	quote: None,
1067	path: item.url.clone(),
1068	reason: item.notice.reason,
1069	},
1070	)
1071	.await;
1072	if let Err(error) = sent {
1073	worker::console_error!("inbox: email to {} not sent: {error}", item.notice.username);
1074	}
1075	}
1076	Ok(())
1077	}
1078
1079	// --- Reading and changing ----------------------------------------------------
1080
1081	#[derive(Deserialize)]
1082	pub(crate) struct Row {
1083	pub(crate) id: String,
1084	reason: String,
1085	severity: String,
1086	title: String,
1087	body: String,
1088	event: Option<String>,
1089	workspace: Option<String>,
1090	pub(crate) repo_id: Option<String>,
1091	repo: Option<String>,
1092	subject: Option<String>,
1093	number: Option<f64>,
1094	run_id: Option<String>,
1095	link: Option<String>,
1096	actor: Option<String>,
1097	activity: Option<f64>,
1098	created_at: String,
1099	updated_at: Option<String>,
1100	read_at: Option<String>,
1101	done_at: Option<String>,
1102	saved: f64,
1103	snoozed_until: Option<String>,
1104	bumped: Option<String>,
1105	}
1106
1107	impl Row {
1108	pub(crate) fn into_item(self) -> InboxItem {
1109	let subject = self.subject.as_deref().and_then(SubjectKind::parse);
1110	let number = self.number.map(\|n\| n as u32);
1111	InboxItem {
1112	url: url(self.repo.as_deref(), subject, number, self.run_id.as_deref(), self.link.as_deref()),
1113	id: self.id,
1114	reason: Reason::parse(&self.reason).unwrap_or(Reason::Subscribed),
1115	severity: Severity::parse(&self.severity).unwrap_or(Severity::Info),
1116	title: self.title,
1117	body: self.body,
1118	event: self.event,
1119	repo: self.repo,
1120	workspace: self.workspace,
1121	subject,
1122	number,
1123	actor: self.actor,
1124	count: self.activity.map_or(1, \|n\| n as u32),
1125	updated_at: self.updated_at.unwrap_or_else(\|\| self.created_at.clone()),
1126	created_at: self.created_at,
1127	read_at: self.read_at,
1128	done_at: self.done_at,
1129	saved: self.saved != 0.0,
1130	snoozed_until: self.snoozed_until,
1131	}
1132	}
1133	}
1134
1135	pub(crate) const COLUMNS: &str = "id, reason, severity, title, body, event, workspace, repo_id, repo, subject, number, run_id,
1136	link, actor, activity, created_at, updated_at, read_at, done_at, saved, snoozed_until, bumped";
1137
1138	/// The conditions that pick a view's items, after `username = ?1`; `?2` is now.
1139	fn view_filter(view: InboxView) -> &'static str {
1140	match view {
1141	InboxView::Inbox => "done_at IS NULL AND (snoozed_until IS NULL OR snoozed_until <= ?2)",
1142	InboxView::Saved => "saved = 1",
1143	InboxView::Done => "done_at IS NOT NULL",
1144	}
1145	}
1146
1147	/// Whether a list ranks unread warnings first: the inbox itself, unfiltered.
1148	fn ranked(a: &ListInboxArgs) -> bool {
1149	a.view == InboxView::Inbox
1150	&& a.severity.is_none()
1151	&& a.reason.is_none()
1152	&& !a.participating
1153	&& a.repo_id.is_none()
1154	&& !a.unread
1155	&& a.since.is_none()
1156	&& a.updated_before.is_none()
1157	}
1158
1159	/// The conditions a list's filters add, and the values they bind, numbered
1160	/// after the `bound` values already given.
1161	fn filters(a: &ListInboxArgs, bound: usize) -> (Vec<String>, Vec<String>) {
1162	let mut conditions = Vec::new();
1163	let mut values: Vec<String> = Vec::new();
1164	let mut bind = \|value: &str, condition: &str\| {
1165	values.push(value.to_owned());
1166	conditions.push(condition.replace('?', &format!("?{}", bound + values.len())));
1167	};
1168	if let Some(severity) = a.severity {
1169	bind(severity.as_str(), "severity = ?");
1170	}
1171	if let Some(reason) = a.reason {
1172	bind(reason.as_str(), "reason = ?");
1173	}
1174	if let Some(repo_id) = &a.repo_id {
1175	bind(repo_id, "repo_id = ?");
1176	}
1177	if let Some(since) = &a.since {
1178	bind(since.trim(), "COALESCE(updated_at, created_at) >= ?");
1179	}
1180	if let Some(before) = &a.updated_before {
1181	bind(before.trim(), "COALESCE(updated_at, created_at) < ?");
1182	}
1183	if a.participating {
1184	conditions.push("reason NOT IN ('manual', 'subscribed')".to_owned());
1185	}
1186	if a.unread {
1187	conditions.push("read_at IS NULL".to_owned());
1188	}
1189	(conditions, values)
1190	}
1191
1192	pub async fn list(db: &D1Database, repos: &Fetcher, a: ListInboxArgs) -> Result<InboxPage> {
1193	let Some(viewer) = &a.viewer else {
1194	return Ok(InboxPage::default());
1195	};
1196	let username = viewer.username.to_lowercase();
1197	let now = rfc3339(now_ms());
1198	let limit = a.limit.unwrap_or(DEFAULT_INBOX_PAGE).clamp(1, MAX_INBOX_PAGE);
1199	let mut values: Vec<JsValue> = vec![username.as_str().into(), now.as_str().into()];
1200	let mut conditions = vec!["username = ?1".to_owned(), view_filter(a.view).to_owned()];
1201	let (filtered, bound) = filters(&a, values.len());
1202	conditions.extend(filtered);
1203	values.extend(bound.iter().map(\|value\| JsValue::from(value.as_str())));
1204	let ranked = ranked(&a);
1205	// Unread warnings lead the first page, and are left out of the rest.
1206	let leading = "severity = 'warning' AND read_at IS NULL";
1207	let mut rest = conditions.clone();
1208	let mut rest_values = values.clone();
1209	if ranked {
1210	rest.push(format!("NOT ({leading})"));
1211	}
1212	if let Some(before) = &a.before {
1213	rest_values.push(before.as_str().into());
1214	rest.push(format!("bumped < ?{}", rest_values.len()));
1215	}
1216	rest_values.push((limit + 1).into());
1217	let order = if a.view == InboxView::Done { "done_at DESC, bumped DESC" } else { "bumped DESC" };
1218	let mut statements = vec![
1219	db.prepare(format!(
1220	"SELECT {COLUMNS} FROM inbox_items WHERE {} ORDER BY {order} LIMIT ?{}",
1221	rest.join(" AND "),
1222	rest_values.len()
1223	))
1224	.bind(&rest_values)?,
1225	];
1226	if ranked && a.before.is_none() {
1227	statements.push(
1228	db.prepare(format!(
1229	"SELECT {COLUMNS} FROM inbox_items WHERE {} AND {leading} ORDER BY bumped DESC LIMIT ?3",
1230	conditions.join(" AND ")
1231	))
1232	.bind(&[username.as_str().into(), now.as_str().into(), MAX_RANKED.into()])?,
1233	);
1234	}
1235	let results = db.batch(statements).await?;
1236	let mut rows = results.first().map(\|result\| result.results::<Row>()).transpose()?.unwrap_or_default();
1237	let next = if rows.len() > limit as usize {
1238	rows.truncate(limit as usize);
1239	rows.last().map(\|row\| row.bumped.clone().unwrap_or_else(\|\| row.id.clone()))
1240	} else {
1241	None
1242	};
1243	let mut items: Vec<Row> = results.get(1).map(\|result\| result.results::<Row>()).transpose()?.unwrap_or_default();
1244	items.extend(rows);
1245	let items = readable_only(db, repos, &a.viewer, &username, items).await?;
1246	Ok(InboxPage {
1247	items: items.into_iter().map(Row::into_item).collect(),
1248	next,
1249	})
1250	}
1251
1252	/// Rows about repositories the viewer can still read; the rest are
1253	/// removed from their inbox as they are found.
1254	pub(crate) async fn readable_only(db: &D1Database, repos: &Fetcher, viewer: &g1t_contracts::Viewer, username: &str, mut rows: Vec<Row>) -> Result<Vec<Row>> {
1255	let ids: Vec<String> = rows
1256	.iter()
1257	.filter_map(\|row\| row.repo_id.clone())
1258	.collect::<HashSet<_>>()
1259	.into_iter()
1260	.collect();
1261	if ids.is_empty() {
1262	return Ok(rows);
1263	}
1264	let readable: Vec<Repo> = g1t_kit::call(
1265	repos,
1266	"readable",
1267	&ReadableArgs {
1268	ids: ids.clone(),
1269	viewer: viewer.clone(),
1270	},
1271	)
1272	.await?;
1273	let readable: HashSet<String> = readable.into_iter().map(\|repo\| repo.id).collect();
1274	let gone: Vec<String> = ids.into_iter().filter(\|id\| !readable.contains(id)).collect();
1275	if !gone.is_empty() {
1276	forget(db, username, &gone).await?;
1277	rows.retain(\|row\| row.repo_id.as_ref().is_none_or(\|id\| !gone.contains(id)));
1278	}
1279	Ok(rows)
1280	}
1281
1282	/// Removes a person's items about repositories they cannot read.
1283	async fn forget(db: &D1Database, username: &str, repo_ids: &[String]) -> Result<()> {
1284	let marks = vec!["?"; repo_ids.len()].join(", ");
1285	let mut values: Vec<JsValue> = vec![username.into()];
1286	values.extend(repo_ids.iter().map(\|id\| JsValue::from(id.as_str())));
1287	db.batch(vec![
1288	db.prepare(format!(
1289	"DELETE FROM inbox_activity WHERE item_id IN (SELECT id FROM inbox_items WHERE username = ? AND repo_id IN ({marks}))"
1290	))
1291	.bind(&values)?,
1292	db.prepare(format!("DELETE FROM inbox_items WHERE username = ? AND repo_id IN ({marks})"))
1293	.bind(&values)?,
1294	])
1295	.await?;
1296	Ok(())
1297	}
1298
1299	#[derive(Deserialize)]
1300	struct CountRow {
1301	severity: String,
1302	n: f64,
1303	}
1304
1305	pub async fn counts(db: &D1Database, a: InboxCountsArgs) -> Result<InboxCounts> {
1306	let rows = db
1307	.prepare(
1308	"SELECT severity, count(*) AS n FROM inbox_items
1309	WHERE username = ? AND read_at IS NULL AND done_at IS NULL
1310	AND (snoozed_until IS NULL OR snoozed_until <= ?)
1311	GROUP BY severity",
1312	)
1313	.bind(&[a.username.to_lowercase().into(), rfc3339(now_ms()).into()])?
1314	.all()
1315	.await?
1316	.results::<CountRow>()?;
1317	let mut counts = InboxCounts::default();
1318	for row in rows {
1319	let n = row.n as u32;
1320	counts.unread += n;
1321	match Severity::parse(&row.severity) {
1322	Some(Severity::Error) => counts.error += n,
1323	Some(Severity::Warning) => counts.warning += n,
1324	Some(Severity::Success) => counts.success += n,
1325	Some(Severity::Info) \| None => counts.info += n,
1326	}
1327	}
1328	Ok(counts)
1329	}
1330
1331	/// The change a mark makes, as a `SET` clause; `?1` is now.
1332	fn mark_change(mark: InboxMark) -> &'static str {
1333	match mark {
1334	InboxMark::Read => "read_at = COALESCE(read_at, ?1)",
1335	InboxMark::Unread => "read_at = NULL",
1336	InboxMark::Done => "done_at = ?1, read_at = COALESCE(read_at, ?1)",
1337	InboxMark::Undone => "done_at = NULL",
1338	InboxMark::Save => "saved = 1",
1339	InboxMark::Unsave => "saved = 0",
1340	InboxMark::Snooze => "snoozed_until = ?2, read_at = COALESCE(read_at, ?1)",
1341	InboxMark::Unsnooze => "snoozed_until = NULL",
1342	}
1343	}
1344
1345	#[derive(Deserialize)]
1346	struct IdRow {
1347	#[allow(dead_code)]
1348	id: String,
1349	}
1350
1351	/// Changes the person's own items. Returns how many changed.
1352	pub async fn mark(db: &D1Database, a: MarkInboxArgs) -> Result<u32> {
1353	let now = rfc3339(now_ms());
1354	let until = match (a.mark, a.until.as_deref().map(str::trim)) {
1355	// Times compare as text (`g1t_contracts::time`); a snooze is for later.
1356	(InboxMark::Snooze, Some(until)) if until.len() == now.len() && until > now.as_str() => until.to_owned(),
1357	(InboxMark::Snooze, _) => return Ok(0),
1358	_ => String::new(),
1359	};
1360	let mut values: Vec<JsValue> = vec![now.as_str().into(), until.as_str().into(), a.username.to_lowercase().into()];
1361	let target = if a.all && a.ids.is_empty() {
1362	let mut target = "done_at IS NULL".to_owned();
1363	let mut bind = \|values: &mut Vec<JsValue>, value: &str, condition: &str\| {
1364	values.push(value.into());
1365	target.push_str(&format!(" AND {}", condition.replace('?', &format!("?{}", values.len()))));
1366	};
1367	if let Some(severity) = a.severity {
1368	bind(&mut values, severity.as_str(), "severity = ?");
1369	}
1370	if let Some(repo_id) = &a.repo_id {
1371	bind(&mut values, repo_id, "repo_id = ?");
1372	}
1373	if let Some(last_read_at) = a.last_read_at.as_deref().map(str::trim).filter(\|at\| !at.is_empty()) {
1374	bind(&mut values, last_read_at, "COALESCE(updated_at, created_at) <= ?");
1375	}
1376	target
1377	} else {
1378	let ids: Vec<&String> = a.ids.iter().take(MAX_MARK).collect();
1379	if ids.is_empty() {
1380	return Ok(0);
1381	}
1382	let first = values.len() + 1;
1383	values.extend(ids.iter().map(\|id\| JsValue::from(id.as_str())));
1384	let marks: Vec<String> = (first..values.len() + 1).map(\|at\| format!("?{at}")).collect();
1385	format!("id IN ({})", marks.join(", "))
1386	};
1387	let changed = db
1388	.prepare(format!(
1389	"UPDATE inbox_items SET {} WHERE username = ?3 AND {target} RETURNING id",
1390	mark_change(a.mark)
1391	))
1392	.bind(&values)?
1393	.all()
1394	.await?
1395	.results::<IdRow>()?;
1396	Ok(changed.len() as u32)
1397	}
1398
1399	#[derive(Deserialize)]
1400	struct ActivityRow {
1401	reason: String,
1402	severity: String,
1403	title: String,
1404	body: String,
1405	event: Option<String>,
1406	actor: Option<String>,
1407	created_at: String,
1408	}
1409
1410	/// One of the viewer's threads, with its history and their subscription.
1411	pub async fn thread(db: &D1Database, repos: &Fetcher, work: &Fetcher, a: ThreadArgs) -> Result<Option<InboxThread>> {
1412	let Some(viewer) = &a.viewer else {
1413	return Ok(None);
1414	};
1415	let username = viewer.username.to_lowercase();
1416	let row = db
1417	.prepare(format!("SELECT {COLUMNS} FROM inbox_items WHERE id = ? AND username = ?"))
1418	.bind(&[a.id.as_str().into(), username.as_str().into()])?
1419	.first::<Row>(None)
1420	.await?;
1421	let Some(row) = readable_only(db, repos, &a.viewer, &username, row.into_iter().collect()).await?.pop() else {
1422	return Ok(None);
1423	};
1424	let activity = db
1425	.prepare(
1426	"SELECT reason, severity, title, body, event, actor, created_at FROM inbox_activity
1427	WHERE item_id = ? ORDER BY id DESC LIMIT ?",
1428	)
1429	.bind(&[row.id.as_str().into(), MAX_ACTIVITY.into()])?
1430	.all()
1431	.await?
1432	.results::<ActivityRow>()?
1433	.into_iter()
1434	.map(\|row\| InboxActivity {
1435	reason: Reason::parse(&row.reason).unwrap_or(Reason::Subscribed),
1436	severity: Severity::parse(&row.severity).unwrap_or(Severity::Info),
1437	title: row.title,
1438	body: row.body,
1439	event: row.event,
1440	actor: row.actor,
1441	created_at: row.created_at,
1442	})
1443	.collect();
1444	let item = row.into_item();
1445	let subscription = match (item.subject, item.number) {
1446	(Some(SubjectKind::Issue \| SubjectKind::Pull), Some(_)) => {
1447	subscriptions::subscription(
1448	db,
1449	work,
1450	SubscriptionArgs {
1451	viewer: a.viewer.clone(),
1452	id: Some(item.id.clone()),
1453	..SubscriptionArgs::default()
1454	},
1455	)
1456	.await?
1457	}
1458	_ => None,
1459	};
1460	Ok(Some(InboxThread {
1461	item,
1462	activity,
1463	subscription,
1464	}))
1465	}
1466
1467	// --- Keeping up ------------------------------------------------------------------
1468
1469	/// Moves rows with renamed workspaces and repositories, and drops those
1470	/// of purged repositories and deleted workspaces.
1471	pub async fn follow(db: &D1Database, events: &[Event]) -> Result<()> {
1472	let mut statements = Vec::new();
1473	for event in events {
1474	let text = \|key: &str\| event.data[key].as_str().unwrap_or_default().to_lowercase();
1475	match event.kind.as_str() {
1476	"workspace.renamed" => {
1477	let Ok(renamed) = serde_json::from_value::<WorkspaceRenamed>(event.data.clone()) else {
1478	continue;
1479	};
1480	let (from, to) = (renamed.from.to_lowercase(), renamed.to.to_lowercase());
1481	if from == to {
1482	continue;
1483	}
1484	statements.push(
1485	db.prepare(
1486	"UPDATE inbox_items SET repo = ?2 \|\| substr(repo, length(?1) + 1), workspace = ?2,
1487	link = CASE WHEN link LIKE '/' \|\| ?1 \|\| '/%' THEN '/' \|\| ?2 \|\| substr(link, length(?1) + 2) ELSE link END
1488	WHERE workspace = ?1",
1489	)
1490	.bind(&[from.as_str().into(), to.as_str().into()])?,
1491	);
1492	statements.push(
1493	db.prepare("UPDATE inbox_watching SET repo = ?2 \|\| substr(repo, length(?1) + 1) WHERE repo LIKE ?1 \|\| '/%'")
1494	.bind(&[from.as_str().into(), to.as_str().into()])?,
1495	);
1496	}
1497	"repo.renamed" => {
1498	let path = format!("{}/{}", text("namespace"), text("to"));
1499	for table in ["inbox_items", "inbox_watching"] {
1500	statements.push(
1501	db.prepare(format!("UPDATE {table} SET repo = ? WHERE repo_id = ?"))
1502	.bind(&[path.as_str().into(), text("repoId").into()])?,
1503	);
1504	}
1505	}
1506	"repo.transferred" => {
1507	let path = format!("{}/{}", text("to"), text("name"));
1508	statements.push(
1509	db.prepare("UPDATE inbox_items SET repo = ?, workspace = ? WHERE repo_id = ?")
1510	.bind(&[path.as_str().into(), text("to").into(), text("repoId").into()])?,
1511	);
1512	statements.push(
1513	db.prepare("UPDATE inbox_watching SET repo = ? WHERE repo_id = ?")
1514	.bind(&[path.as_str().into(), text("repoId").into()])?,
1515	);
1516	}
1517	"repo.purged" => {
1518	for sql in [
1519	"DELETE FROM inbox_activity WHERE item_id IN (SELECT id FROM inbox_items WHERE repo_id = ?)",
1520	"DELETE FROM inbox_items WHERE repo_id = ?",
1521	"DELETE FROM inbox_subscriptions WHERE repo_id = ?",
1522	"DELETE FROM inbox_watching WHERE repo_id = ?",
1523	] {
1524	statements.push(db.prepare(sql).bind(&[text("repoId").into()])?);
1525	}
1526	}
1527	"workspace.deleted" => {
1528	statements.push(
1529	db.prepare("DELETE FROM inbox_items WHERE workspace = ?")
1530	.bind(&[text("slug").into()])?,
1531	);
1532	statements.push(
1533	db.prepare("DELETE FROM inbox_watching WHERE repo LIKE ? \|\| '/%'")
1534	.bind(&[text("slug").into()])?,
1535	);
1536	}
1537	_ => {}
1538	}
1539	}
1540	if !statements.is_empty() {
1541	db.batch(statements).await?;
1542	}
1543	Ok(())
1544	}
1545
1546	/// Removes items done more than [`DONE_DAYS`] ago, and any not saved older
1547	/// than [`MAX_DAYS`], with their history. Returns how many went.
1548	pub async fn purge(db: &D1Database, now: u64) -> Result<u32> {
1549	let done = crate::audit::keep_from(now, DONE_DAYS);
1550	let oldest = crate::audit::keep_from(now, MAX_DAYS);
1551	let statements: Vec<D1PreparedStatement> = vec![
1552	db.prepare("DELETE FROM inbox_items WHERE saved = 0 AND done_at < ?")
1553	.bind(&[done.into()])?,
1554	db.prepare("DELETE FROM inbox_items WHERE saved = 0 AND COALESCE(updated_at, created_at) < ?")
1555	.bind(&[oldest.into()])?,
1556	db.prepare("DELETE FROM inbox_activity WHERE NOT EXISTS (SELECT 1 FROM inbox_items WHERE inbox_items.id = inbox_activity.item_id)"),
1557	];
1558	let results = db.batch(statements).await?;
1559	let mut removed = 0;
1560	for result in results.into_iter().take(2) {
1561	removed += result.meta()?.and_then(\|meta\| meta.changes).unwrap_or(0) as u32;
1562	}
1563	Ok(removed)
1564	}
1565
1566	#[cfg(test)]
1567	mod tests {
1568	use super::*;
1569	use crate::subscriptions::{State, Subscription, Watcher};
1570	use serde_json::json;
1571
1572	fn event(kind: &str, actor: Option<&str>, data: serde_json::Value) -> Event {
1573	Event {
1574	id: "evt_1".into(),
1575	kind: kind.into(),
1576	source: "work".into(),
1577	time: "2026-10-07T12:00:00.000Z".into(),
1578	repo_id: Some("rep_1".into()),
1579	actor: actor.map(str::to_owned),
1580	data,
1581	}
1582	}
1583
1584	fn person(id: &str, username: &str) -> Principal {
1585	Principal {
1586	id: id.into(),
1587	username: username.into(),
1588	}
1589	}
1590
1591	fn actor(id: &str, username: &str) -> Actor {
1592	Actor {
1593	id: Some(id.into()),
1594	username: Some(username.into()),
1595	}
1596	}
1597
1598	/// A pull request ana opened, assigned to bo, for an issue cy filed and dee is assigned.
1599	fn pull() -> InboxSubject {
1600	InboxSubject {
1601	kind: Some(SubjectKind::Pull),
1602	title: "Add the inbox".into(),
1603	author: person("usr_ana", "ana"),
1604	assignees: vec!["bo".into()],
1605	issue: Some(Box::new(InboxSubject {
1606	kind: Some(SubjectKind::Issue),
1607	title: "An inbox".into(),
1608	author: person("usr_cy", "cy"),
1609	assignees: vec!["dee".into()],
1610	..InboxSubject::default()
1611	})),
1612	..InboxSubject::default()
1613	}
1614	}
1615
1616	/// A change g1t made for ana.
1617	fn g1t_pull() -> InboxSubject {
1618	InboxSubject {
1619	author: person(AGENT_ID, "g1t"),
1620	requested_by: Some(person("usr_ana", "ana")),
1621	..pull()
1622	}
1623	}
1624
1625	fn nobody() -> Audience {
1626	Audience::default()
1627	}
1628
1629	fn told(notices: &[Notice]) -> Vec<(&str, Reason, Severity)> {
1630	notices
1631	.iter()
1632	.map(\|notice\| (notice.username.as_str(), notice.reason, notice.severity))
1633	.collect()
1634	}
1635
1636	fn comment(author: Principal, body: &str, mentions: &[&str]) -> InboxSubject {
1637	InboxSubject {
1638	comment: Some(InboxComment {
1639	author,
1640	excerpt: body.into(),
1641	mentions: mentions.iter().map(\|name\| (*name).to_owned()).collect(),
1642	..InboxComment::default()
1643	}),
1644	..pull()
1645	}
1646	}
1647
1648	fn subscribed(rows: &[(&str, State, Option<Reason>)]) -> Audience {
1649	Audience {
1650	subscriptions: rows
1651	.iter()
1652	.map(\|(name, state, reason)\| Subscription {
1653	username: (*name).into(),
1654	state: *state,
1655	reason: *reason,
1656	})
1657	.collect(),
1658	watchers: Vec::new(),
1659	}
1660	}
1661
1662	fn watched(rows: &[(&str, WatchLevel, &[&str])]) -> Audience {
1663	Audience {
1664	subscriptions: Vec::new(),
1665	watchers: rows
1666	.iter()
1667	.map(\|(name, level, events)\| Watcher {
1668	username: (*name).into(),
1669	level: *level,
1670	events: events.iter().map(\|kind\| (*kind).to_owned()).collect(),
1671	})
1672	.collect(),
1673	}
1674	}
1675
1676	#[test]
1677	fn only_events_that_tell_someone_are_read() {
1678	let asked = \|kind: &str, data\| wants(&event(kind, None, data));
1679	assert_eq!(
1680	asked("checks.completed", json!({ "repoId": "rep_1", "number": 4, "status": "failed" })),
1681	Some(Wanted { repo_id: "rep_1".into(), number: Some(4), comment_id: None })
1682	);
1683	assert_eq!(asked("checks.completed", json!({ "repoId": "rep_1", "number": 4, "status": "passed" })), None);
1684	assert_eq!(asked("review.completed", json!({ "repoId": "rep_1", "number": 4 })), None);
1685	assert_eq!(asked("workflow.completed", json!({ "repoId": "rep_1", "conclusion": "success" })), None);
1686	assert_eq!(
1687	asked("workflow.completed", json!({ "repoId": "rep_1", "conclusion": "failure" })),
1688	Some(Wanted { repo_id: "rep_1".into(), number: None, comment_id: None })
1689	);
1690	assert_eq!(
1691	asked("comment.created", json!({ "repoId": "rep_1", "number": 2, "commentId": "cmt_1" })).map(\|w\| w.comment_id),
1692	Some(Some("cmt_1".into()))
1693	);
1694	assert_eq!(asked("git.push", json!({ "repoId": "rep_1" })), None);
1695	for kind in ["issue.opened", "pull.review_requested", "pull.stalled", "issue.assigned", "pull.closed", "issue.reopened"] {
1696	assert!(asked(kind, json!({ "repoId": "rep_1", "number": 1 })).is_some(), "{kind}");
1697	}
1698	assert_eq!(asked("deployment.failed", json!({ "repoId": "rep_1" })).map(\|w\| w.number), Some(None));
1699	}
1700
1701	#[test]
1702	fn a_waiting_or_stopped_agent_needs_the_pull_requests_and_the_issues_people_first() {
1703	let asked = event("agent.asked", Some("usr_agent"), json!({ "number": 7 }));
1704	let notices_ = notices(&asked, "acme/rocket", &Actor::default(), Some(&pull()), &nobody());
1705	assert_eq!(
1706	told(&notices_),
1707	vec![
1708	("ana", Reason::Agent, Severity::Warning),
1709	("cy", Reason::Agent, Severity::Warning),
1710	("dee", Reason::Agent, Severity::Warning),
1711	]
1712	);
1713	assert_eq!(notices_[0].title, "An agent is waiting on acme/rocket#7");
1714	assert_eq!(notices_[0].body, "Add the inbox");
1715	// Stopping says why; even someone who unsubscribed hears of it.
1716	let stalled = event("pull.stalled", None, json!({ "number": 7, "detail": "Its checks could not be run." }));
1717	let audience = subscribed(&[("ana", State::Unsubscribed, None)]);
1718	let notices_ = notices(&stalled, "acme/rocket", &Actor::default(), Some(&g1t_pull()), &audience);
1719	assert_eq!(notices_[0].username, "ana");
1720	assert_eq!(notices_[0].title, "g1t stopped on acme/rocket#7 and needs you");
1721	assert_eq!(notices_[0].body, "Its checks could not be run.");
1722	// Ignoring the thread silences even that.
1723	let audience = subscribed(&[("ana", State::Ignored, None)]);
1724	let notices_ = notices(&stalled, "acme/rocket", &Actor::default(), Some(&g1t_pull()), &audience);
1725	assert!(!notices_.iter().any(\|notice\| notice.username == "ana"));
1726	}
1727
1728	#[test]
1729	fn whatever_an_agent_waits_on_closes_when_it_goes_on() {
1730	for kind in RESUMES {
1731	assert_eq!(resolves(&event(kind, None, json!({ "repoId": "rep_1", "number": 7 }))), Some("rep_1#7".into()));
1732	}
1733	assert_eq!(resolves(&event("comment.created", None, json!({ "repoId": "rep_1", "number": 7 }))), None);
1734	}
1735
1736	#[test]
1737	fn a_deployments_reviewers_are_told() {
1738	let asked = event(
1739	"deployment.review_requested",
1740	Some("usr_ana"),
1741	json!({
1742	"repoId": "rep_1", "runId": "run_7", "environment": "production", "workflow": "Deploy",
1743	"title": "Ship it", "notify": ["cy", "ana"], "link": "/acme/rocket/actions/runs/run_7"
1744	}),
1745	);
1746	let wanted = wants(&asked).unwrap();
1747	assert_eq!(wanted.number, None);
1748	let thread = thread_of(&asked, &wanted, None);
1749	assert_eq!(thread.key, "rep_1/review/run_7/production");
1750	assert_eq!(thread.link.as_deref(), Some("/acme/rocket/actions/runs/run_7"));
1751	let told = notices(&asked, "acme/rocket", &actor("usr_ana", "ana"), None, &nobody());
1752	let names: Vec<&str> = told.iter().map(\|notice\| notice.username.as_str()).collect();
1753	// Whoever started the run is told too, when they review it.
1754	assert_eq!(names, ["cy", "ana"]);
1755	assert!(told[0].title.contains("waiting for your review to deploy to production"));
1756	}
1757
1758	#[test]
1759	fn teams_asked_to_review_tell_the_people_they_name() {
1760	let requested = event(
1761	"pull.review_requested",
1762	Some("usr_ana"),
1763	json!({ "number": 7, "reviewers": ["cy"], "teams": [
1764	{ "team": "acme/backend", "notified": ["cy", "dee", "ana"], "assigned": ["cy"] }
1765	] }),
1766	);
1767	let notices_ = notices(&requested, "acme/rocket", &actor("usr_ana", "ana"), Some(&pull()), &nobody());
1768	// cy was picked and is told as a reviewer; dee through the team;
1769	// never whoever asked.
1770	assert_eq!(
1771	told(&notices_),
1772	vec![("cy", Reason::ReviewRequested, Severity::Warning), ("dee", Reason::ReviewRequested, Severity::Warning)]
1773	);
1774	assert_eq!(notices_[0].title, "ana asked you to review acme/rocket#7");
1775	assert_eq!(notices_[1].title, "ana asked @acme/backend to review acme/rocket#7");
1776	assert_eq!(
1777	subscribes(&requested, Some(&pull())),
1778	vec![("cy".into(), Reason::ReviewRequested), ("dee".into(), Reason::ReviewRequested), ("ana".into(), Reason::ReviewRequested)]
1779	);
1780	// Asked by the CODEOWNERS file.
1781	let owned = event(
1782	"pull.review_requested",
1783	None,
1784	json!({ "number": 7, "reviewers": ["bo"], "codeOwners": true, "teams": [{ "team": "acme/docs", "notified": ["wren"], "assigned": [] }] }),
1785	);
1786	let notices_ = notices(&owned, "acme/rocket", &Actor::default(), Some(&pull()), &nobody());
1787	assert_eq!(notices_[0].title, "acme/rocket#7 changes files you own");
1788	assert_eq!(notices_[1].title, "acme/rocket#7 changes files @acme/docs owns");
1789	}
1790
1791	#[test]
1792	fn a_team_mention_tells_its_people_once_and_a_name_wins() {
1793	let mut on = comment(person("usr_bo", "bo"), "cc @ana @acme/backend", &["ana"]);
1794	if let Some(comment) = on.comment.as_mut() {
1795	comment.team_mentions = vec![TeamMentioned {
1796	team: "acme/backend".into(),
1797	members: vec!["ana".into(), "cy".into(), "bo".into()],
1798	}];
1799	}
1800	let created = event("comment.created", Some("usr_bo"), json!({ "number": 7, "commentId": "cmt_1" }));
1801	let notices_ = notices(&created, "acme/rocket", &actor("usr_bo", "bo"), Some(&on), &nobody());
1802	let mentioned: Vec<(&str, Reason)> = notices_.iter().map(\|n\| (n.username.as_str(), n.reason)).filter(\|(_, r)\| matches!(r, Reason::Mention \| Reason::TeamMention)).collect();
1803	assert_eq!(mentioned, vec![("ana", Reason::Mention), ("cy", Reason::TeamMention)]);
1804	assert_eq!(
1805	notices_.iter().find(\|n\| n.username == "cy").unwrap().title,
1806	"bo mentioned @acme/backend on acme/rocket#7"
1807	);
1808	let subscribed = subscribes(&created, Some(&on));
1809	assert!(subscribed.contains(&("cy".into(), Reason::TeamMention)));
1810	assert!(subscribed.contains(&("ana".into(), Reason::Mention)));
1811	}
1812
1813	#[test]
1814	fn a_description_tells_the_people_and_teams_it_mentions() {
1815	let mut on = pull();
1816	on.mentions = vec!["dee".into()];
1817	on.team_mentions = vec![TeamMentioned {
1818	team: "acme/web".into(),
1819	members: vec!["eve".into()],
1820	}];
1821	let opened = event("pull.opened", Some("usr_ana"), json!({ "number": 7 }));
1822	let notices_ = notices(&opened, "acme/rocket", &actor("usr_ana", "ana"), Some(&on), &nobody());
1823	assert!(told(&notices_).contains(&("dee", Reason::Mention, Severity::Info)));
1824	assert!(told(&notices_).contains(&("eve", Reason::TeamMention, Severity::Info)));
1825	assert_eq!(
1826	subscribes(&opened, Some(&on)),
1827	vec![("dee".into(), Reason::Mention), ("eve".into(), Reason::TeamMention)]
1828	);
1829	}
1830
1831	#[test]
1832	fn reviewers_and_assignees_asked_are_told_never_whoever_asked() {
1833	let requested = event("pull.review_requested", Some("usr_ana"), json!({ "number": 7, "reviewers": ["bo", "g1t", "ana"] }));
1834	let notices_ = notices(&requested, "acme/rocket", &actor("usr_ana", "ana"), Some(&pull()), &nobody());
1835	assert_eq!(told(&notices_), vec![("bo", Reason::ReviewRequested, Severity::Warning)]);
1836	assert_eq!(notices_[0].title, "ana asked you to review acme/rocket#7");
1837	let assigned = event("issue.assigned", Some("usr_ana"), json!({ "number": 3, "added": ["ana", "eve"] }));
1838	let notices_ = notices(&assigned, "acme/rocket", &actor("usr_ana", "ana"), Some(&pull()), &nobody());
1839	assert_eq!(told(&notices_), vec![("eve", Reason::Assign, Severity::Info)]);
1840	assert_eq!(notices_[0].title, "ana assigned you to acme/rocket#3");
1841	// Both subscribe whoever they name, never g1t.
1842	assert_eq!(subscribes(&requested, Some(&pull())), vec![("bo".into(), Reason::ReviewRequested), ("ana".into(), Reason::ReviewRequested)]);
1843	assert_eq!(subscribes(&assigned, Some(&pull())), vec![("ana".into(), Reason::Assign), ("eve".into(), Reason::Assign)]);
1844	}
1845
1846	#[test]
1847	fn failures_go_to_whoever_answers_for_the_change() {
1848	let failed = event("checks.completed", None, json!({ "number": 7, "status": "failed" }));
1849	assert_eq!(
1850	told(&notices(&failed, "acme/rocket", &Actor::default(), Some(&pull()), &nobody())),
1851	vec![("ana", Reason::CiActivity, Severity::Error)]
1852	);
1853	// g1t's change is the person's who asked for it, never g1t's.
1854	let notices_ = notices(&failed, "acme/rocket", &Actor::default(), Some(&g1t_pull()), &nobody());
1855	assert_eq!(told(&notices_), vec![("ana", Reason::CiActivity, Severity::Error)]);
1856	let errored = event("checks.completed", None, json!({ "number": 7, "status": "errored" }));
1857	assert_eq!(
1858	notices(&errored, "acme/rocket", &Actor::default(), Some(&pull()), &nobody())[0].title,
1859	"Checks could not run on acme/rocket#7"
1860	);
1861	}
1862
1863	#[test]
1864	fn a_workflow_that_fails_tells_its_pull_requests_owner_even_if_they_pushed() {
1865	let failed = event("workflow.completed", Some("usr_ana"), json!({ "pull": 7, "workflow": "CI", "conclusion": "failure" }));
1866	let notices_ = notices(&failed, "acme/rocket", &actor("usr_ana", "ana"), Some(&pull()), &nobody());
1867	assert_eq!(told(&notices_), vec![("ana", Reason::CiActivity, Severity::Error)]);
1868	assert_eq!(notices_[0].title, "CI failed on acme/rocket#7");
1869	// On a branch: whoever pushed.
1870	let pushed = event(
1871	"workflow.completed",
1872	Some("usr_bo"),
1873	json!({ "workflow": "Deploy", "path": ".g1t/workflows/deploy.yml", "conclusion": "failure", "ref": "refs/heads/main", "number": 12, "sha": "abcdef0123", "runId": "run_9" }),
1874	);
1875	let notices_ = notices(&pushed, "acme/rocket", &actor("usr_bo", "bo"), None, &nobody());
1876	assert_eq!(told(&notices_), vec![("bo", Reason::CiActivity, Severity::Error)]);
1877	assert_eq!(notices_[0].title, "Deploy failed on main in acme/rocket");
1878	assert_eq!(notices_[0].body, "Run 12 at abcdef0");
1879	// Nobody to tell when g1t pushed.
1880	assert!(notices(&pushed, "acme/rocket", &actor("g1t", "g1t"), None, &nobody()).is_empty());
1881	// Every failure of a workflow on a branch is one thread.
1882	let wanted = wants(&pushed).unwrap();
1883	let thread = thread_of(&pushed, &wanted, None);
1884	assert_eq!(thread.key, "rep_1/run/.g1t/workflows/deploy.yml@main");
1885	assert_eq!(thread.run_id.as_deref(), Some("run_9"));
1886	}
1887
1888	#[test]
1889	fn deployments_tell_whoever_answers_for_them_and_watchers() {
1890	let failed = event(
1891	"deployment.failed",
1892	Some("usr_bo"),
1893	json!({ "repoId": "rep_1", "projectId": "prj_1", "project": "rocket", "deploymentId": "dpl_1", "error": "The build failed.", "path": "/acme/rocket/deployments/dpl_1" }),
1894	);
1895	let audience = watched(&[("cy", WatchLevel::Custom, &["deployments"]), ("dee", WatchLevel::Custom, &["issues"]), ("eve", WatchLevel::All, &[])]);
1896	let notices_ = notices(&failed, "acme/rocket", &actor("usr_bo", "bo"), None, &audience);
1897	assert_eq!(
1898	told(&notices_),
1899	vec![
1900	("bo", Reason::CiActivity, Severity::Error),
1901	("cy", Reason::Subscribed, Severity::Error),
1902	("eve", Reason::Subscribed, Severity::Error),
1903	]
1904	);
1905	assert_eq!(notices_[0].title, "Production of rocket failed to deploy");
1906	assert_eq!(notices_[0].body, "The build failed.");
1907	let thread = thread_of(&failed, &wants(&failed).unwrap(), None);
1908	assert_eq!(thread.key, "rep_1/deploy/prj_1/production");
1909	assert_eq!(url(Some("acme/rocket"), thread.kind, None, None, thread.link.as_deref()), "/acme/rocket/deployments/dpl_1");
1910	// A success is news to the owner only after a failure.
1911	let live = event("deployment.succeeded", Some("usr_bo"), json!({ "repoId": "rep_1", "projectId": "prj_1", "project": "rocket", "commit": "abcdef0123" }));
1912	assert!(notices(&live, "acme/rocket", &actor("usr_bo", "bo"), None, &nobody()).is_empty());
1913	let recovered = event("deployment.succeeded", Some("usr_bo"), json!({ "repoId": "rep_1", "projectId": "prj_1", "project": "rocket", "recovered": true }));
1914	let notices_ = notices(&recovered, "acme/rocket", &actor("usr_bo", "bo"), None, &nobody());
1915	assert_eq!(told(&notices_), vec![("bo", Reason::CiActivity, Severity::Success)]);
1916	assert_eq!(notices_[0].title, "Production of rocket is live again");
1917	// A preview's is the pull request's owner's.
1918	let preview = event("deployment.failed", None, json!({ "repoId": "rep_1", "projectId": "prj_1", "number": 7, "triggeredBy": "g1t" }));
1919	assert_eq!(told(&notices(&preview, "acme/rocket", &Actor::default(), Some(&pull()), &nobody())), vec![("ana", Reason::CiActivity, Severity::Error)]);
1920	assert_eq!(thread_of(&preview, &wants(&preview).unwrap(), Some(&pull())).key, "rep_1/deploy/prj_1/7");
1921	}
1922
1923	#[test]
1924	fn security_alerts_tell_the_pusher_the_owners_and_member_watchers() {
1925	let blocked = event(
1926	"secret_scanning_alert.created",
1927	None,
1928	json!({
1929	"repoId": "rep_1", "alertId": "sec_1", "alertType": "secret_scanning", "severity": "critical",
1930	"title": "An AWS access key in config/prod.env", "link": "/acme/rocket/security/secret-scanning/sec_1",
1931	"state": "open", "pusher": "bo", "notify": ["ana"], "members": ["ana", "bo", "cy"]
1932	}),
1933	);
1934	assert_eq!(wants(&blocked), Some(Wanted { repo_id: "rep_1".into(), number: None, comment_id: None }));
1935	// cy watches for security alerts and is a member; eve watches
1936	// everything but cannot see findings; dee watches issues only.
1937	let audience = watched(&[
1938	("cy", WatchLevel::Custom, &["security"]),
1939	("dee", WatchLevel::Custom, &["issues"]),
1940	("eve", WatchLevel::All, &[]),
1941	]);
1942	let notices_ = notices(&blocked, "acme/rocket", &Actor::default(), None, &audience);
1943	assert_eq!(
1944	told(&notices_),
1945	vec![
1946	("bo", Reason::SecurityAlert, Severity::Error),
1947	("ana", Reason::SecurityAlert, Severity::Error),
1948	("cy", Reason::SecurityAlert, Severity::Error),
1949	]
1950	);
1951	assert_eq!(notices_[0].title, "A push to acme/rocket was blocked: it adds a secret");
1952	assert_eq!(notices_[0].body, "An AWS access key in config/prod.env");
1953	let thread = thread_of(&blocked, &wants(&blocked).unwrap(), None);
1954	assert_eq!(thread.key, "rep_1/security/sec_1");
1955	assert_eq!(url(Some("acme/rocket"), thread.kind, None, None, thread.link.as_deref()), "/acme/rocket/security/secret-scanning/sec_1");
1956	// A bypass request goes to the reviewers named, never to watchers.
1957	let requested = event(
1958	"secret_scanning.bypass_requested",
1959	Some("usr_bo"),
1960	json!({ "repoId": "rep_1", "alertId": "sec_1", "requestId": "byp_1", "title": "An AWS access key in a.env", "notify": ["ana"], "link": "/acme/-/security/bypass-requests" }),
1961	);
1962	let notices_ = notices(&requested, "acme/rocket", &actor("usr_bo", "bo"), None, &audience);
1963	assert_eq!(told(&notices_), vec![("ana", Reason::SecurityAlert, Severity::Warning)]);
1964	assert_eq!(notices_[0].title, "bo asked to bypass push protection in acme/rocket");
1965	// Fixes and dismissals are not news to the inbox.
1966	assert_eq!(wants(&event("code_scanning_alert.fixed", None, json!({ "repoId": "rep_1" }))), None);
1967	}
1968
1969	#[test]
1970	fn nobody_hears_of_what_they_did_themselves() {
1971	let merged = event("pull.merged", Some("usr_ana"), json!({ "number": 7 }));
1972	let by_ana = notices(&merged, "acme/rocket", &actor("usr_ana", "ana"), Some(&pull()), &nobody());
1973	// The assignee still hears; ana, who merged, does not.
1974	assert_eq!(told(&by_ana), vec![("bo", Reason::StateChange, Severity::Success)]);
1975	let by_bo = notices(&merged, "acme/rocket", &actor("usr_bo", "bo"), Some(&pull()), &nobody());
1976	assert_eq!(told(&by_bo), vec![("ana", Reason::StateChange, Severity::Success)]);
1977	assert_eq!(by_bo[0].title, "bo merged acme/rocket#7");
1978	let by_queue = notices(&merged, "acme/rocket", &Actor::default(), Some(&pull()), &nobody());
1979	assert_eq!(by_queue[0].title, "acme/rocket#7 was merged");
1980	}
1981
1982	#[test]
1983	fn closing_and_reopening_tells_everyone_subscribed_and_watchers() {
1984	let closed = event("issue.closed", Some("usr_bo"), json!({ "number": 3, "reason": "completed" }));
1985	let issue = InboxSubject {
1986	kind: Some(SubjectKind::Issue),
1987	title: "Crash".into(),
1988	author: person("usr_cy", "cy"),
1989	assignees: vec!["bo".into()],
1990	..InboxSubject::default()
1991	};
1992	let mut audience = subscribed(&[("eve", State::Subscribed, Some(Reason::Comment)), ("fay", State::Unsubscribed, None)]);
1993	audience.watchers = watched(&[("gus", WatchLevel::All, &[]), ("hal", WatchLevel::Custom, &["pulls"])]).watchers;
1994	let notices_ = notices(&closed, "acme/rocket", &actor("usr_bo", "bo"), Some(&issue), &audience);
1995	assert_eq!(
1996	told(&notices_),
1997	vec![
1998	("cy", Reason::StateChange, Severity::Info),
1999	("eve", Reason::StateChange, Severity::Info),
2000	("gus", Reason::Subscribed, Severity::Info),
2001	]
2002	);
2003	assert_eq!(notices_[0].title, "bo closed acme/rocket#3");
2004	let by_pull = event("issue.closed", None, json!({ "number": 3, "resolvedBy": 9 }));
2005	assert_eq!(notices(&by_pull, "acme/rocket", &Actor::default(), Some(&issue), &nobody())[0].title, "acme/rocket#3 was closed by #9");
2006	let reopened = event("issue.reopened", Some("usr_cy"), json!({ "number": 3 }));
2007	assert_eq!(told(&notices(&reopened, "acme/rocket", &actor("usr_cy", "cy"), Some(&issue), &nobody())), vec![("bo", Reason::StateChange, Severity::Info)]);
2008	}
2009
2010	#[test]
2011	fn opening_tells_who_it_names_and_watchers_of_its_kind() {
2012	let opened = event("pull.opened", Some("usr_ana"), json!({ "number": 7 }));
2013	let subject = InboxSubject { reviewers: vec!["cy".into(), "g1t".into()], ..pull() };
2014	let audience = watched(&[("bo", WatchLevel::All, &[]), ("dee", WatchLevel::Custom, &["issues"]), ("eve", WatchLevel::Custom, &["pulls"]), ("fay", WatchLevel::Participating, &[])]);
2015	let notices_ = notices(&opened, "acme/rocket", &actor("usr_ana", "ana"), Some(&subject), &audience);
2016	assert_eq!(
2017	told(&notices_),
2018	vec![
2019	("bo", Reason::Assign, Severity::Info),
2020	("cy", Reason::ReviewRequested, Severity::Warning),
2021	("eve", Reason::Subscribed, Severity::Info),
2022	]
2023	);
2024	assert_eq!(notices_[2].title, "ana opened acme/rocket#7");
2025	}
2026
2027	#[test]
2028	fn ignoring_a_repository_silences_it_and_unsubscribing_keeps_only_what_is_asked() {
2029	let created = event("comment.created", Some("usr_bo"), json!({ "number": 7, "commentId": "cmt_1" }));
2030	let on = comment(person("usr_bo", "bo"), "@cy have a look", &["cy"]);
2031	let audience = watched(&[("cy", WatchLevel::Ignore, &[]), ("ana", WatchLevel::All, &[])]);
2032	assert!(notices(&created, "acme/rocket", &actor("usr_bo", "bo"), Some(&on), &audience).iter().all(\|notice\| notice.username != "cy"));
2033	let audience = subscribed(&[("cy", State::Unsubscribed, None), ("ana", State::Unsubscribed, None)]);
2034	let notices_ = notices(&created, "acme/rocket", &actor("usr_bo", "bo"), Some(&on), &audience);
2035	// cy was mentioned, which is asked of them; ana unsubscribed from the conversation.
2036	assert_eq!(told(&notices_), vec![("cy", Reason::Mention, Severity::Info)]);
2037	}
2038
2039	#[test]
2040	fn g1t_finishing_or_reviewing_tells_the_person_it_worked_for() {
2041	// The agent acts as the person it works for: still an outcome they hear of.
2042	let ready = event("pull.ready", Some("usr_ana"), json!({ "number": 7 }));
2043	let notices_ = notices(&ready, "acme/rocket", &actor("usr_ana", "ana"), Some(&g1t_pull()), &nobody());
2044	assert_eq!(told(&notices_), vec![("ana", Reason::Author, Severity::Success)]);
2045	assert_eq!(notices_[0].title, "g1t finished acme/rocket#7");
2046	// A person's draft marked ready is not news to them.
2047	assert!(notices(&ready, "acme/rocket", &actor("usr_ana", "ana"), Some(&pull()), &nobody()).is_empty());
2048
2049	let approve = event("review.completed", None, json!({ "number": 7, "verdict": "approve" }));
2050	assert_eq!(
2051	told(&notices(&approve, "acme/rocket", &Actor::default(), Some(&pull()), &nobody())),
2052	vec![("ana", Reason::Author, Severity::Success)]
2053	);
2054	let changes = event("review.completed", None, json!({ "number": 7, "verdict": "request_changes" }));
2055	assert_eq!(
2056	told(&notices(&changes, "acme/rocket", &Actor::default(), Some(&pull()), &nobody())),
2057	vec![("ana", Reason::Author, Severity::Info)]
2058	);
2059	}
2060
2061	#[test]
2062	fn comments_tell_those_mentioned_then_everyone_subscribed_never_the_writer() {
2063	let created = event("comment.created", Some("usr_bo"), json!({ "number": 7, "commentId": "cmt_1" }));
2064	let on = comment(person("usr_bo", "bo"), "@cy @bo have a look", &["cy", "bo", "g1t"]);
2065	let audience = subscribed(&[("eve", State::Subscribed, Some(Reason::Comment)), ("fay", State::Subscribed, Some(Reason::Manual))]);
2066	let notices_ = notices(&created, "acme/rocket", &actor("usr_bo", "bo"), Some(&on), &audience);
2067	assert_eq!(
2068	told(&notices_),
2069	vec![
2070	("cy", Reason::Mention, Severity::Info),
2071	("ana", Reason::Author, Severity::Info),
2072	("eve", Reason::Comment, Severity::Info),
2073	("fay", Reason::Manual, Severity::Info),
2074	]
2075	);
2076	assert_eq!(notices_[0].title, "bo mentioned you on acme/rocket#7");
2077	assert_eq!(notices_[1].title, "bo commented on acme/rocket#7");
2078	assert_eq!(notices_[0].body, "@cy @bo have a look");
2079	// Mentioned and the owner: told once, as mentioned.
2080	let on = comment(person("usr_bo", "bo"), "@ana", &["ana"]);
2081	assert_eq!(
2082	told(&notices(&created, "acme/rocket", &actor("usr_bo", "bo"), Some(&on), &nobody())),
2083	vec![("ana", Reason::Mention, Severity::Info)]
2084	);
2085	// The owner's own comment tells the assignee, not the owner.
2086	let on = comment(person("usr_ana", "ana"), "thanks", &[]);
2087	assert_eq!(
2088	told(&notices(&created, "acme/rocket", &actor("usr_ana", "ana"), Some(&on), &nobody())),
2089	vec![("bo", Reason::Assign, Severity::Info)]
2090	);
2091	// Something that happened, not something written, tells nobody.
2092	let mut on = comment(person("usr_bo", "bo"), "assigned cy", &[]);
2093	on.comment.as_mut().unwrap().event = true;
2094	assert!(notices(&created, "acme/rocket", &actor("usr_bo", "bo"), Some(&on), &nobody()).is_empty());
2095	// An approval is good news.
2096	let mut on = comment(person("usr_cy", "cy"), "", &[]);
2097	on.comment.as_mut().unwrap().verdict = Some("approve".into());
2098	let notices_ = notices(&created, "acme/rocket", &actor("usr_cy", "cy"), Some(&on), &nobody());
2099	assert_eq!(told(&notices_)[0], ("ana", Reason::Author, Severity::Success));
2100	assert_eq!(notices_[0].body, "Add the inbox");
2101	// Writing and being mentioned subscribe, never g1t.
2102	let on = comment(person("usr_bo", "bo"), "@cy", &["cy"]);
2103	assert_eq!(subscribes(&created, Some(&on)), vec![("bo".into(), Reason::Comment), ("cy".into(), Reason::Mention)]);
2104	let on = comment(person(AGENT_ID, "g1t"), "done", &[]);
2105	assert!(subscribes(&created, Some(&on)).is_empty());
2106	}
2107
2108	#[test]
2109	fn issues_and_pull_requests_are_one_thread_each() {
2110	let created = event("comment.created", Some("usr_bo"), json!({ "repoId": "rep_1", "number": 7, "commentId": "cmt_1" }));
2111	let wanted = wants(&created).unwrap();
2112	let thread = thread_of(&created, &wanted, Some(&pull()));
2113	assert_eq!(thread.key, "rep_1#7");
2114	assert_eq!(thread.kind, Some(SubjectKind::Pull));
2115	let merged = event("pull.merged", None, json!({ "repoId": "rep_1", "number": 7 }));
2116	assert_eq!(thread_of(&merged, &wants(&merged).unwrap(), Some(&pull())).key, thread.key);
2117	}
2118
2119	#[test]
2120	fn items_link_to_what_they_are_about() {
2121	assert_eq!(url(Some("acme/rocket"), Some(SubjectKind::Pull), Some(7), None, None), "/acme/rocket/pull/7");
2122	assert_eq!(url(Some("acme/rocket"), Some(SubjectKind::Issue), Some(3), None, None), "/acme/rocket/issues/3");
2123	assert_eq!(url(Some("acme/rocket"), Some(SubjectKind::Run), None, Some("run_1"), None), "/acme/rocket/actions/runs/run_1");
2124	assert_eq!(url(Some("acme/rocket"), Some(SubjectKind::Deploy), None, None, Some("/acme/site/deployments/dpl_1")), "/acme/site/deployments/dpl_1");
2125	assert_eq!(url(Some("acme/rocket"), None, None, None, None), "/acme/rocket");
2126	assert_eq!(url(None, None, None, None, None), "/inbox");
2127	}
2128
2129	#[test]
2130	fn long_titles_are_cut_to_a_line() {
2131	let long = "x".repeat(400);
2132	assert_eq!(clip(&long, MAX_TITLE).chars().count(), MAX_TITLE);
2133	assert_eq!(clip(" short ", MAX_TITLE), "short");
2134	}
2135
2136	#[test]
2137	fn marks_change_only_what_they_say() {
2138	assert_eq!(mark_change(InboxMark::Read), "read_at = COALESCE(read_at, ?1)");
2139	assert!(mark_change(InboxMark::Done).contains("done_at = ?1"));
2140	assert_eq!(mark_change(InboxMark::Unsnooze), "snoozed_until = NULL");
2141	assert!(ranked(&ListInboxArgs::default()));
2142	assert!(!ranked(&ListInboxArgs { reason: Some(Reason::Mention), ..ListInboxArgs::default() }));
2143	assert!(!ranked(&ListInboxArgs { participating: true, ..ListInboxArgs::default() }));
2144	}
2145
2146	#[test]
2147	fn filters_bind_in_order_after_what_is_bound() {
2148	let a = ListInboxArgs {
2149	reason: Some(Reason::Mention),
2150	repo_id: Some("rep_1".into()),
2151	participating: true,
2152	unread: true,
2153	..ListInboxArgs::default()
2154	};
2155	// Two values come first: the username and now.
2156	let (conditions, values) = filters(&a, 2);
2157	assert_eq!(
2158	conditions,
2159	vec!["reason = ?3", "repo_id = ?4", "reason NOT IN ('manual', 'subscribed')", "read_at IS NULL"]
2160	);
2161	assert_eq!(values, vec!["mention", "rep_1"]);
2162	}
2163
2164	#[test]
2165	fn the_bump_keeps_whatever_is_most_urgent_while_unread() {
2166	assert!(BUMP.contains("WHERE NOT EXISTS (SELECT 1 FROM inbox_activity WHERE event_id = ?4 AND username = ?2)"));
2167	assert!(BUMP.contains("ON CONFLICT (username, thread) DO UPDATE"));
2168	assert!(BUMP.contains("activity = inbox_items.activity + 1"));
2169	// Its numbered parameters run from 1 to 19.
2170	for at in 1..=19 {
2171	assert!(BUMP.contains(&format!("?{at}")), "?{at}");
2172	}
2173	assert!(!BUMP.contains("?20"));
2174	}
2175	}