//! Notification fan-out for issue and pull request activity.
//!
//! [`dispatch`] works out who hears about an activity, auto-subscribes the people involved and
//! writes their inbox rows. It runs inline (database only) and never fails its caller: errors
//! are logged.

use std::collections::{BTreeMap, BTreeSet};

use forge_core::mail::OutgoingMail;
use forge_core::notifications::{Reason, SubState, WatchLevel};
use forge_core::store::orgs::OrgMemberRole;
use forge_core::{Action, Issue, Repo, Result, User, authorize};
use http::HeaderMap;

use crate::i18n;
use crate::mail::{
    self,
    templates::{self, Verb, Why},
};
use crate::notify_hub::publish_unread;
use crate::render::markdown::mentioned_usernames;
use crate::state::AppState;

/// Something that happened on an issue or pull request, as far as notifications care.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Activity {
    Opened,
    Comment,
    Closed,
    Reopened,
    Merged,
    Review,
    /// The user with `assignee_id` was assigned.
    Assigned {
        assignee_id: String,
    },
    /// The pull request head moved.
    Pushed,
    /// The body or a comment was edited. Only newly mentioned users hear about it.
    Edited,
}

impl Activity {
    /// The activity name stored on inbox rows.
    pub fn as_str(&self) -> &'static str {
        match self {
            Activity::Opened => "opened",
            Activity::Comment => "comment",
            Activity::Closed => "closed",
            Activity::Reopened => "reopened",
            Activity::Merged => "merged",
            Activity::Review => "review",
            Activity::Assigned { .. } => "assigned",
            Activity::Pushed => "pushed",
            Activity::Edited => "edited",
        }
    }

    /// Closing, reopening and merging make subscribers hear about a state change.
    fn changes_state(&self) -> bool {
        matches!(
            self,
            Activity::Closed | Activity::Reopened | Activity::Merged
        )
    }
}

/// Notifies the people who should hear about `activity` on `issue` in `repo`, for the actor
/// `actor_id` (an empty id means a system change, so nobody is excluded). `mention_text` is
/// searched for `@mentions`. Never fails: problems are logged.
pub async fn dispatch(
    state: &AppState,
    repo: &Repo,
    issue: &Issue,
    actor_id: &str,
    activity: Activity,
    mention_text: Option<&str>,
) {
    let mentions = mention_text.map(mentioned_usernames).unwrap_or_default();
    fan_out(
        state,
        repo,
        issue,
        actor_id,
        &activity,
        &mentions,
        mention_text,
    )
    .await;
}

/// Notifies the users mentioned in `new_text` but not in `old_text`. Never fails.
pub async fn dispatch_edit(
    state: &AppState,
    repo: &Repo,
    issue: &Issue,
    actor_id: &str,
    old_text: &str,
    new_text: &str,
) {
    let before = mentioned_usernames(old_text);
    let added: Vec<String> = mentioned_usernames(new_text)
        .into_iter()
        .filter(|name| !before.contains(name))
        .collect();
    if added.is_empty() {
        return;
    }
    fan_out(
        state,
        repo,
        issue,
        actor_id,
        &Activity::Edited,
        &added,
        None,
    )
    .await;
}

/// Sends an `Assigned` activity for each username that is in `after` but not in `before`.
/// Never fails.
pub async fn dispatch_assigned(
    state: &AppState,
    repo: &Repo,
    issue: &Issue,
    actor_id: &str,
    before: &[String],
    after: &[String],
) {
    for name in after.iter().filter(|name| !before.contains(name)) {
        match state.store.user_by_username(name).await {
            Ok(Some(user)) => {
                let activity = Activity::Assigned {
                    assignee_id: user.id,
                };
                dispatch(state, repo, issue, actor_id, activity, None).await;
            }
            Ok(None) => {}
            Err(e) => {
                tracing::error!(error = %e, repo_id = %repo.id, name, "notify: assignee lookup failed");
            }
        }
    }
}

/// Tells the admins of `repo` that a lockfile scan opened `opened` security alerts on it. The
/// admins are the owner (for an organization, its owners) and the collaborators with the Admin
/// role. Writes inbox rows only. Never fails: problems are logged.
pub async fn advisories_opened(state: &AppState, repo: &Repo, opened: usize) {
    if let Err(e) = notify_advisory_admins(state, repo, opened).await {
        tracing::error!(error = %e, repo_id = %repo.id, "notify: advisory alert notice failed");
    }
}

