forge
ExploreLog in

Tahinli/forge

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