forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
35.8 KiB
1//! Notification fan-out for issue and pull request activity.
2//!
3//! [`dispatch`] works out who hears about an activity, auto-subscribes the people involved and
4//! writes their inbox rows. It runs inline (database only) and never fails its caller: errors
5//! are logged.
6
7use std::collections::{BTreeMap, BTreeSet};
8
9use forge_core::mail::OutgoingMail;
10use forge_core::notifications::{Reason, SubState, WatchLevel};
11use forge_core::store::orgs::OrgMemberRole;
12use forge_core::{Action, Issue, Repo, Result, User, authorize};
13use http::HeaderMap;
14
15use crate::i18n;
16use crate::mail::{
17 self,
18 templates::{self, Verb, Why},
19};
20use crate::notify_hub::publish_unread;
21use crate::render::markdown::mentioned_usernames;
22use crate::state::AppState;
23
24/// Something that happened on an issue or pull request, as far as notifications care.
25#[derive(Debug, Clone, PartialEq, Eq)]
26pub enum Activity {
27 Opened,
28 Comment,
29 Closed,
30 Reopened,
31 Merged,
32 Review,
33 /// The user with `assignee_id` was assigned.
34 Assigned {
35 assignee_id: String,
36 },
37 /// The pull request head moved.
38 Pushed,
39 /// The body or a comment was edited. Only newly mentioned users hear about it.
40 Edited,
41}
42
43impl Activity {
44 /// The activity name stored on inbox rows.
45 pub fn as_str(&self) -> &'static str {
46 match self {
47 Activity::Opened => "opened",
48 Activity::Comment => "comment",
49 Activity::Closed => "closed",
50 Activity::Reopened => "reopened",
51 Activity::Merged => "merged",
52 Activity::Review => "review",
53 Activity::Assigned { .. } => "assigned",
54 Activity::Pushed => "pushed",
55 Activity::Edited => "edited",
56 }
57 }
58
59 /// Closing, reopening and merging make subscribers hear about a state change.
60 fn changes_state(&self) -> bool {
61 matches!(
62 self,
63 Activity::Closed | Activity::Reopened | Activity::Merged
64 )
65 }
66}
67
68/// Notifies the people who should hear about `activity` on `issue` in `repo`, for the actor
69/// `actor_id` (an empty id means a system change, so nobody is excluded). `mention_text` is
70/// searched for `@mentions`. Never fails: problems are logged.
71pub async fn dispatch(
72 state: &AppState,
73 repo: &Repo,
74 issue: &Issue,
75 actor_id: &str,
76 activity: Activity,
77 mention_text: Option<&str>,
78) {
79 let mentions = mention_text.map(mentioned_usernames).unwrap_or_default();
80 fan_out(
81 state,
82 repo,
83 issue,
84 actor_id,
85 &activity,
86 &mentions,
87 mention_text,
88 )
89 .await;
90}
91
92/// Notifies the users mentioned in `new_text` but not in `old_text`. Never fails.
93pub async fn dispatch_edit(
94 state: &AppState,
95 repo: &Repo,
96 issue: &Issue,
97 actor_id: &str,
98 old_text: &str,
99 new_text: &str,
100) {
101 let before = mentioned_usernames(old_text);
102 let added: Vec<String> = mentioned_usernames(new_text)
103 .into_iter()
104 .filter(|name| !before.contains(name))
105 .collect();
106 if added.is_empty() {
107 return;
108 }
109 fan_out(
110 state,
111 repo,
112 issue,
113 actor_id,
114 &Activity::Edited,
115 &added,
116 None,
117 )
118 .await;
119}
120
121/// Sends an `Assigned` activity for each username that is in `after` but not in `before`.
122/// Never fails.
123pub async fn dispatch_assigned(
124 state: &AppState,
125 repo: &Repo,
126 issue: &Issue,
127 actor_id: &str,
128 before: &[String],
129 after: &[String],
130) {
131 for name in after.iter().filter(|name| !before.contains(name)) {
132 match state.store.user_by_username(name).await {
133 Ok(Some(user)) => {
134 let activity = Activity::Assigned {
135 assignee_id: user.id,
136 };
137 dispatch(state, repo, issue, actor_id, activity, None).await;
138 }
139 Ok(None) => {}
140 Err(e) => {
141 tracing::error!(error = %e, repo_id = %repo.id, name, "notify: assignee lookup failed");
142 }
143 }
144 }
145}
146
147/// Tells the admins of `repo` that a lockfile scan opened `opened` security alerts on it. The
148/// admins are the owner (for an organization, its owners) and the collaborators with the Admin
149/// role. Writes inbox rows only. Never fails: problems are logged.
150pub async fn advisories_opened(state: &AppState, repo: &Repo, opened: usize) {
151 if let Err(e) = notify_advisory_admins(state, repo, opened).await {
152 tracing::error!(error = %e, repo_id = %repo.id, "notify: advisory alert notice failed");
153 }
154}
155
156/// The users who administer `repo`, without duplicates.
157async fn advisory_admins(state: &AppState, repo: &Repo) -> Result<Vec<String>> {
158 let mut admins: BTreeSet<String> = BTreeSet::new();
159 if let Some(owner) = state.store.user_by_id(&repo.owner_id).await? {
160 if owner.is_org() {
161 for (member, role) in state.store.org_members(&owner.id).await? {
162 if role == OrgMemberRole::Owner {
163 admins.insert(member.id);
164 }
165 }
166 } else {
167 admins.insert(owner.id);
168 }
169 }
170 for (user, role) in state.store.list_collaborators(&repo.id).await? {
171 if role == forge_core::Role::Admin {
172 admins.insert(user.id);
173 }
174 }
175 Ok(admins.into_iter().collect())
176}
177
178async fn notify_advisory_admins(state: &AppState, repo: &Repo, opened: usize) -> Result<()> {
179 let count = u64::try_from(opened).unwrap_or(u64::MAX);
180 for admin in advisory_admins(state, repo).await? {
181 state
182 .store
183 .notify_new_alerts(&admin, &repo.id, count)
184 .await?;
185 publish_unread(state, &admin).await;
186 }
187 Ok(())
188}
189
190async fn fan_out(
191 state: &AppState,
192 repo: &Repo,
193 issue: &Issue,
194 actor_id: &str,
195 activity: &Activity,
196 mentions: &[String],
197 body_text: Option<&str>,
198) {
199 if let Err(e) = run(state, repo, issue, actor_id, activity, mentions, body_text).await {
200 tracing::error!(
201 error = %e,
202 repo_id = %repo.id,
203 issue_id = %issue.id,
204 activity = activity.as_str(),
205 "notify: dispatch failed"
206 );
207 }
208}
209
210/// Keeps the stronger of the reasons already recorded for a user and `reason`.
211fn add(candidates: &mut BTreeMap<String, Reason>, user_id: String, reason: Reason) {
212 match candidates.get(&user_id) {
213 Some(prev) if prev.rank() >= reason.rank() => {}
214 _ => {
215 candidates.insert(user_id, reason);
216 }
217 }
218}
219
220fn strongest(a: Reason, b: Reason) -> Reason {
221 if a.rank() >= b.rank() { a } else { b }
222}
223
224/// The user when they exist and can Read the repo now, by their effective role (collaborator
225/// or organization).
226async fn readable_user(state: &AppState, repo: &Repo, user_id: &str) -> Result<Option<User>> {
227 let Some(user) = state.store.user_by_id(user_id).await? else {
228 return Ok(None);
229 };
230 if user.is_org() {
231 return Ok(None);
232 }
233 let role = state.store.collaborator_role(&repo.id, user_id).await?;
234 let org_role = state
235 .store
236 .org_repo_role(&repo.owner_id, user_id, &repo.id)
237 .await?;
238 Ok(authorize(Some(&user), repo, role, org_role, Action::Read)
239 .is_ok()
240 .then_some(user))
241}
242
243/// The people who hear about something addressed to `user_id`. A person hears about it when
244/// they can read the repo. An organization (for example the owner of a repo) hands it to its
245/// owners who can read the repo; the org itself never receives anything.
246async fn recipients_for(state: &AppState, repo: &Repo, user_id: &str) -> Result<Vec<String>> {
247 let Some(user) = state.store.user_by_id(user_id).await? else {
248 return Ok(Vec::new());
249 };
250 if !user.is_org() {
251 return Ok(readable_user(state, repo, user_id)
252 .await?
253 .map(|u| vec![u.id])
254 .unwrap_or_default());
255 }
256 let mut owners = Vec::new();
257 for (member, role) in state.store.org_members(&user.id).await? {
258 if role == OrgMemberRole::Owner && readable_user(state, repo, &member.id).await?.is_some() {
259 owners.push(member.id);
260 }
261 }
262 Ok(owners)
263}
264
265async fn run(
266 state: &AppState,
267 repo: &Repo,
268 issue: &Issue,
269 actor_id: &str,
270 activity: &Activity,
271 mentions: &[String],
272 body_text: Option<&str>,
273) -> Result<()> {
274 let store = &state.store;
275 let actor = (!actor_id.is_empty()).then_some(actor_id);
276 let issue_id = issue.id.as_str();
277 // Built once per run. Nothing is queued when instance mail is off.
278 let plan = match mail_plan(state, repo, issue, actor_id, activity, body_text).await {
279 Ok(plan) => plan,
280 Err(e) => {
281 tracing::error!(error = %e, issue_id, "notify: email plan failed");
282 None
283 }
284 };
285
286 // Candidates with the reason they would hear about this activity.
287 let mut candidates: BTreeMap<String, Reason> = BTreeMap::new();
288 let mut mentioned: BTreeSet<String> = BTreeSet::new();
289
290 if *activity != Activity::Edited {
291 for (user_id, stored) in store.subscribers(issue_id).await? {
292 let reason = if activity.changes_state() {
293 strongest(stored, Reason::StateChange)
294 } else {
295 stored
296 };
297 add(&mut candidates, user_id, reason);
298 }
299 // Pushes reach subscribers only.
300 if *activity != Activity::Pushed {
301 for user_id in store.watchers(&repo.id).await? {
302 if is_ignored(state, &user_id, issue_id).await? {
303 continue;
304 }
305 add(&mut candidates, user_id, Reason::Watching);
306 }
307 }
308 if let Activity::Assigned { assignee_id } = activity
309 && !is_ignored(state, assignee_id, issue_id).await?
310 {
311 add(&mut candidates, assignee_id.clone(), Reason::Assign);
312 }
313 }
314
315 // Mentions reach the user whatever their watch level or subscription.
316 for name in mentions {
317 if let Some(user) = store.user_by_username(name).await? {
318 mentioned.insert(user.id.clone());
319 add(&mut candidates, user.id, Reason::Mention);
320 }
321 }
322
323 let mut recipients: BTreeMap<String, Reason> = BTreeMap::new();
324 for (user_id, reason) in candidates {
325 let is_mention = mentioned.contains(&user_id);
326 for recipient in recipients_for(state, repo, &user_id).await? {
327 if Some(recipient.as_str()) == actor {
328 continue;
329 }
330 if !is_mention && store.watch_level(&recipient, &repo.id).await? == WatchLevel::Ignoring
331 {
332 continue;
333 }
334 add(&mut recipients, recipient, reason);
335 }
336 }
337
338 // Auto-subscribe. Existing rows, including explicit ignores, are left alone.
339 if let Some(actor_id) = actor {
340 match activity {
341 Activity::Opened => {
342 store
343 .auto_subscribe(actor_id, issue_id, Reason::Author)
344 .await?
345 }
346 Activity::Comment | Activity::Review => {
347 store
348 .auto_subscribe(actor_id, issue_id, Reason::Comment)
349 .await?;
350 }
351 _ => {}
352 }
353 }
354 if let Activity::Assigned { assignee_id } = activity
355 && readable_user(state, repo, assignee_id).await?.is_some()
356 {
357 store
358 .auto_subscribe(assignee_id, issue_id, Reason::Assign)
359 .await?;
360 }
361 for user_id in &mentioned {
362 if readable_user(state, repo, user_id).await?.is_some() {
363 store
364 .auto_subscribe(user_id, issue_id, Reason::Mention)
365 .await?;
366 }
367 }
368
369 for (user_id, reason) in recipients {
370 match store
371 .upsert_notification(
372 &user_id,
373 &repo.id,
374 issue_id,
375 reason,
376 activity.as_str(),
377 actor,
378 )
379 .await
380 {
381 Ok(()) => publish_unread(state, &user_id).await,
382 Err(e) => tracing::error!(error = %e, user_id, issue_id, "notify: inbox write failed"),
383 }
384 if let Some(plan) = &plan
385 && let Err(e) = queue_email(state, plan, &user_id, reason).await
386 {
387 tracing::error!(error = %e, user_id, issue_id, "notify: email copy failed");
388 }
389 }
390 Ok(())
391}
392
393/// What the email copies of one activity share. Built once per `run`; the reason, the address and
394/// the Message-ID depend on the recipient and are worked out in [`queue_email`].
395struct MailPlan {
396 /// The host of `base_url`, used as the Message-ID domain.
397 host: String,
398 /// The thread root, `<issue-{id}@host>`.
399 root: String,
400 /// One id per `run`, so each recipient's Message-ID is unique to this event.
401 event_id: String,
402 /// True for an opening: that mail is the root of the thread.
403 opens_thread: bool,
404 actor: Option<String>,
405 verb: Verb,
406 /// `owner/repo`.
407 repo: String,
408 number: i64,
409 title: String,
410 comment: Option<String>,
411 url: String,
412 settings_url: String,
413}
414
415/// The mail plan for this activity, or `None` when instance mail is off. Reads the mail settings
416/// once.
417async fn mail_plan(
418 state: &AppState,
419 repo: &Repo,
420 issue: &Issue,
421 actor_id: &str,
422 activity: &Activity,
423 body_text: Option<&str>,
424) -> Result<Option<MailPlan>> {
425 if !state.store.mail_settings().await?.enabled {
426 return Ok(None);
427 }
428 let base = state.config.base_url.trim_end_matches('/');
429 let host = host_of(base);
430 let actor = match actor_id {
431 "" => None,
432 id => state.store.user_by_id(id).await?.map(|u| u.username),
433 };
434 let section = if state
435 .store
436 .pull_by_number(&repo.id, issue.number)
437 .await?
438 .is_some()
439 {
440 "pulls"
441 } else {
442 "issues"
443 };
444 Ok(Some(MailPlan {
445 root: format!("<issue-{}@{host}>", issue.id),
446 host,
447 event_id: ulid::Ulid::generate().to_string(),
448 opens_thread: *activity == Activity::Opened,
449 actor,
450 verb: verb_for(activity),
451 repo: format!("{}/{}", repo.owner_name, repo.name),
452 number: issue.number,
453 title: issue.title.clone(),
454 comment: body_text.map(str::to_owned),
455 url: format!(
456 "{base}/{}/{}/{section}/{}",
457 repo.owner_name, repo.name, issue.number
458 ),
459 settings_url: format!("{base}/settings/notifications"),
460 }))
461}
462
463fn verb_for(activity: &Activity) -> Verb {
464 match activity {
465 Activity::Opened => Verb::Opened,
466 Activity::Comment => Verb::Commented,
467 Activity::Closed => Verb::Closed,
468 Activity::Reopened => Verb::Reopened,
469 Activity::Merged => Verb::Merged,
470 Activity::Review => Verb::Reviewed,
471 Activity::Assigned { .. } => Verb::AssignedYou,
472 Activity::Pushed => Verb::Pushed,
473 Activity::Edited => Verb::MentionedYou,
474 }
475}
476
477fn why_for(reason: Reason) -> Why {
478 match reason {
479 Reason::Mention => Why::Mention,
480 Reason::Assign => Why::Assign,
481 Reason::ReviewRequested => Why::ReviewRequested,
482 Reason::Author => Why::Author,
483 Reason::Comment => Why::Comment,
484 Reason::StateChange | Reason::Subscribed => Why::Subscribed,
485 Reason::Watching => Why::Watching,
486 }
487}
488
489/// Queues the email copy of one notification for `user_id`. Nothing is queued unless the user has
490/// a verified address and email notifications on. The idempotency key is per event and user, so
491/// a repeated run sends nothing twice.
492async fn queue_email(
493 state: &AppState,
494 plan: &MailPlan,
495 user_id: &str,
496 reason: Reason,
497) -> Result<()> {
498 let Some(user) = state.store.user_by_id(user_id).await? else {
499 return Ok(());
500 };
501 let Some(address) = user.email.as_deref().filter(|address| !address.is_empty()) else {
502 return Ok(());
503 };
504 if !user.email_notifications || !state.store.email_is_verified(user_id).await? {
505 return Ok(());
506 }
507 let lang = i18n::resolve(Some(user.language.as_str()), &HeaderMap::new());
508 let (subject, body) = templates::notification(
509 lang,
510 &templates::Notification {
511 actor: plan.actor.as_deref(),
512 verb: plan.verb,
513 repo: &plan.repo,
514 number: plan.number,
515 title: &plan.title,
516 comment: plan.comment.as_deref(),
517 url: &plan.url,
518 why: why_for(reason),
519 settings_url: &plan.settings_url,
520 },
521 );
522 let (message_id, in_reply_to) = if plan.opens_thread {
523 (plan.root.clone(), None)
524 } else {
525 (
526 format!("<{}.{user_id}@{}>", plan.event_id, plan.host),
527 Some(plan.root.clone()),
528 )
529 };
530 let outgoing = OutgoingMail {
531 to: address.to_owned(),
532 subject,
533 body,
534 message_id: Some(message_id),
535 in_reply_to,
536 };
537 let idem_key = format!("notif:{}:{user_id}", plan.event_id);
538 mail::enqueue_message(state, &idem_key, &outgoing).await?;
539 Ok(())
540}
541
542/// The host of `base_url`, without scheme, port or path. Used as the Message-ID domain.
543fn host_of(base_url: &str) -> String {
544 let rest = base_url
545 .split_once("://")
546 .map_or(base_url, |(_, rest)| rest);
547 let authority = rest.split(['/', '?', '#']).next().unwrap_or_default();
548 let host = authority
549 .rsplit_once('@')
550 .map_or(authority, |(_, host)| host);
551 let host = host.split(':').next().unwrap_or_default();
552 if host.is_empty() {
553 "localhost".to_owned()
554 } else {
555 host.to_owned()
556 }
557}
558
559/// True when the user has an explicit `ignored` subscription on the issue.
560async fn is_ignored(state: &AppState, user_id: &str, issue_id: &str) -> Result<bool> {
561 Ok(matches!(
562 state.store.subscription(user_id, issue_id).await?,
563 Some((SubState::Ignored, _))
564 ))
565}
566
567#[cfg(test)]
568mod tests {
569 use forge_core::{Notification, Role, Visibility};
570
571 use super::*;
572 use crate::test_support::test_state;
573
574 struct Fixture {
575 state: AppState,
576 repo: Repo,
577 issue: Issue,
578 owner: User,
579 alice: User,
580 bob: User,
581 carol: User,
582 _dir: tempfile::TempDir,
583 }
584
585 async fn fixture(visibility: Visibility) -> Fixture {
586 let dir = tempfile::tempdir().expect("tempdir");
587 let state = test_state(dir.path()).await;
588 let store = &state.store;
589 let owner = store
590 .create_user("owner", None, "unused", false)
591 .await
592 .unwrap();
593 let alice = store
594 .create_user("alice", None, "unused", false)
595 .await
596 .unwrap();
597 let bob = store
598 .create_user("bob", None, "unused", false)
599 .await
600 .unwrap();
601 let carol = store
602 .create_user("carol", None, "unused", false)
603 .await
604 .unwrap();
605 let repo = store
606 .create_repo(&owner.id, "proj", "", visibility, "main")
607 .await
608 .unwrap();
609 let issue = store
610 .create_issue(&repo.id, &owner.id, "a problem", "body")
611 .await
612 .unwrap();
613 Fixture {
614 state,
615 repo,
616 issue,
617 owner,
618 alice,
619 bob,
620 carol,
621 _dir: dir,
622 }
623 }
624
625 async fn inbox(fx: &Fixture, user: &User) -> Vec<Notification> {
626 fx.state
627 .store
628 .list_notifications(&user.id, false, None, 50)
629 .await
630 .unwrap()
631 }
632
633 #[test]
634 fn mentions_skip_code_links_and_emails() {
635 let text = "hi @Alice and @bob. mail a@b.com `@code`\n\n\
636 ```\n@blocked\n```\n[@linked](https://x.test) @alice again";
637 assert_eq!(mentioned_usernames(text), vec!["alice", "bob"]);
638 }
639
640 #[test]
641 fn mentions_in_html_blocks_are_found() {
642 let text = "<div>\n@carol\n</div>\n";
643 assert_eq!(mentioned_usernames(text), vec!["carol"]);
644 }
645
646 #[test]
647 fn mentions_are_deduped_case_insensitively() {
648 assert_eq!(
649 mentioned_usernames("@Bob @bob @BOB"),
650 vec!["bob"],
651 "one name however it is cased"
652 );
653 assert!(mentioned_usernames("no mentions here").is_empty());
654 }
655
656 #[tokio::test]
657 async fn comment_notifies_subscriber_not_the_commenter() {
658 let fx = fixture(Visibility::Public).await;
659 let store = &fx.state.store;
660 store
661 .subscribe(
662 &fx.bob.id,
663 &fx.issue.id,
664 SubState::Subscribed,
665 Reason::Subscribed,
666 )
667 .await
668 .unwrap();
669 store
670 .subscribe(
671 &fx.alice.id,
672 &fx.issue.id,
673 SubState::Subscribed,
674 Reason::Subscribed,
675 )
676 .await
677 .unwrap();
678
679 dispatch(
680 &fx.state,
681 &fx.repo,
682 &fx.issue,
683 &fx.alice.id,
684 Activity::Comment,
685 Some("hello"),
686 )
687 .await;
688
689 let bob_inbox = inbox(&fx, &fx.bob).await;
690 assert_eq!(bob_inbox.len(), 1);
691 assert_eq!(bob_inbox[0].reason, Reason::Subscribed);
692 assert_eq!(bob_inbox[0].last_activity, "comment");
693 assert!(
694 inbox(&fx, &fx.alice).await.is_empty(),
695 "the actor is never notified"
696 );
697 }
698
699 #[tokio::test]
700 async fn opened_notifies_watchers_and_auto_subscribes_the_author() {
701 let fx = fixture(Visibility::Public).await;
702 fx.state
703 .store
704 .set_watch(&fx.bob.id, &fx.repo.id, WatchLevel::Watching)
705 .await
706 .unwrap();
707
708 dispatch(
709 &fx.state,
710 &fx.repo,
711 &fx.issue,
712 &fx.owner.id,
713 Activity::Opened,
714 None,
715 )
716 .await;
717
718 let bob_inbox = inbox(&fx, &fx.bob).await;
719 assert_eq!(bob_inbox.len(), 1);
720 assert_eq!(bob_inbox[0].reason, Reason::Watching);
721 assert!(inbox(&fx, &fx.owner).await.is_empty());
722 let sub = fx
723 .state
724 .store
725 .subscription(&fx.owner.id, &fx.issue.id)
726 .await
727 .unwrap();
728 assert_eq!(sub, Some((SubState::Subscribed, Reason::Author)));
729 }
730
731 #[tokio::test]
732 async fn ignored_subscription_suppresses_watcher() {
733 let fx = fixture(Visibility::Public).await;
734 let store = &fx.state.store;
735 store
736 .set_watch(&fx.bob.id, &fx.repo.id, WatchLevel::Watching)
737 .await
738 .unwrap();
739 store
740 .subscribe(
741 &fx.bob.id,
742 &fx.issue.id,
743 SubState::Ignored,
744 Reason::Subscribed,
745 )
746 .await
747 .unwrap();
748
749 dispatch(
750 &fx.state,
751 &fx.repo,
752 &fx.issue,
753 &fx.owner.id,
754 Activity::Opened,
755 None,
756 )
757 .await;
758
759 assert!(inbox(&fx, &fx.bob).await.is_empty());
760 }
761
762 #[tokio::test]
763 async fn ignoring_repo_suppresses_but_mention_still_notifies() {
764 let fx = fixture(Visibility::Public).await;
765 let store = &fx.state.store;
766 store
767 .set_watch(&fx.alice.id, &fx.repo.id, WatchLevel::Ignoring)
768 .await
769 .unwrap();
770 store
771 .subscribe(
772 &fx.alice.id,
773 &fx.issue.id,
774 SubState::Subscribed,
775 Reason::Subscribed,
776 )
777 .await
778 .unwrap();
779
780 dispatch(
781 &fx.state,
782 &fx.repo,
783 &fx.issue,
784 &fx.bob.id,
785 Activity::Comment,
786 Some("plain"),
787 )
788 .await;
789 assert!(inbox(&fx, &fx.alice).await.is_empty());
790
791 dispatch(
792 &fx.state,
793 &fx.repo,
794 &fx.issue,
795 &fx.bob.id,
796 Activity::Comment,
797 Some("@alice look"),
798 )
799 .await;
800 let alice_inbox = inbox(&fx, &fx.alice).await;
801 assert_eq!(alice_inbox.len(), 1);
802 assert_eq!(alice_inbox[0].reason, Reason::Mention);
803 }
804
805 #[tokio::test]
806 async fn mention_without_read_access_on_private_repo_is_not_notified() {
807 let fx = fixture(Visibility::Private).await;
808 let store = &fx.state.store;
809
810 dispatch(
811 &fx.state,
812 &fx.repo,
813 &fx.issue,
814 &fx.owner.id,
815 Activity::Comment,
816 Some("@bob hi"),
817 )
818 .await;
819 assert!(inbox(&fx, &fx.bob).await.is_empty());
820 let sub = store.subscription(&fx.bob.id, &fx.issue.id).await.unwrap();
821 assert_eq!(sub, None, "an unreadable mention does not subscribe");
822
823 store
824 .set_collaborator(&fx.repo.id, &fx.bob.id, Role::Read)
825 .await
826 .unwrap();
827 dispatch(
828 &fx.state,
829 &fx.repo,
830 &fx.issue,
831 &fx.owner.id,
832 Activity::Comment,
833 Some("@bob hi"),
834 )
835 .await;
836 let bob_inbox = inbox(&fx, &fx.bob).await;
837 assert_eq!(bob_inbox.len(), 1);
838 assert_eq!(bob_inbox[0].reason, Reason::Mention);
839 }
840
841 #[tokio::test]
842 async fn assignee_is_notified_with_assign_reason() {
843 let fx = fixture(Visibility::Public).await;
844
845 dispatch(
846 &fx.state,
847 &fx.repo,
848 &fx.issue,
849 &fx.owner.id,
850 Activity::Assigned {
851 assignee_id: fx.bob.id.clone(),
852 },
853 None,
854 )
855 .await;
856
857 let bob_inbox = inbox(&fx, &fx.bob).await;
858 assert_eq!(bob_inbox.len(), 1);
859 assert_eq!(bob_inbox[0].reason, Reason::Assign);
860 assert_eq!(bob_inbox[0].last_activity, "assigned");
861 }
862
863 #[tokio::test]
864 async fn pushed_reaches_subscribers_not_watchers() {
865 let fx = fixture(Visibility::Public).await;
866 let store = &fx.state.store;
867 store
868 .set_watch(&fx.bob.id, &fx.repo.id, WatchLevel::Watching)
869 .await
870 .unwrap();
871 store
872 .subscribe(
873 &fx.carol.id,
874 &fx.issue.id,
875 SubState::Subscribed,
876 Reason::Author,
877 )
878 .await
879 .unwrap();
880
881 dispatch(
882 &fx.state,
883 &fx.repo,
884 &fx.issue,
885 &fx.owner.id,
886 Activity::Pushed,
887 None,
888 )
889 .await;
890
891 assert!(inbox(&fx, &fx.bob).await.is_empty());
892 let carol_inbox = inbox(&fx, &fx.carol).await;
893 assert_eq!(carol_inbox.len(), 1);
894 assert_eq!(carol_inbox[0].reason, Reason::Author);
895 }
896
897 #[tokio::test]
898 async fn edit_notifies_only_the_newly_mentioned_user() {
899 let fx = fixture(Visibility::Public).await;
900 fx.state
901 .store
902 .set_watch(&fx.bob.id, &fx.repo.id, WatchLevel::Watching)
903 .await
904 .unwrap();
905
906 dispatch_edit(
907 &fx.state,
908 &fx.repo,
909 &fx.issue,
910 &fx.owner.id,
911 "hi @bob",
912 "hi @bob and @carol",
913 )
914 .await;
915
916 assert!(
917 inbox(&fx, &fx.bob).await.is_empty(),
918 "bob was mentioned before the edit"
919 );
920 let carol_inbox = inbox(&fx, &fx.carol).await;
921 assert_eq!(carol_inbox.len(), 1);
922 assert_eq!(carol_inbox[0].reason, Reason::Mention);
923 assert_eq!(carol_inbox[0].last_activity, "edited");
924 }
925
926 #[tokio::test]
927 async fn advisory_alerts_reach_owner_admin_collaborators_and_org_owners_only() {
928 let dir = tempfile::tempdir().expect("tempdir");
929 let state = test_state(dir.path()).await;
930 let store = &state.store;
931 let owner = store
932 .create_user("owner", None, "unused", false)
933 .await
934 .unwrap();
935 let admin = store
936 .create_user("admin", None, "unused", false)
937 .await
938 .unwrap();
939 let writer = store
940 .create_user("writer", None, "unused", false)
941 .await
942 .unwrap();
943 let repo = store
944 .create_repo(&owner.id, "proj", "", Visibility::Private, "main")
945 .await
946 .unwrap();
947 store
948 .set_collaborator(&repo.id, &admin.id, Role::Admin)
949 .await
950 .unwrap();
951 store
952 .set_collaborator(&repo.id, &writer.id, Role::Write)
953 .await
954 .unwrap();
955
956 advisories_opened(&state, &repo, 2).await;
957
958 assert_eq!(advisory_rows(store, &owner.id).await, vec![2]);
959 assert_eq!(advisory_rows(store, &admin.id).await, vec![2]);
960 assert!(advisory_rows(store, &writer.id).await.is_empty());
961
962 // An org-owned repo: its owners and admin collaborators hear, its plain members do not.
963 let org = store.create_org("acme", &owner.id).await.unwrap();
964 let org_member = store
965 .create_user("member", None, "unused", false)
966 .await
967 .unwrap();
968 let org_owner = store
969 .create_user("orgowner", None, "unused", false)
970 .await
971 .unwrap();
972 store
973 .add_org_member(&org.id, &org_member.id, OrgMemberRole::Member)
974 .await
975 .unwrap();
976 store
977 .add_org_member(&org.id, &org_owner.id, OrgMemberRole::Owner)
978 .await
979 .unwrap();
980 let org_repo = store
981 .create_repo(&org.id, "shared", "", Visibility::Private, "main")
982 .await
983 .unwrap();
984
985 advisories_opened(&state, &org_repo, 1).await;
986
987 assert_eq!(advisory_rows(store, &owner.id).await, vec![1, 2]);
988 assert_eq!(advisory_rows(store, &org_owner.id).await, vec![1]);
989 assert!(advisory_rows(store, &org_member.id).await.is_empty());
990 assert!(advisory_rows(store, &org.id).await.is_empty());
991 }
992
993 /// The `new_alerts` counts of a user's advisory inbox rows, sorted.
994 async fn advisory_rows(store: &forge_core::store::Store, user_id: &str) -> Vec<u64> {
995 let mut counts: Vec<u64> = store
996 .list_advisory_notifications(user_id, false, None, 50)
997 .await
998 .unwrap()
999 .iter()
1000 .map(|notice| notice.new_alerts)
1001 .collect();
1002 counts.sort_unstable();
1003 counts
1004 }
1005
1006 /// Instance mail on or off. Bob has a verified address and is subscribed to the issue.
1007 async fn mail_fixture(enabled: bool) -> Fixture {
1008 let fx = fixture(Visibility::Public).await;
1009 let store = &fx.state.store;
1010 store
1011 .update_mail_settings(&forge_core::mail::MailSettingsUpdate {
1012 host: "smtp.test.example",
1013 port: 587,
1014 security: forge_core::mail::Security::StartTls,
1015 username: "",
1016 from_addr: "forge@test.example",
1017 from_name: "forge",
1018 enabled,
1019 password: None,
1020 })
1021 .await
1022 .unwrap();
1023 store
1024 .set_email(&fx.bob.id, "bob@test.example")
1025 .await
1026 .unwrap();
1027 let token = store
1028 .create_verify_token(&fx.bob.id, "bob@test.example")
1029 .await
1030 .unwrap();
1031 store.redeem_verify_token(&token).await.unwrap();
1032 store
1033 .subscribe(
1034 &fx.bob.id,
1035 &fx.issue.id,
1036 SubState::Subscribed,
1037 Reason::Subscribed,
1038 )
1039 .await
1040 .unwrap();
1041 fx
1042 }
1043
1044 /// The outbox rows addressed to `address`.
1045 async fn mail_to(fx: &Fixture, address: &str) -> Vec<forge_core::mail::OutboxEntry> {
1046 fx.state
1047 .store
1048 .recent_mail(50)
1049 .await
1050 .unwrap()
1051 .into_iter()
1052 .filter(|row| row.to_addr == address)
1053 .collect()
1054 }
1055
1056 #[tokio::test]
1057 async fn comment_emails_the_subscriber_threaded_under_the_issue() {
1058 let fx = mail_fixture(true).await;
1059 dispatch(
1060 &fx.state,
1061 &fx.repo,
1062 &fx.issue,
1063 &fx.alice.id,
1064 Activity::Comment,
1065 Some("hello there"),
1066 )
1067 .await;
1068
1069 let rows = mail_to(&fx, "bob@test.example").await;
1070 assert_eq!(rows.len(), 1, "one mail to the subscriber");
1071 let host = host_of(&fx.state.config.base_url);
1072 assert_eq!(
1073 rows[0].in_reply_to.as_deref(),
1074 Some(format!("<issue-{}@{host}>", fx.issue.id).as_str())
1075 );
1076 let message_id = rows[0].message_id.as_deref().unwrap_or_default();
1077 assert!(message_id.starts_with('<') && message_id.ends_with(&format!("@{host}>")));
1078 assert_ne!(
1079 message_id,
1080 rows[0].in_reply_to.as_deref().unwrap_or_default()
1081 );
1082 assert!(rows[0].body.contains("hello there"), "{}", rows[0].body);
1083 assert!(
1084 rows[0].body.contains("alice commented on this."),
1085 "{}",
1086 rows[0].body
1087 );
1088 assert!(
1089 rows[0].idem_key.starts_with("notif:"),
1090 "{}",
1091 rows[0].idem_key
1092 );
1093 assert!(rows[0].idem_key.ends_with(&format!(":{}", fx.bob.id)));
1094 assert!(
1095 inbox(&fx, &fx.alice).await.is_empty(),
1096 "the actor gets no mail or inbox row"
1097 );
1098 }
1099
1100 #[tokio::test]
1101 async fn opening_mail_is_the_thread_root() {
1102 let fx = mail_fixture(true).await;
1103 dispatch(
1104 &fx.state,
1105 &fx.repo,
1106 &fx.issue,
1107 &fx.alice.id,
1108 Activity::Opened,
1109 Some("the body"),
1110 )
1111 .await;
1112
1113 let rows = mail_to(&fx, "bob@test.example").await;
1114 assert_eq!(rows.len(), 1);
1115 let host = host_of(&fx.state.config.base_url);
1116 let root = format!("<issue-{}@{host}>", fx.issue.id);
1117 assert_eq!(rows[0].message_id.as_deref(), Some(root.as_str()));
1118 assert_eq!(rows[0].in_reply_to, None);
1119 }
1120
1121 #[tokio::test]
1122 async fn no_mail_without_a_verified_address() {
1123 let fx = mail_fixture(true).await;
1124 fx.state
1125 .store
1126 .clear_email_verified(&fx.bob.id)
1127 .await
1128 .unwrap();
1129 dispatch(
1130 &fx.state,
1131 &fx.repo,
1132 &fx.issue,
1133 &fx.alice.id,
1134 Activity::Comment,
1135 Some("hi"),
1136 )
1137 .await;
1138 assert!(mail_to(&fx, "bob@test.example").await.is_empty());
1139 assert_eq!(
1140 inbox(&fx, &fx.bob).await.len(),
1141 1,
1142 "the inbox still gets it"
1143 );
1144 }
1145
1146 #[tokio::test]
1147 async fn no_mail_when_the_user_turned_email_notifications_off() {
1148 let fx = mail_fixture(true).await;
1149 fx.state
1150 .store
1151 .set_email_notifications(&fx.bob.id, false)
1152 .await
1153 .unwrap();
1154 dispatch(
1155 &fx.state,
1156 &fx.repo,
1157 &fx.issue,
1158 &fx.alice.id,
1159 Activity::Comment,
1160 Some("hi"),
1161 )
1162 .await;
1163 assert!(mail_to(&fx, "bob@test.example").await.is_empty());
1164 assert_eq!(inbox(&fx, &fx.bob).await.len(), 1);
1165 }
1166
1167 #[tokio::test]
1168 async fn no_mail_when_instance_mail_is_disabled() {
1169 let fx = mail_fixture(false).await;
1170 dispatch(
1171 &fx.state,
1172 &fx.repo,
1173 &fx.issue,
1174 &fx.alice.id,
1175 Activity::Comment,
1176 Some("hi"),
1177 )
1178 .await;
1179 assert!(mail_to(&fx, "bob@test.example").await.is_empty());
1180 }
1181
1182 #[test]
1183 fn host_of_drops_scheme_port_and_path() {
1184 assert_eq!(
1185 host_of("https://forge.example.com:8443/"),
1186 "forge.example.com"
1187 );
1188 assert_eq!(host_of("http://127.0.0.1:7660"), "127.0.0.1");
1189 assert_eq!(host_of("forge.test/x"), "forge.test");
1190 assert_eq!(host_of("https:///"), "localhost");
1191 }
1192}