/// The users who administer `repo`, without duplicates.
async fn advisory_admins(state: &AppState, repo: &Repo) -> Result<Vec<String>> {
    let mut admins: BTreeSet<String> = BTreeSet::new();
    if let Some(owner) = state.store.user_by_id(&repo.owner_id).await? {
        if owner.is_org() {
            for (member, role) in state.store.org_members(&owner.id).await? {
                if role == OrgMemberRole::Owner {
                    admins.insert(member.id);
                }
            }
        } else {
            admins.insert(owner.id);
        }
    }
    for (user, role) in state.store.list_collaborators(&repo.id).await? {
        if role == forge_core::Role::Admin {
            admins.insert(user.id);
        }
    }
    Ok(admins.into_iter().collect())
}

async fn notify_advisory_admins(state: &AppState, repo: &Repo, opened: usize) -> Result<()> {
    let count = u64::try_from(opened).unwrap_or(u64::MAX);
    for admin in advisory_admins(state, repo).await? {
        state
            .store
            .notify_new_alerts(&admin, &repo.id, count)
            .await?;
        publish_unread(state, &admin).await;
    }
    Ok(())
}

async fn fan_out(
    state: &AppState,
    repo: &Repo,
    issue: &Issue,
    actor_id: &str,
    activity: &Activity,
    mentions: &[String],
    body_text: Option<&str>,
) {
    if let Err(e) = run(state, repo, issue, actor_id, activity, mentions, body_text).await {
        tracing::error!(
            error = %e,
            repo_id = %repo.id,
            issue_id = %issue.id,
            activity = activity.as_str(),
            "notify: dispatch failed"
        );
    }
}

/// Keeps the stronger of the reasons already recorded for a user and `reason`.
fn add(candidates: &mut BTreeMap<String, Reason>, user_id: String, reason: Reason) {
    match candidates.get(&user_id) {
        Some(prev) if prev.rank() >= reason.rank() => {}
        _ => {
            candidates.insert(user_id, reason);
        }
    }
}

fn strongest(a: Reason, b: Reason) -> Reason {
    if a.rank() >= b.rank() { a } else { b }
}

/// The user when they exist and can Read the repo now, by their effective role (collaborator
/// or organization).
async fn readable_user(state: &AppState, repo: &Repo, user_id: &str) -> Result<Option<User>> {
    let Some(user) = state.store.user_by_id(user_id).await? else {
        return Ok(None);
    };
    if user.is_org() {
        return Ok(None);
    }
    let role = state.store.collaborator_role(&repo.id, user_id).await?;
    let org_role = state
        .store
        .org_repo_role(&repo.owner_id, user_id, &repo.id)
        .await?;
    Ok(authorize(Some(&user), repo, role, org_role, Action::Read)
        .is_ok()
        .then_some(user))
}

/// The people who hear about something addressed to `user_id`. A person hears about it when
/// they can read the repo. An organization (for example the owner of a repo) hands it to its
/// owners who can read the repo; the org itself never receives anything.
async fn recipients_for(state: &AppState, repo: &Repo, user_id: &str) -> Result<Vec<String>> {
    let Some(user) = state.store.user_by_id(user_id).await? else {
        return Ok(Vec::new());
    };
    if !user.is_org() {
        return Ok(readable_user(state, repo, user_id)
            .await?
            .map(|u| vec![u.id])
            .unwrap_or_default());
    }
    let mut owners = Vec::new();
    for (member, role) in state.store.org_members(&user.id).await? {
        if role == OrgMemberRole::Owner && readable_user(state, repo, &member.id).await?.is_some() {
            owners.push(member.id);
        }
    }
    Ok(owners)
}

async fn run(
    state: &AppState,
    repo: &Repo,
    issue: &Issue,
    actor_id: &str,
    activity: &Activity,
    mentions: &[String],
    body_text: Option<&str>,
) -> Result<()> {
    let store = &state.store;
    let actor = (!actor_id.is_empty()).then_some(actor_id);
    let issue_id = issue.id.as_str();
    // Built once per run. Nothing is queued when instance mail is off.
    let plan = match mail_plan(state, repo, issue, actor_id, activity, body_text).await {
        Ok(plan) => plan,
        Err(e) => {
            tracing::error!(error = %e, issue_id, "notify: email plan failed");
            None
        }
    };

    // Candidates with the reason they would hear about this activity.
    let mut candidates: BTreeMap<String, Reason> = BTreeMap::new();
    let mut mentioned: BTreeSet<String> = BTreeSet::new();

    if *activity != Activity::Edited {
        for (user_id, stored) in store.subscribers(issue_id).await? {
            let reason = if activity.changes_state() {
                strongest(stored, Reason::StateChange)
            } else {
                stored
            };
            add(&mut candidates, user_id, reason);
        }
        // Pushes reach subscribers only.
        if *activity != Activity::Pushed {
            for user_id in store.watchers(&repo.id).await? {
                if is_ignored(state, &user_id, issue_id).await? {
                    continue;
                }
                add(&mut candidates, user_id, Reason::Watching);
            }
        }
        if let Activity::Assigned { assignee_id } = activity
            && !is_ignored(state, assignee_id, issue_id).await?
        {
            add(&mut candidates, assignee_id.clone(), Reason::Assign);
        }
    }

    // Mentions reach the user whatever their watch level or subscription.
    for name in mentions {
        if let Some(user) = store.user_by_username(name).await? {
            mentioned.insert(user.id.clone());
            add(&mut candidates, user.id, Reason::Mention);
        }
    }

    let mut recipients: BTreeMap<String, Reason> = BTreeMap::new();
    for (user_id, reason) in candidates {
        let is_mention = mentioned.contains(&user_id);
        for recipient in recipients_for(state, repo, &user_id).await? {
            if Some(recipient.as_str()) == actor {
                continue;
            }
            if !is_mention && store.watch_level(&recipient, &repo.id).await? == WatchLevel::Ignoring
            {
                continue;
            }
            add(&mut recipients, recipient, reason);
        }
    }

    // Auto-subscribe. Existing rows, including explicit ignores, are left alone.
    if let Some(actor_id) = actor {
        match activity {
            Activity::Opened => {
                store
                    .auto_subscribe(actor_id, issue_id, Reason::Author)
                    .await?
            }
            Activity::Comment | Activity::Review => {
                store
                    .auto_subscribe(actor_id, issue_id, Reason::Comment)
                    .await?;
            }
            _ => {}
        }
    }
    if let Activity::Assigned { assignee_id } = activity
        && readable_user(state, repo, assignee_id).await?.is_some()
    {
        store
            .auto_subscribe(assignee_id, issue_id, Reason::Assign)
            .await?;
    }
    for user_id in &mentioned {
        if readable_user(state, repo, user_id).await?.is_some() {
            store
                .auto_subscribe(user_id, issue_id, Reason::Mention)
                .await?;
        }
    }

    for (user_id, reason) in recipients {
        match store
            .upsert_notification(
                &user_id,
                &repo.id,
                issue_id,
                reason,
                activity.as_str(),
                actor,
            )
            .await
        {
            Ok(()) => publish_unread(state, &user_id).await,
            Err(e) => tracing::error!(error = %e, user_id, issue_id, "notify: inbox write failed"),
        }
        if let Some(plan) = &plan
            && let Err(e) = queue_email(state, plan, &user_id, reason).await
        {
            tracing::error!(error = %e, user_id, issue_id, "notify: email copy failed");
        }
    }
    Ok(())
}

/// What the email copies of one activity share. Built once per `run`; the reason, the address and
/// the Message-ID depend on the recipient and are worked out in [`queue_email`].
struct MailPlan {
    /// The host of `base_url`, used as the Message-ID domain.
    host: String,
    /// The thread root, `<issue-{id}@host>`.
    root: String,
    /// One id per `run`, so each recipient's Message-ID is unique to this event.
    event_id: String,
    /// True for an opening: that mail is the root of the thread.
    opens_thread: bool,
    actor: Option<String>,
    verb: Verb,
    /// `owner/repo`.
    repo: String,
    number: i64,
    title: String,
    comment: Option<String>,
    url: String,
    settings_url: String,
}

/// The mail plan for this activity, or `None` when instance mail is off. Reads the mail settings
/// once.
async fn mail_plan(
    state: &AppState,
    repo: &Repo,
    issue: &Issue,
    actor_id: &str,
    activity: &Activity,
    body_text: Option<&str>,
) -> Result<Option<MailPlan>> {
    if !state.store.mail_settings().await?.enabled {
        return Ok(None);
    }
    let base = state.config.base_url.trim_end_matches('/');
    let host = host_of(base);
    let actor = match actor_id {
        "" => None,
        id => state.store.user_by_id(id).await?.map(|u| u.username),
    };
    let section = if state
        .store
        .pull_by_number(&repo.id, issue.number)
        .await?
        .is_some()
    {
        "pulls"
    } else {
        "issues"
    };
    Ok(Some(MailPlan {
        root: format!("<issue-{}@{host}>", issue.id),
        host,
        event_id: ulid::Ulid::generate().to_string(),
        opens_thread: *activity == Activity::Opened,
        actor,
        verb: verb_for(activity),
        repo: format!("{}/{}", repo.owner_name, repo.name),
        number: issue.number,
        title: issue.title.clone(),
        comment: body_text.map(str::to_owned),
        url: format!(
            "{base}/{}/{}/{section}/{}",
            repo.owner_name, repo.name, issue.number
        ),
        settings_url: format!("{base}/settings/notifications"),
    }))
}

fn verb_for(activity: &Activity) -> Verb {
    match activity {
        Activity::Opened => Verb::Opened,
        Activity::Comment => Verb::Commented,
        Activity::Closed => Verb::Closed,
        Activity::Reopened => Verb::Reopened,
        Activity::Merged => Verb::Merged,
        Activity::Review => Verb::Reviewed,
        Activity::Assigned { .. } => Verb::AssignedYou,
        Activity::Pushed => Verb::Pushed,
        Activity::Edited => Verb::MentionedYou,
    }
}

fn why_for(reason: Reason) -> Why {
    match reason {
        Reason::Mention => Why::Mention,
        Reason::Assign => Why::Assign,
        Reason::ReviewRequested => Why::ReviewRequested,
        Reason::Author => Why::Author,
        Reason::Comment => Why::Comment,
        Reason::StateChange | Reason::Subscribed => Why::Subscribed,
        Reason::Watching => Why::Watching,
    }
}

/// Queues the email copy of one notification for `user_id`. Nothing is queued unless the user has
/// a verified address and email notifications on. The idempotency key is per event and user, so
/// a repeated run sends nothing twice.
async fn queue_email(
    state: &AppState,
    plan: &MailPlan,
    user_id: &str,
    reason: Reason,
) -> Result<()> {
    let Some(user) = state.store.user_by_id(user_id).await? else {
        return Ok(());
    };
    let Some(address) = user.email.as_deref().filter(|address| !address.is_empty()) else {
        return Ok(());
    };
    if !user.email_notifications || !state.store.email_is_verified(user_id).await? {
        return Ok(());
    }
    let lang = i18n::resolve(Some(user.language.as_str()), &HeaderMap::new());
    let (subject, body) = templates::notification(
        lang,
        &templates::Notification {
            actor: plan.actor.as_deref(),
            verb: plan.verb,
            repo: &plan.repo,
            number: plan.number,
            title: &plan.title,
            comment: plan.comment.as_deref(),
            url: &plan.url,
            why: why_for(reason),
            settings_url: &plan.settings_url,
        },
    );
    let (message_id, in_reply_to) = if plan.opens_thread {
        (plan.root.clone(), None)
    } else {
        (
            format!("<{}.{user_id}@{}>", plan.event_id, plan.host),
            Some(plan.root.clone()),
        )
    };
    let outgoing = OutgoingMail {
        to: address.to_owned(),
        subject,
        body,
        message_id: Some(message_id),
        in_reply_to,
    };
    let idem_key = format!("notif:{}:{user_id}", plan.event_id);
    mail::enqueue_message(state, &idem_key, &outgoing).await?;
    Ok(())
}

/// The host of `base_url`, without scheme, port or path. Used as the Message-ID domain.
fn host_of(base_url: &str) -> String {
    let rest = base_url
        .split_once("://")
        .map_or(base_url, |(_, rest)| rest);
    let authority = rest.split(['/', '?', '#']).next().unwrap_or_default();
    let host = authority
        .rsplit_once('@')
        .map_or(authority, |(_, host)| host);
    let host = host.split(':').next().unwrap_or_default();
    if host.is_empty() {
        "localhost".to_owned()
    } else {
        host.to_owned()
    }
}

/// True when the user has an explicit `ignored` subscription on the issue.
async fn is_ignored(state: &AppState, user_id: &str, issue_id: &str) -> Result<bool> {
    Ok(matches!(
        state.store.subscription(user_id, issue_id).await?,
        Some((SubState::Ignored, _))
    ))
}

#[cfg(test)]
mod tests {
    use forge_core::{Notification, Role, Visibility};

    use super::*;
    use crate::test_support::test_state;

    struct Fixture {
        state: AppState,
        repo: Repo,
        issue: Issue,
        owner: User,
        alice: User,
        bob: User,
        carol: User,
        _dir: tempfile::TempDir,
    }

    async fn fixture(visibility: Visibility) -> Fixture {
        let dir = tempfile::tempdir().expect("tempdir");
        let state = test_state(dir.path()).await;
        let store = &state.store;
        let owner = store
            .create_user("owner", None, "unused", false)
            .await
            .unwrap();
        let alice = store
            .create_user("alice", None, "unused", false)
            .await
            .unwrap();
        let bob = store
            .create_user("bob", None, "unused", false)
            .await
            .unwrap();
        let carol = store
            .create_user("carol", None, "unused", false)
            .await
            .unwrap();
        let repo = store
            .create_repo(&owner.id, "proj", "", visibility, "main")
            .await
            .unwrap();
        let issue = store
            .create_issue(&repo.id, &owner.id, "a problem", "body")
            .await
            .unwrap();
        Fixture {
            state,
            repo,
            issue,
            owner,
            alice,
            bob,
            carol,
            _dir: dir,
        }
    }

    async fn inbox(fx: &Fixture, user: &User) -> Vec<Notification> {
        fx.state
            .store
            .list_notifications(&user.id, false, None, 50)
            .await
            .unwrap()
    }

    #[test]
    fn mentions_skip_code_links_and_emails() {
        let text = "hi @Alice and @bob. mail a@b.com `@code`\n\n\
                    ```\n@blocked\n```\n[@linked](https://x.test) @alice again";
        assert_eq!(mentioned_usernames(text), vec!["alice", "bob"]);
    }

    #[test]
    fn mentions_in_html_blocks_are_found() {
        let text = "<div>\n@carol\n</div>\n";
        assert_eq!(mentioned_usernames(text), vec!["carol"]);
    }

    #[test]
    fn mentions_are_deduped_case_insensitively() {
        assert_eq!(
            mentioned_usernames("@Bob @bob @BOB"),
            vec!["bob"],
            "one name however it is cased"
        );
        assert!(mentioned_usernames("no mentions here").is_empty());
    }

    #[tokio::test]
    async fn comment_notifies_subscriber_not_the_commenter() {
        let fx = fixture(Visibility::Public).await;
        let store = &fx.state.store;
        store
            .subscribe(
                &fx.bob.id,
                &fx.issue.id,
                SubState::Subscribed,
                Reason::Subscribed,
            )
            .await
            .unwrap();
        store
            .subscribe(
                &fx.alice.id,
                &fx.issue.id,
                SubState::Subscribed,
                Reason::Subscribed,
            )
            .await
            .unwrap();

        dispatch(
            &fx.state,
            &fx.repo,
            &fx.issue,
            &fx.alice.id,
            Activity::Comment,
            Some("hello"),
        )
        .await;

        let bob_inbox = inbox(&fx, &fx.bob).await;
        assert_eq!(bob_inbox.len(), 1);
        assert_eq!(bob_inbox[0].reason, Reason::Subscribed);
        assert_eq!(bob_inbox[0].last_activity, "comment");
        assert!(
            inbox(&fx, &fx.alice).await.is_empty(),
            "the actor is never notified"
        );
    }

    #[tokio::test]
    async fn opened_notifies_watchers_and_auto_subscribes_the_author() {
        let fx = fixture(Visibility::Public).await;
        fx.state
            .store
            .set_watch(&fx.bob.id, &fx.repo.id, WatchLevel::Watching)
            .await
            .unwrap();

        dispatch(
            &fx.state,
            &fx.repo,
            &fx.issue,
            &fx.owner.id,
            Activity::Opened,
            None,
        )
        .await;

        let bob_inbox = inbox(&fx, &fx.bob).await;
        assert_eq!(bob_inbox.len(), 1);
        assert_eq!(bob_inbox[0].reason, Reason::Watching);
        assert!(inbox(&fx, &fx.owner).await.is_empty());
        let sub = fx
            .state
            .store
            .subscription(&fx.owner.id, &fx.issue.id)
            .await
            .unwrap();
        assert_eq!(sub, Some((SubState::Subscribed, Reason::Author)));
    }

    #[tokio::test]
    async fn ignored_subscription_suppresses_watcher() {
        let fx = fixture(Visibility::Public).await;
        let store = &fx.state.store;
        store
            .set_watch(&fx.bob.id, &fx.repo.id, WatchLevel::Watching)
            .await
            .unwrap();
        store
            .subscribe(
                &fx.bob.id,
                &fx.issue.id,
                SubState::Ignored,
                Reason::Subscribed,
            )
            .await
            .unwrap();

        dispatch(
            &fx.state,
            &fx.repo,
            &fx.issue,
            &fx.owner.id,
            Activity::Opened,
            None,
        )
        .await;

        assert!(inbox(&fx, &fx.bob).await.is_empty());
    }

    #[tokio::test]
    async fn ignoring_repo_suppresses_but_mention_still_notifies() {
        let fx = fixture(Visibility::Public).await;
        let store = &fx.state.store;
        store
            .set_watch(&fx.alice.id, &fx.repo.id, WatchLevel::Ignoring)
            .await
            .unwrap();
        store
            .subscribe(
                &fx.alice.id,
                &fx.issue.id,
                SubState::Subscribed,
                Reason::Subscribed,
            )
            .await
            .unwrap();

        dispatch(
            &fx.state,
            &fx.repo,
            &fx.issue,
            &fx.bob.id,
            Activity::Comment,
            Some("plain"),
        )
        .await;
        assert!(inbox(&fx, &fx.alice).await.is_empty());

        dispatch(
            &fx.state,
            &fx.repo,
            &fx.issue,
            &fx.bob.id,
            Activity::Comment,
            Some("@alice look"),
        )
        .await;
        let alice_inbox = inbox(&fx, &fx.alice).await;
        assert_eq!(alice_inbox.len(), 1);
        assert_eq!(alice_inbox[0].reason, Reason::Mention);
    }

    #[tokio::test]
    async fn mention_without_read_access_on_private_repo_is_not_notified() {
        let fx = fixture(Visibility::Private).await;
        let store = &fx.state.store;

        dispatch(
            &fx.state,
            &fx.repo,
            &fx.issue,
            &fx.owner.id,
            Activity::Comment,
            Some("@bob hi"),
        )
        .await;
        assert!(inbox(&fx, &fx.bob).await.is_empty());
        let sub = store.subscription(&fx.bob.id, &fx.issue.id).await.unwrap();
        assert_eq!(sub, None, "an unreadable mention does not subscribe");

        store
            .set_collaborator(&fx.repo.id, &fx.bob.id, Role::Read)
            .await
            .unwrap();
        dispatch(
            &fx.state,
            &fx.repo,
            &fx.issue,
            &fx.owner.id,
            Activity::Comment,
            Some("@bob hi"),
        )
        .await;
        let bob_inbox = inbox(&fx, &fx.bob).await;
        assert_eq!(bob_inbox.len(), 1);
        assert_eq!(bob_inbox[0].reason, Reason::Mention);
    }

    #[tokio::test]
    async fn assignee_is_notified_with_assign_reason() {
        let fx = fixture(Visibility::Public).await;

        dispatch(
            &fx.state,
            &fx.repo,
            &fx.issue,
            &fx.owner.id,
            Activity::Assigned {
                assignee_id: fx.bob.id.clone(),
            },
            None,
        )
        .await;

        let bob_inbox = inbox(&fx, &fx.bob).await;
        assert_eq!(bob_inbox.len(), 1);
        assert_eq!(bob_inbox[0].reason, Reason::Assign);
        assert_eq!(bob_inbox[0].last_activity, "assigned");
    }

    #[tokio::test]
    async fn pushed_reaches_subscribers_not_watchers() {
        let fx = fixture(Visibility::Public).await;
        let store = &fx.state.store;
        store
            .set_watch(&fx.bob.id, &fx.repo.id, WatchLevel::Watching)
            .await
            .unwrap();
        store
            .subscribe(
                &fx.carol.id,
                &fx.issue.id,
                SubState::Subscribed,
                Reason::Author,
            )
            .await
            .unwrap();

        dispatch(
            &fx.state,
            &fx.repo,
            &fx.issue,
            &fx.owner.id,
            Activity::Pushed,
            None,
        )
        .await;

        assert!(inbox(&fx, &fx.bob).await.is_empty());
        let carol_inbox = inbox(&fx, &fx.carol).await;
        assert_eq!(carol_inbox.len(), 1);
        assert_eq!(carol_inbox[0].reason, Reason::Author);
    }

    #[tokio::test]
    async fn edit_notifies_only_the_newly_mentioned_user() {
        let fx = fixture(Visibility::Public).await;
        fx.state
            .store
            .set_watch(&fx.bob.id, &fx.repo.id, WatchLevel::Watching)
            .await
            .unwrap();

        dispatch_edit(
            &fx.state,
            &fx.repo,
            &fx.issue,
            &fx.owner.id,
            "hi @bob",
            "hi @bob and @carol",
        )
        .await;

        assert!(
            inbox(&fx, &fx.bob).await.is_empty(),
            "bob was mentioned before the edit"
        );
        let carol_inbox = inbox(&fx, &fx.carol).await;
        assert_eq!(carol_inbox.len(), 1);
        assert_eq!(carol_inbox[0].reason, Reason::Mention);
        assert_eq!(carol_inbox[0].last_activity, "edited");
    }

    #[tokio::test]
    async fn advisory_alerts_reach_owner_admin_collaborators_and_org_owners_only() {
        let dir = tempfile::tempdir().expect("tempdir");
        let state = test_state(dir.path()).await;
        let store = &state.store;
        let owner = store
            .create_user("owner", None, "unused", false)
            .await
            .unwrap();
        let admin = store
            .create_user("admin", None, "unused", false)
            .await
            .unwrap();
        let writer = store
            .create_user("writer", None, "unused", false)
            .await
            .unwrap();
        let repo = store
            .create_repo(&owner.id, "proj", "", Visibility::Private, "main")
            .await
            .unwrap();
        store
            .set_collaborator(&repo.id, &admin.id, Role::Admin)
            .await
            .unwrap();
        store
            .set_collaborator(&repo.id, &writer.id, Role::Write)
            .await
            .unwrap();

        advisories_opened(&state, &repo, 2).await;

        assert_eq!(advisory_rows(store, &owner.id).await, vec![2]);
        assert_eq!(advisory_rows(store, &admin.id).await, vec![2]);
        assert!(advisory_rows(store, &writer.id).await.is_empty());

        // An org-owned repo: its owners and admin collaborators hear, its plain members do not.
        let org = store.create_org("acme", &owner.id).await.unwrap();
        let org_member = store
            .create_user("member", None, "unused", false)
            .await
            .unwrap();
        let org_owner = store
            .create_user("orgowner", None, "unused", false)
            .await
            .unwrap();
        store
            .add_org_member(&org.id, &org_member.id, OrgMemberRole::Member)
            .await
            .unwrap();
        store
            .add_org_member(&org.id, &org_owner.id, OrgMemberRole::Owner)
            .await
            .unwrap();
        let org_repo = store
            .create_repo(&org.id, "shared", "", Visibility::Private, "main")
            .await
            .unwrap();

        advisories_opened(&state, &org_repo, 1).await;

        assert_eq!(advisory_rows(store, &owner.id).await, vec![1, 2]);
        assert_eq!(advisory_rows(store, &org_owner.id).await, vec![1]);
        assert!(advisory_rows(store, &org_member.id).await.is_empty());
        assert!(advisory_rows(store, &org.id).await.is_empty());
    }

    /// The `new_alerts` counts of a user's advisory inbox rows, sorted.
    async fn advisory_rows(store: &forge_core::store::Store, user_id: &str) -> Vec<u64> {
        let mut counts: Vec<u64> = store
            .list_advisory_notifications(user_id, false, None, 50)
            .await
            .unwrap()
            .iter()
            .map(|notice| notice.new_alerts)
            .collect();
        counts.sort_unstable();
        counts
    }

    /// Instance mail on or off. Bob has a verified address and is subscribed to the issue.
    async fn mail_fixture(enabled: bool) -> Fixture {
        let fx = fixture(Visibility::Public).await;
        let store = &fx.state.store;
        store
            .update_mail_settings(&forge_core::mail::MailSettingsUpdate {
                host: "smtp.test.example",
                port: 587,
                security: forge_core::mail::Security::StartTls,
                username: "",
                from_addr: "forge@test.example",
                from_name: "forge",
                enabled,
                password: None,
            })
            .await
            .unwrap();
        store
            .set_email(&fx.bob.id, "bob@test.example")
            .await
            .unwrap();
        let token = store
            .create_verify_token(&fx.bob.id, "bob@test.example")
            .await
            .unwrap();
        store.redeem_verify_token(&token).await.unwrap();
        store
            .subscribe(
                &fx.bob.id,
                &fx.issue.id,
                SubState::Subscribed,
                Reason::Subscribed,
            )
            .await
            .unwrap();
        fx
    }

    /// The outbox rows addressed to `address`.
    async fn mail_to(fx: &Fixture, address: &str) -> Vec<forge_core::mail::OutboxEntry> {
        fx.state
            .store
            .recent_mail(50)
            .await
            .unwrap()
            .into_iter()
            .filter(|row| row.to_addr == address)
            .collect()
    }

    #[tokio::test]
    async fn comment_emails_the_subscriber_threaded_under_the_issue() {
        let fx = mail_fixture(true).await;
        dispatch(
            &fx.state,
            &fx.repo,
            &fx.issue,
            &fx.alice.id,
            Activity::Comment,
            Some("hello there"),
        )
        .await;

        let rows = mail_to(&fx, "bob@test.example").await;
        assert_eq!(rows.len(), 1, "one mail to the subscriber");
        let host = host_of(&fx.state.config.base_url);
        assert_eq!(
            rows[0].in_reply_to.as_deref(),
            Some(format!("<issue-{}@{host}>", fx.issue.id).as_str())
        );
        let message_id = rows[0].message_id.as_deref().unwrap_or_default();
        assert!(message_id.starts_with('<') && message_id.ends_with(&format!("@{host}>")));
        assert_ne!(
            message_id,
            rows[0].in_reply_to.as_deref().unwrap_or_default()
        );
        assert!(rows[0].body.contains("hello there"), "{}", rows[0].body);
        assert!(
            rows[0].body.contains("alice commented on this."),
            "{}",
            rows[0].body
        );
        assert!(
            rows[0].idem_key.starts_with("notif:"),
            "{}",
            rows[0].idem_key
        );
        assert!(rows[0].idem_key.ends_with(&format!(":{}", fx.bob.id)));
        assert!(
            inbox(&fx, &fx.alice).await.is_empty(),
            "the actor gets no mail or inbox row"
        );
    }

    #[tokio::test]
    async fn opening_mail_is_the_thread_root() {
        let fx = mail_fixture(true).await;
        dispatch(
            &fx.state,
            &fx.repo,
            &fx.issue,
            &fx.alice.id,
            Activity::Opened,
            Some("the body"),
        )
        .await;

        let rows = mail_to(&fx, "bob@test.example").await;
        assert_eq!(rows.len(), 1);
        let host = host_of(&fx.state.config.base_url);
        let root = format!("<issue-{}@{host}>", fx.issue.id);
        assert_eq!(rows[0].message_id.as_deref(), Some(root.as_str()));
        assert_eq!(rows[0].in_reply_to, None);
    }

    #[tokio::test]
    async fn no_mail_without_a_verified_address() {
        let fx = mail_fixture(true).await;
        fx.state
            .store
            .clear_email_verified(&fx.bob.id)
            .await
            .unwrap();
        dispatch(
            &fx.state,
            &fx.repo,
            &fx.issue,
            &fx.alice.id,
            Activity::Comment,
            Some("hi"),
        )
        .await;
        assert!(mail_to(&fx, "bob@test.example").await.is_empty());
        assert_eq!(
            inbox(&fx, &fx.bob).await.len(),
            1,
            "the inbox still gets it"
        );
    }

    #[tokio::test]
    async fn no_mail_when_the_user_turned_email_notifications_off() {
        let fx = mail_fixture(true).await;
        fx.state
            .store
            .set_email_notifications(&fx.bob.id, false)
            .await
            .unwrap();
        dispatch(
            &fx.state,
            &fx.repo,
            &fx.issue,
            &fx.alice.id,
            Activity::Comment,
            Some("hi"),
        )
        .await;
        assert!(mail_to(&fx, "bob@test.example").await.is_empty());
        assert_eq!(inbox(&fx, &fx.bob).await.len(), 1);
    }

    #[tokio::test]
    async fn no_mail_when_instance_mail_is_disabled() {
        let fx = mail_fixture(false).await;
        dispatch(
            &fx.state,
            &fx.repo,
            &fx.issue,
            &fx.alice.id,
            Activity::Comment,
            Some("hi"),
        )
        .await;
        assert!(mail_to(&fx, "bob@test.example").await.is_empty());
    }

    #[test]
    fn host_of_drops_scheme_port_and_path() {
        assert_eq!(
            host_of("https://forge.example.com:8443/"),
            "forge.example.com"
        );
        assert_eq!(host_of("http://127.0.0.1:7660"), "127.0.0.1");
        assert_eq!(host_of("forge.test/x"), "forge.test");
        assert_eq!(host_of("https:///"), "localhost");
    }
}