//! Notifications. The inbox at `GET /notifications` (Unread by default, `?all=1` for everything,
//! grouped by repo, 50 rows a page with a `?before=` cursor) and its routes: open a row, mark a
//! row or a whole repo or inbox read, toggle the subscription on an issue or pull, and set the
//! repo watch level. Rows for repos the viewer can no longer read are hidden and marked read.
//! The sidebar box and the watch form are shared with the issue, pull and repo header views.

use std::collections::HashMap;
use std::time::Duration;

use forge_core::authz::{Action, authorize};
use forge_core::discussions::{DiscussionNotification, DiscussionReason};
use forge_core::store::advisories::AdvisoryNotice;
use forge_core::store::transfers::PendingTransfer;
use forge_core::{Notification, Reason, Role, SubState, User, WatchLevel};
use futures_util::{StreamExt, future, stream};
use serde::Deserialize;
use topcoat::Result;
use topcoat::context::{Cx, app_context};
use topcoat::router::content::Form;
use topcoat::router::content::sse::{Event, KeepAlive, Sse};
use topcoat::router::error::{SeeOther, bad_request, not_found, see_other};
use topcoat::router::response::{IntoResponse, Response};
use topcoat::router::{Body, HeaderValue, StatusCode, page, path_param, query_params, route};
use topcoat::view::{View, component, view};

use crate::auth::{current_user, percent_encode, require_csrf, require_user};
use crate::i18n::misc::{self as g, K};
use crate::i18n::{self, Key, Lang};
use crate::notify_hub::publish_unread;
use crate::pages::issue_parts::{parse_number, stamp_unix};
use crate::pages::repo_ctx::{RepoCtx, load_repo};
use crate::pages::repo_transfer::transfer_list_for;
use crate::pages::star::return_path;
use crate::state::AppState;
use crate::views::{csrf_field, display_name, iso_time, now_unix, relative_time, repo_href, shell};

/// Rows per inbox page.
const PAGE_SIZE: u32 = 50;

path_param!(owner);
path_param!(repo);
path_param!(number);
path_param!(note);
path_param!(discussion_id);
path_param!(advisory_repo_id);

#[query_params(error = bad_request)]
struct InboxQuery {
    all: Option<String>,
    before: Option<String>,
}

#[query_params(error = bad_request)]
struct RepoQuery {
    repo: Option<String>,
}

#[derive(Deserialize)]
struct CsrfForm {
    #[serde(default)]
    csrf: String,
}

#[derive(Deserialize)]
struct SubscriptionForm {
    #[serde(default)]
    csrf: String,
    #[serde(default)]
    action: String,
}

#[derive(Deserialize)]
struct WatchForm {
    #[serde(default)]
    csrf: String,
    #[serde(default)]
    level: String,
}

/// One inbox row, with its strings ready for the view.
pub(crate) struct RowView {
    item_class: &'static str,
    icon_class: &'static str,
    kind_label: &'static str,
    open_href: String,
    read_action: String,
    title_line: String,
    reason: &'static str,
    activity: String,
    iso: String,
    absolute: String,
    relative: String,
    unread: bool,
}

/// The rows of one repo, in the order they first appear in the inbox.
pub(crate) struct RepoGroup {
    heading: String,
    href: String,
    mark_action: String,
    rows: Vec<RowView>,
}

/// One inbox entry before it becomes a row: an issue or pull notification, a discussion one, or
/// the security alerts of a repo.
enum Entry {
    Issue(Notification),
    Discussion(DiscussionNotification),
    Advisory(AdvisoryNotice),
}

impl Entry {
    fn repo_id(&self) -> &str {
        match self {
            Entry::Issue(note) => &note.repo_id,
            Entry::Discussion(note) => &note.repo_id,
            Entry::Advisory(note) => &note.repo_id,
        }
    }

    fn repo_owner(&self) -> &str {
        match self {
            Entry::Issue(note) => &note.repo_owner,
            Entry::Discussion(note) => &note.repo_owner,
            Entry::Advisory(note) => &note.repo_owner,
        }
    }

    fn repo_name(&self) -> &str {
        match self {
            Entry::Issue(note) => &note.repo_name,
            Entry::Discussion(note) => &note.repo_name,
            Entry::Advisory(note) => &note.repo_name,
        }
    }

    fn updated_at(&self) -> &str {
        match self {
            Entry::Issue(note) => &note.updated_at,
            Entry::Discussion(note) => &note.updated_at,
            Entry::Advisory(note) => &note.updated_at,
        }
    }
}

/// The row for an issue or pull inbox entry.
fn issue_row(lang: Lang, note: &Notification, tz: &str, now: i64) -> RowView {
    let (kind_label, icon_class) = match (note.is_pull, note.issue_state.as_str()) {
        (false, "open") => (g::t(lang, K::KindOpenIssue), "notif-icon issue-open"),
        (false, _) => (g::t(lang, K::KindClosedIssue), "notif-icon issue-closed"),
        (true, "open") => (g::t(lang, K::KindOpenPull), "notif-icon pull-open"),
        (true, _) => (g::t(lang, K::KindClosedPull), "notif-icon pull-closed"),
    };
    let actor = note
        .last_actor_name
        .as_deref()
        .map(|name| display_name(lang, name));
    let unix = stamp_unix(&note.updated_at);
    RowView {
        item_class: if note.unread {
            "notif-row unread"
        } else {
            "notif-row"
        },
        icon_class,
        kind_label,
        open_href: format!("/notifications/{}/open", percent_encode(&note.id)),
        read_action: format!("/notifications/{}/read", percent_encode(&note.id)),
        title_line: format!(
            "{}/{} #{} {}",
            note.repo_owner, note.repo_name, note.number, note.title
        ),
        reason: reason_label(lang, note.reason),
        activity: g::activity_line(lang, activity_verb(&note.last_activity), actor.as_deref()),
        iso: iso_time(unix),
        absolute: forge_core::tz::format_in_tz(unix, tz),
        relative: relative_time(lang, unix, now),
        unread: note.unread,
    }
}

/// The row for a discussion inbox entry: a mention, or a reply to the viewer's discussion.
fn discussion_row(lang: Lang, note: &DiscussionNotification, tz: &str, now: i64) -> RowView {
    let (reason, activity) = match note.reason {
        DiscussionReason::Mention => (g::t(lang, K::ReasonMentioned), K::VerbMentionedYou),
        DiscussionReason::Reply => (g::t(lang, K::ReasonReplied), K::VerbDiscussionReply),
    };
    let actor = note
        .last_actor_name
        .as_deref()
        .map(|name| display_name(lang, name));
    let unix = stamp_unix(&note.updated_at);
    RowView {
        item_class: if note.unread {
            "notif-row unread"
        } else {
            "notif-row"
        },
        icon_class: "notif-icon discussion",
        kind_label: g::t(lang, K::KindDiscussion),
        open_href: format!(
            "/notifications/discussions/{}/open",
            percent_encode(&note.discussion_id)
        ),
        read_action: format!(
            "/notifications/discussions/{}/read",
            percent_encode(&note.discussion_id)
        ),
        title_line: format!(
            "{}/{} #{} {}",
            note.repo_owner, note.repo_name, note.number, note.title
        ),
        reason,
        activity: g::activity_line(lang, activity, actor.as_deref()),
        iso: iso_time(unix),
        absolute: forge_core::tz::format_in_tz(unix, tz),
        relative: relative_time(lang, unix, now),
        unread: note.unread,
    }
}

/// The row for a repo's security alerts: how many alerts a scan opened since the viewer last read
/// the row. It links to the repo's security page through the open route.
fn advisory_row(lang: Lang, note: &AdvisoryNotice, tz: &str, now: i64) -> RowView {
    let unix = stamp_unix(&note.updated_at);
    let count = note.new_alerts;
    RowView {
        item_class: if note.unread {
            "notif-row unread"
        } else {
            "notif-row"
        },
        icon_class: "notif-icon advisory",
        kind_label: g::t(lang, K::KindSecurityAlerts),
        open_href: format!(
            "/notifications/advisories/{}/open",
            percent_encode(&note.repo_id)
        ),
        read_action: format!(
            "/notifications/advisories/{}/read",
            percent_encode(&note.repo_id)
        ),
        title_line: format!("{}/{}", note.repo_owner, note.repo_name),
        reason: g::t(lang, K::ReasonSecurity),
        activity: g::new_alerts(lang, count),
        iso: iso_time(unix),
        absolute: forge_core::tz::format_in_tz(unix, tz),
        relative: relative_time(lang, unix, now),
        unread: note.unread,
    }
}

#[page("/notifications")]
async fn inbox(cx: &Cx) -> Result<impl View> {
    let viewer = require_user(cx).await?;
    let state = app_context::<AppState>(cx);
    let query = query_params::<InboxQuery>(cx)?;
    let all = query.all.as_deref() == Some("1");
    let before = query.before.clone().filter(|s| !s.is_empty());
    let user = &viewer.0;
    let lang = i18n::lang(cx, Some(user));
    let tz = user.timezone.clone();
    let now = now_unix();

    let rows = state
        .store
        .list_notifications(&user.id, !all, before.as_deref(), PAGE_SIZE)
        .await?;
    let discussion_rows = state
        .store
        .list_discussion_notifications(&user.id, !all, before.as_deref(), PAGE_SIZE)
        .await?;
    let advisory_rows = state
        .store
        .list_advisory_notifications(&user.id, !all, before.as_deref(), PAGE_SIZE)
        .await?;
    // Every list is newest first and holds at most one page, so the top page of the merge is
    // complete. A `more` link is due when any list was full or the merge has rows past the page.
    let lists_full = rows.len() == PAGE_SIZE as usize
        || discussion_rows.len() == PAGE_SIZE as usize
        || advisory_rows.len() == PAGE_SIZE as usize;
    let mut merged: Vec<Entry> = rows
        .into_iter()
        .map(Entry::Issue)
        .chain(discussion_rows.into_iter().map(Entry::Discussion))
        .chain(advisory_rows.into_iter().map(Entry::Advisory))
        .collect();
    merged.sort_by(|a, b| b.updated_at().cmp(a.updated_at()));
    let more_rows = lists_full || merged.len() > PAGE_SIZE as usize;
    merged.truncate(PAGE_SIZE as usize);
    let more = if more_rows {
        merged.last().map(|last| {
            let mut href = format!(
                "/notifications?before={}",
                percent_encode(last.updated_at())
            );
            if all {
                href.push_str("&all=1");
            }
            href
        })
    } else {
        None
    };

    let mut readable: HashMap<String, bool> = HashMap::new();
    let mut hidden = Vec::new();
    let mut groups: Vec<RepoGroup> = Vec::new();
    let mut index: HashMap<String, usize> = HashMap::new();
    for entry in merged {
        let repo_id = entry.repo_id().to_owned();
        let visible = match readable.get(&repo_id) {
            Some(visible) => *visible,
            None => {
                let visible = can_read(state, user, &repo_id).await?;
                readable.insert(repo_id.clone(), visible);
                visible
            }
        };
        // Security alerts are for the repo's managers only, the same rule as the security page.
        let visible = visible
            && (!matches!(entry, Entry::Advisory(_))
                || can_manage_alerts(state, user, &repo_id).await?);
        if !visible {
            hidden.push(entry);
            continue;
        }
        let slot = match index.get(&repo_id) {
            Some(slot) => *slot,
            None => {
                let heading = format!("{}/{}", entry.repo_owner(), entry.repo_name());
                groups.push(RepoGroup {
                    href: repo_href(entry.repo_owner(), entry.repo_name(), "", ""),
                    mark_action: format!(
                        "/notifications/mark_all_read?repo={}",
                        percent_encode(&repo_id)
                    ),
                    heading,
                    rows: Vec::new(),
                });
                index.insert(repo_id.clone(), groups.len() - 1);
                groups.len() - 1
            }
        };
        let row = match entry {
            Entry::Issue(note) => issue_row(lang, &note, &tz, now),
            Entry::Discussion(note) => discussion_row(lang, &note, &tz, now),
            Entry::Advisory(note) => advisory_row(lang, &note, &tz, now),
        };
        groups[slot].rows.push(row);
    }
    // Rows for repos the viewer can no longer read are marked read, so they stop counting.
    for entry in &hidden {
        match entry {
            Entry::Issue(note) => state.store.mark_read(&user.id, &note.id).await?,
            Entry::Discussion(note) => {
                state
                    .store
                    .mark_discussion_notification_read(&user.id, &note.discussion_id)
                    .await?;
            }
            Entry::Advisory(note) => {
                state
                    .store
                    .mark_advisory_notification_read(&user.id, &note.repo_id)
                    .await?;
            }
        }
    }

    if !hidden.is_empty() {
        publish_unread(state, &user.id).await;
    }
    let unread_count = state.store.unread_count(&user.id).await?;
    let transfers = state.store.pending_transfers_for_user(&user.id).await?;
    let csrf = viewer.1.csrf_token.clone();
    Ok(view! {
        shell(title: i18n::t(lang, Key::Notifications), viewer: Some(&viewer),
            inbox_list(lang: lang, all: all, unread_count: unread_count, groups: groups, more: more, csrf: csrf, transfers: transfers)
        )
    })
}

/// Opens one inbox row: marks it read and sends the viewer to the issue or pull. Another user's
/// row, or a repo the viewer can no longer read, is 404.
#[route(GET "/notifications/{note}/open")]
async fn open_notification(cx: &Cx) -> Result<SeeOther> {
    let (user, _) = require_user(cx).await?;
    let state = app_context::<AppState>(cx);
    let id = path_param::<Note>(cx);
    let Some(target) = state.store.notification_target(&user.id, id).await? else {
        return Err(not_found().into());
    };
    state.store.mark_read(&user.id, id).await?;
    publish_unread(state, &user.id).await;
    if !can_read(state, &user, &target.repo_id).await? {
        return Err(not_found().into());
    }
    let section = if target.is_pull { "pulls" } else { "issues" };
    let href = repo_href(
        &target.owner,
        &target.name,
        section,
        &target.number.to_string(),
    );
    Ok(see_other(href))
}

/// Opens one discussion inbox row: marks it read and sends the viewer to the discussion. Another
/// user's row, or a repo the viewer can no longer read, is 404.
#[route(GET "/notifications/discussions/{discussion_id}/open")]
async fn open_discussion_notification(cx: &Cx) -> Result<SeeOther> {
    let (user, _) = require_user(cx).await?;
    let state = app_context::<AppState>(cx);
    let id = path_param::<DiscussionId>(cx);
    let Some(note) = state.store.discussion_notification(&user.id, id).await? else {
        return Err(not_found().into());
    };
    state
        .store
        .mark_discussion_notification_read(&user.id, id)
        .await?;
    publish_unread(state, &user.id).await;
    if !can_read(state, &user, &note.repo_id).await? {
        return Err(not_found().into());
    }
    let href = repo_href(
        &note.repo_owner,
        &note.repo_name,
        "discussions",
        &note.number.to_string(),
    );
    Ok(see_other(href))
}

/// Marks one discussion inbox row read, then returns to the page the viewer came from.
#[route(POST "/notifications/discussions/{discussion_id}/read")]
async fn read_discussion_notification(cx: &Cx, Form(form): Form<CsrfForm>) -> Result<SeeOther> {
    let (user, _) = require_csrf(cx, &form.csrf).await?;
    let state = app_context::<AppState>(cx);
    let id = path_param::<DiscussionId>(cx);
    state
        .store
        .mark_discussion_notification_read(&user.id, id)
        .await?;
    publish_unread(state, &user.id).await;
    Ok(see_other(back(cx)))
}

/// Opens a repo's security alerts row: marks it read and sends the viewer to the repo's security
/// page. Another user's row, or a repo whose alerts the viewer may not manage, is 404.
#[route(GET "/notifications/advisories/{advisory_repo_id}/open")]
async fn open_advisory_notification(cx: &Cx) -> Result<SeeOther> {
    let (user, _) = require_user(cx).await?;
    let state = app_context::<AppState>(cx);
    let repo_id = path_param::<AdvisoryRepoId>(cx);
    let Some(notice) = state.store.advisory_notification(&user.id, repo_id).await? else {
        return Err(not_found().into());
    };
    state
        .store
        .mark_advisory_notification_read(&user.id, repo_id)
        .await?;
    publish_unread(state, &user.id).await;
    if !can_manage_alerts(state, &user, repo_id).await? {
        return Err(not_found().into());
    }
    Ok(see_other(repo_href(
        &notice.repo_owner,
        &notice.repo_name,
        "security",
        "",
    )))
}

/// Marks one repo's security alerts row read, then returns to the page the viewer came from.
#[route(POST "/notifications/advisories/{advisory_repo_id}/read")]
async fn read_advisory_notification(cx: &Cx, Form(form): Form<CsrfForm>) -> Result<SeeOther> {
    let (user, _) = require_csrf(cx, &form.csrf).await?;
    let state = app_context::<AppState>(cx);
    let repo_id = path_param::<AdvisoryRepoId>(cx);
    state
        .store
        .mark_advisory_notification_read(&user.id, repo_id)
        .await?;
    publish_unread(state, &user.id).await;
    Ok(see_other(back(cx)))
}

/// Marks one inbox row read, then returns to the page the viewer came from.
#[route(POST "/notifications/{note}/read")]
async fn read_notification(cx: &Cx, Form(form): Form<CsrfForm>) -> Result<SeeOther> {
    let (user, _) = require_csrf(cx, &form.csrf).await?;
    let state = app_context::<AppState>(cx);
    let id = path_param::<Note>(cx);
    state.store.mark_read(&user.id, id).await?;
    publish_unread(state, &user.id).await;
    Ok(see_other(back(cx)))
}

/// Marks the viewer's unread rows read: in one repo with `?repo=<id>`, otherwise everywhere.
#[route(POST "/notifications/mark_all_read")]
async fn mark_all_read(cx: &Cx, Form(form): Form<CsrfForm>) -> Result<SeeOther> {
    let (user, _) = require_csrf(cx, &form.csrf).await?;
    let state = app_context::<AppState>(cx);
    let query = query_params::<RepoQuery>(cx)?;
    let repo_id = query.repo.as_deref().filter(|s| !s.is_empty());
    state.store.mark_all_read(&user.id, repo_id).await?;
    state
        .store
        .mark_all_discussion_notifications_read(&user.id, repo_id)
        .await?;
    state
        .store
        .mark_all_advisory_notifications_read(&user.id, repo_id)
        .await?;
    publish_unread(state, &user.id).await;
    Ok(see_other(back(cx)))
}

/// How often an idle stream sends a `: keepalive` comment, so a reverse proxy does not drop the
/// connection. This is the only timer in the feature: counts are pushed as they change.
const STREAM_KEEP_ALIVE: Duration = Duration::from_secs(30);

/// One `unread` event carrying the count.
fn unread_event(count: u64) -> topcoat::Result<Event> {
    Ok(Event::new().event("unread").data(count.to_string()))
}

/// The bell's live count. Sends an `unread` event with the viewer's unread count now, then one
/// after each change to the inbox, until the viewer disconnects or the server shuts down. An
/// anonymous viewer gets a bare 401: an event stream has no login page to redirect to.
#[route(GET "/notifications/stream")]
async fn notification_stream(cx: &Cx) -> Result<Response> {
    let Some((user, _)) = current_user(cx).await? else {
        let mut res = Response::new(Body::empty());
        *res.status_mut() = StatusCode::UNAUTHORIZED;
        return Ok(res);
    };
    let state = app_context::<AppState>(cx);
    let subscription = state.notify_hub.subscribe(&user.id);
    // Read after subscribing: a change that lands in between is pushed again, never missed.
    let count = state.store.unread_count(&user.id).await?;
    let first = stream::once(future::ready(unread_event(count)));
    let changes = stream::unfold(subscription, |mut sub| async move {
        let count = sub.next_count().await?;
        Some((unread_event(count), sub))
    });
    let events = Sse::new(first.chain(changes)).keep_alive(
        KeepAlive::new()
            .interval(STREAM_KEEP_ALIVE)
            .text("keepalive"),
    );
    let mut res = events.into_response(cx)?;
    // Nginx buffers a proxied response unless told not to, which would hold the events back.
    res.headers_mut()
        .insert("x-accel-buffering", HeaderValue::from_static("no"));
    Ok(res)
}

/// Subscribes or unsubscribes the viewer from an issue.
#[route(POST "/{owner}/{repo}/issues/{number}/subscription")]
async fn issue_subscription(cx: &Cx, Form(form): Form<SubscriptionForm>) -> Result<SeeOther> {
    set_subscription(cx, false, &form).await
}

/// Subscribes or unsubscribes the viewer from a pull request.
#[route(POST "/{owner}/{repo}/pulls/{number}/subscription")]
async fn pull_subscription(cx: &Cx, Form(form): Form<SubscriptionForm>) -> Result<SeeOther> {
    set_subscription(cx, true, &form).await
}

async fn set_subscription(cx: &Cx, pull: bool, form: &SubscriptionForm) -> Result<SeeOther> {
    let ctx = load_repo(cx, path_param::<Owner>(cx), path_param::<Repo>(cx)).await?;
    let (user, _) = require_csrf(cx, &form.csrf).await?;
    let state = app_context::<AppState>(cx);
    let number = parse_number(path_param::<Number>(cx))?;
    let issue = if pull {
        state
            .store
            .pull_by_number(&ctx.repo.id, number)
            .await?
            .map(|pr| pr.issue)
    } else {
        state.store.issue_by_number(&ctx.repo.id, number).await?
    };
    let Some(issue) = issue else {
        return Err(not_found().into());
    };
    // Keep the reason the thread already has; a new subscription is for an explicit choice.
    let reason = state
        .store
        .subscription(&user.id, &issue.id)
        .await?
        .map_or(Reason::Subscribed, |(_, reason)| reason);
    match form.action.as_str() {
        "subscribe" => {
            state
                .store
                .subscribe(&user.id, &issue.id, SubState::Subscribed, reason)
                .await?
        }
        "unsubscribe" => {
            state
                .store
                .subscribe(&user.id, &issue.id, SubState::Ignored, reason)
                .await?
        }
        _ => return Err(bad_request("unknown subscription action").into()),
    }
    let section = if pull { "pulls" } else { "issues" };
    let href = repo_href(
        &ctx.repo.owner_name,
        &ctx.repo.name,
        section,
        &number.to_string(),
    );
    Ok(see_other(href))
}

/// Sets the viewer's watch level on a repo. The viewer must be able to read the repo.
#[route(POST "/{owner}/{repo}/watch")]
async fn set_watch(cx: &Cx, Form(form): Form<WatchForm>) -> Result<SeeOther> {
    let ctx = load_repo(cx, path_param::<Owner>(cx), path_param::<Repo>(cx)).await?;
    let (user, _) = require_csrf(cx, &form.csrf).await?;
    let level = match form.level.as_str() {
        "participating" => WatchLevel::Participating,
        "watching" => WatchLevel::Watching,
        "ignoring" => WatchLevel::Ignoring,
        _ => return Err(bad_request("unknown watch level").into()),
    };
    let state = app_context::<AppState>(cx);
    state.store.set_watch(&user.id, &ctx.repo.id, level).await?;
    Ok(see_other(repo_href(
        &ctx.repo.owner_name,
        &ctx.repo.name,
        "",
        "",
    )))
}

/// Whether the viewer may manage the repo's security alerts: Write on the repo, the same rule as
/// the security page. Rows for a repo the viewer may not manage are hidden from the inbox.
async fn can_manage_alerts(state: &AppState, user: &User, repo_id: &str) -> Result<bool> {
    let Some(repo) = state.store.repo_by_id(repo_id).await? else {
        return Ok(false);
    };
    let role: Option<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::Write).is_ok())
}

/// Whether the viewer may read the repo right now, by their current collaborator role.
pub(crate) async fn can_read(state: &AppState, user: &User, repo_id: &str) -> Result<bool> {
    let Some(repo) = state.store.repo_by_id(repo_id).await? else {
        return Ok(false);
    };
    let role: Option<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())
}

/// The sidebar box of an issue or pull for a signed-in viewer. Viewing the thread marks the
/// viewer's inbox row for it read. `base` is the item's URL. `None` for anonymous viewers.
pub(crate) async fn notify_box_for(
    state: &AppState,
    ctx: &RepoCtx,
    lang: Lang,
    issue_id: &str,
    base: &str,
) -> Result<Option<NotifyBox>> {
    let Some((user, session)) = &ctx.viewer else {
        return Ok(None);
    };
    state.store.mark_issue_read(&user.id, issue_id).await?;
    publish_unread(state, &user.id).await;
    let level = state.store.watch_level(&user.id, &ctx.repo.id).await?;
    let subscription = state.store.subscription(&user.id, issue_id).await?;
    let (subscribed, reason) = if level == WatchLevel::Ignoring {
        (false, g::t(lang, K::NotReceivingRepo).to_owned())
    } else {
        match subscription {
            Some((SubState::Ignored, _)) => (false, g::t(lang, K::NotReceivingThread).to_owned()),
            Some((SubState::Subscribed, reason)) => (true, because(lang, reason)),
            None if level == WatchLevel::Watching => {
                (true, g::receiving_because(lang, K::WhyWatching))
            }
            None => (false, g::t(lang, K::NotReceivingThread).to_owned()),
        }
    };
    Ok(Some(NotifyBox {
        action: format!("{base}/subscription"),
        csrf: session.csrf_token.clone(),
        subscribed,
        reason,
    }))
}

/// The notification sidebar box data: where the form posts, the session CSRF token, whether the
/// viewer is receiving notifications, and the sentence explaining why.
pub(crate) struct NotifyBox {
    pub(crate) action: String,
    pub(crate) csrf: String,
    pub(crate) subscribed: bool,
    pub(crate) reason: String,
}

/// "You're receiving notifications because ..." for a stored subscription reason.
fn because(lang: Lang, reason: Reason) -> String {
    let why = match reason {
        Reason::Mention => K::WhyMentioned,
        Reason::Assign => K::WhyAssigned,
        Reason::ReviewRequested => K::WhyReviewRequested,
        Reason::Author => K::WhyAuthor,
        Reason::Comment => K::WhyCommented,
        Reason::StateChange | Reason::Subscribed => K::WhySubscribed,
        Reason::Watching => K::WhyWatching,
    };
    g::receiving_because(lang, why)
}

/// The human label of a notification reason.
fn reason_label(lang: Lang, reason: Reason) -> &'static str {
    match reason {
        Reason::Mention => g::t(lang, K::ReasonMentioned),
        Reason::Assign => g::t(lang, K::ReasonAssigned),
        Reason::ReviewRequested => g::t(lang, K::ReasonReviewRequested),
        Reason::Author => g::t(lang, K::ReasonAuthor),
        Reason::Comment => g::t(lang, K::ReasonCommented),
        Reason::StateChange => g::t(lang, K::ReasonStateChange),
        Reason::Subscribed => g::t(lang, K::ReasonSubscribed),
        Reason::Watching => g::t(lang, K::ReasonWatching),
    }
}

/// The verb shown for the last activity on a thread.
fn activity_verb(activity: &str) -> K {
    match activity {
        "opened" => K::VerbOpened,
        "comment" => K::VerbCommented,
        "closed" => K::VerbClosed,
        "reopened" => K::VerbReopened,
        "merged" => K::VerbMerged,
        "review" => K::VerbReviewed,
        "assigned" => K::VerbAssigned,
        "mentioned" => K::VerbMentionedYou,
        "pushed" => K::VerbPushedTo,
        _ => K::VerbUpdated,
    }
}

/// Where the POST forms return: the Referer when it is a same-site path, else the inbox.
fn back(cx: &Cx) -> String {
    return_path(cx).unwrap_or_else(|| "/notifications".to_owned())
}

/// The inbox body: heading, mark-all button, tabs, the repo groups and the older-rows link.
#[component]
async fn inbox_list(
    lang: Lang,
    all: bool,
    unread_count: u64,
    groups: Vec<RepoGroup>,
    more: Option<String>,
    csrf: String,
    transfers: Vec<PendingTransfer>,
) -> Result<impl View> {
    let empty = groups.is_empty();
    let unread_class = if all { "" } else { "active" };
    let all_class = if all { "active" } else { "" };
    let unread_text = format!("{} ({unread_count})", g::t(lang, K::Unread));
    Ok(view! {
        <section class="page-head">
            <h1>(i18n::t(lang, Key::Notifications))</h1>
            if !empty {
                <form method="post" action="/notifications/mark_all_read" class="notif-inline-form">
                    csrf_field(token: csrf.as_str())
                    <button type="submit" class="button">(g::t(lang, K::MarkAllAsRead))</button>
                </form>
            }
        </section>
        transfer_list_for(lang: lang, transfers: transfers, csrf: csrf.clone())
        <nav class="notif-tabs" aria-label=(g::t(lang, K::NotificationFilter))>
            <a href="/notifications" class=(unread_class)>(unread_text.as_str())</a>
            <a href="/notifications?all=1" class=(all_class)>(g::t(lang, K::AllTab))</a>
        </nav>
        if empty {
            <p class="muted">
                if all {
                    (g::t(lang, K::NoNotificationsYet))
                } else {
                    (g::t(lang, K::NoUnreadNotifications))
                }
            </p>
        }
        for group in groups.into_iter() {
            <section class="card notif-group">
                <header class="notif-group-head">
                    <a class="notif-repo" href=(group.href.as_str())>(group.heading.as_str())</a>
                    <form method="post" action=(group.mark_action.as_str()) class="notif-inline-form">
                        csrf_field(token: csrf.as_str())
                        <button type="submit" class="link-button">(g::t(lang, K::MarkAllRead))</button>
                    </form>
                </header>
                <ul class="notif-list">
                    for row in group.rows.into_iter() {
                        <li class=(row.item_class)>
                            <span class=(row.icon_class) aria-hidden="true"></span>
                            <span class="notif-sr">(row.kind_label)</span>
                            <a class="notif-title" href=(row.open_href.as_str())>(row.title_line.as_str())</a>
                            <span class="badge">(row.reason)</span>
                            <span class="muted small">(row.activity.as_str())</span>
                            <time class="muted small" datetime=(row.iso.as_str()) title=(row.absolute.as_str())>(row.relative.as_str())</time>
                            if row.unread {
                                <form method="post" action=(row.read_action.as_str()) class="notif-inline-form">
                                    csrf_field(token: csrf.as_str())
                                    <button type="submit" class="link-button">(g::t(lang, K::MarkRead))</button>
                                </form>
                            }
                        </li>
                    }
                </ul>
            </section>
        }
        if let Some(href) = more {
            <p class="notif-more"><a class="button" href=(href.as_str())>(g::t(lang, K::Older))</a></p>
        }
    })
}

/// The Notifications box in the issue and pull sidebar: the reason and a Subscribe or
/// Unsubscribe button.
#[component]
pub(crate) async fn notify_box(
    lang: Lang,
    action: String,
    csrf: String,
    subscribed: bool,
    reason: String,
) -> Result<impl View> {
    let (label, value_text) = if subscribed {
        (g::t(lang, K::Unsubscribe), "unsubscribe")
    } else {
        (g::t(lang, K::Subscribe), "subscribe")
    };
    Ok(view! {
        <section class="card notif-box">
            <h2>(i18n::t(lang, Key::Notifications))</h2>
            <p class="muted small">(reason.as_str())</p>
            <form method="post" action=(action.as_str()) class="notif-inline-form">
                csrf_field(token: csrf.as_str())
                <button type="submit" name="action" value=(value_text)>(label)</button>
            </form>
        </section>
    })
}

/// The Watch control in the repo header: a dropdown whose summary shows the current level, one
/// POST form per level inside it (the current one marked), and the watcher count. The forms post
/// without JavaScript; nav.js submits them in place and closes the menu.
#[component]
pub(crate) async fn watch_form(
    lang: Lang,
    action: String,
    csrf: String,
    level: &'static str,
    watchers: u64,
) -> Result<impl View> {
    let count = g::watchers_count(lang, watchers);
    let current = match level {
        "watching" => g::t(lang, K::OptWatching),
        "ignoring" => g::t(lang, K::OptIgnoring),
        _ => g::t(lang, K::OptParticipating),
    };
    let class = |value: &str| -> &'static str {
        if level == value {
            "watch-option is-current"
        } else {
            "watch-option"
        }
    };
    let (cls_participating, cls_watching, cls_ignoring) =
        (class("participating"), class("watching"), class("ignoring"));
    Ok(view! {
        <div class="watch-control">
            <details class="menu watch-menu">
                <summary>
                    <svg class="watch-eye" viewBox="0 0 24 24" aria-hidden="true"><path d="M1 12s4-7 11-7 11 7 11 7-4 7-11 7S1 12 1 12z"/><circle cx="12" cy="12" r="3"/></svg>
                    <span>(current)</span>
                </summary>
                <ul class="menu-list watch-list">
                    <li>
                        <form method="post" action=(action.as_str()) class="watch-option-form">
                            csrf_field(token: csrf.as_str())
                            <input type="hidden" name="level" value="participating">
                            <button type="submit" class=(cls_participating)>
                                <span class="watch-option-title">(g::t(lang, K::OptParticipating))</span>
                                <span class="watch-option-desc">(g::t(lang, K::WatchDescParticipating))</span>
                            </button>
                        </form>
                    </li>
                    <li>
                        <form method="post" action=(action.as_str()) class="watch-option-form">
                            csrf_field(token: csrf.as_str())
                            <input type="hidden" name="level" value="watching">
                            <button type="submit" class=(cls_watching)>
                                <span class="watch-option-title">(g::t(lang, K::OptWatching))</span>
                                <span class="watch-option-desc">(g::t(lang, K::WatchDescWatching))</span>
                            </button>
                        </form>
                    </li>
                    <li>
                        <form method="post" action=(action.as_str()) class="watch-option-form">
                            csrf_field(token: csrf.as_str())
                            <input type="hidden" name="level" value="ignoring">
                            <button type="submit" class=(cls_ignoring)>
                                <span class="watch-option-title">(g::t(lang, K::OptIgnoring))</span>
                                <span class="watch-option-desc">(g::t(lang, K::WatchDescIgnoring))</span>
                            </button>
                        </form>
                    </li>
                </ul>
            </details>
            <span class="muted small">(count.as_str())</span>
        </div>
    })
}

#[cfg(test)]
mod tests {
    use forge_core::{Issue, Reason, Role, Store, SubState, Visibility, WatchLevel};
    use futures_util::StreamExt;
    use topcoat::router::request::Request;
    use topcoat::router::{Body, Router, to_bytes};

    use crate::auth::SESSION_COOKIE;

    const CSRF: &str = "notif-test-csrf";

    struct Site {
        router: Router,
        store: Store,
        _dir: tempfile::TempDir,
    }

    struct Reply {
        status: u16,
        location: Option<String>,
        body: String,
    }

    async fn site() -> Site {
        let dir = tempfile::tempdir().expect("tempdir");
        let state = crate::test_support::test_state(dir.path()).await;
        let store = state.store.clone();
        let router = crate::test_support::app_router(state);
        Site {
            router,
            store,
            _dir: dir,
        }
    }

    /// Signs `user_id` in with `token` and returns the cookie header value.
    async fn sign_in(store: &Store, user_id: &str, token: &str) -> String {
        crate::test_support::create_session(store, user_id, token, CSRF).await;
        format!("__Host-{SESSION_COOKIE}={token}")
    }

    async fn call(
        site: &Site,
        method: &str,
        uri: &str,
        cookie: Option<&str>,
        form: Option<String>,
    ) -> Reply {
        let mut builder = Request::builder().method(method).uri(uri);
        if let Some(cookie) = cookie {
            builder = builder.header("cookie", cookie);
        }
        let body = match form {
            Some(form) => {
                builder = builder.header("content-type", "application/x-www-form-urlencoded");
                Body::from(form)
            }
            None => Body::empty(),
        };
        let res = site
            .router
            .handle(builder.body(body).expect("request"))
            .await;
        let status = res.status().as_u16();
        let location = res
            .headers()
            .get("location")
            .and_then(|v| v.to_str().ok())
            .map(str::to_owned);
        let bytes = to_bytes(res.into_body(), usize::MAX).await.expect("body");
        Reply {
            status,
            location,
            body: String::from_utf8_lossy(&bytes).into_owned(),
        }
    }

    async fn get(site: &Site, uri: &str, cookie: Option<&str>) -> Reply {
        call(site, "GET", uri, cookie, None).await
    }

    async fn post(site: &Site, uri: &str, cookie: &str, extra: &str) -> Reply {
        call(
            site,
            "POST",
            uri,
            Some(cookie),
            Some(format!("csrf={CSRF}{extra}")),
        )
        .await
    }

    /// Alice owns `demo` with one issue (#1), and Bob has a notification row for it.
    struct World {
        alice: forge_core::User,
        bob: forge_core::User,
        repo: forge_core::Repo,
        issue: Issue,
    }

    async fn world(site: &Site, visibility: Visibility) -> World {
        let store = &site.store;
        let alice = store
            .create_user("alice", None, "x", false)
            .await
            .expect("alice");
        let bob = store
            .create_user("bob", None, "x", false)
            .await
            .expect("bob");
        let repo = store
            .create_repo(&alice.id, "demo", "", visibility, "main")
            .await
            .expect("repo");
        let issue = store
            .create_issue(&repo.id, &alice.id, "First", "body")
            .await
            .expect("issue");
        World {
            alice,
            bob,
            repo,
            issue,
        }
    }

    async fn notify(site: &Site, w: &World, reason: Reason) {
        site.store
            .upsert_notification(
                &w.bob.id,
                &w.repo.id,
                &w.issue.id,
                reason,
                "comment",
                Some(&w.alice.id),
            )
            .await
            .expect("notification");
    }

    async fn first_note_id(site: &Site, user_id: &str) -> String {
        site.store
            .list_notifications(user_id, false, None, 50)
            .await
            .expect("list")
            .first()
            .expect("one notification")
            .id
            .clone()
    }

    #[tokio::test]
    async fn inbox_lists_unread_and_all_shows_read() {
        let site = site().await;
        let w = world(&site, Visibility::Public).await;
        let cookie = sign_in(&site.store, &w.bob.id, "bob-token").await;
        notify(&site, &w, Reason::Mention).await;
        let id = first_note_id(&site, &w.bob.id).await;
        site.store.mark_read(&w.bob.id, &id).await.expect("read");

        let unread = get(&site, "/notifications", Some(&cookie)).await;
        assert_eq!(unread.status, 200);
        assert!(!unread.body.contains("alice/demo #1 First"));
        assert!(unread.body.contains("You have no unread notifications."));

        let all = get(&site, "/notifications?all=1", Some(&cookie)).await;
        assert_eq!(all.status, 200);
        assert!(all.body.contains("alice/demo #1 First"));
        assert!(all.body.contains("mentioned"));
    }

    #[tokio::test]
    async fn unread_inbox_lists_unread_rows_grouped_by_repo() {
        let site = site().await;
        let w = world(&site, Visibility::Public).await;
        let cookie = sign_in(&site.store, &w.bob.id, "bob-token").await;
        notify(&site, &w, Reason::Assign).await;

        let page = get(&site, "/notifications", Some(&cookie)).await;
        assert_eq!(page.status, 200);
        assert!(page.body.contains("alice/demo #1 First"));
        assert!(page.body.contains("assigned"));
        assert!(page.body.contains("/notifications/mark_all_read?repo="));
        assert!(page.body.contains("Mark all as read"));
    }

    #[tokio::test]
    async fn open_marks_read_and_redirects_to_the_issue() {
        let site = site().await;
        let w = world(&site, Visibility::Public).await;
        let cookie = sign_in(&site.store, &w.bob.id, "bob-token").await;
        notify(&site, &w, Reason::Comment).await;
        let id = first_note_id(&site, &w.bob.id).await;

        let res = get(&site, &format!("/notifications/{id}/open"), Some(&cookie)).await;
        assert_eq!(res.status, 303);
        assert_eq!(res.location.as_deref(), Some("/alice/demo/issues/1"));
        assert_eq!(site.store.unread_count(&w.bob.id).await.expect("count"), 0);
    }

    #[tokio::test]
    async fn another_users_notification_is_404() {
        let site = site().await;
        let w = world(&site, Visibility::Public).await;
        let alice_cookie = sign_in(&site.store, &w.alice.id, "alice-token").await;
        notify(&site, &w, Reason::Comment).await;
        let id = first_note_id(&site, &w.bob.id).await;

        let res = get(
            &site,
            &format!("/notifications/{id}/open"),
            Some(&alice_cookie),
        )
        .await;
        assert_eq!(res.status, 404);
        assert_eq!(site.store.unread_count(&w.bob.id).await.expect("count"), 1);

        let read = post(
            &site,
            &format!("/notifications/{id}/read"),
            &alice_cookie,
            "",
        )
        .await;
        assert_eq!(read.status, 303);
        assert_eq!(site.store.unread_count(&w.bob.id).await.expect("count"), 1);
    }

    #[tokio::test]
    async fn mark_all_read_clears_the_inbox() {
        let site = site().await;
        let w = world(&site, Visibility::Public).await;
        let cookie = sign_in(&site.store, &w.bob.id, "bob-token").await;
        notify(&site, &w, Reason::Comment).await;

        let res = post(&site, "/notifications/mark_all_read", &cookie, "").await;
        assert_eq!(res.status, 303);
        assert_eq!(site.store.unread_count(&w.bob.id).await.expect("count"), 0);
    }

    #[tokio::test]
    async fn the_stream_needs_a_session() {
        let site = site().await;
        let res = get(&site, "/notifications/stream", None).await;
        assert_eq!(res.status, 401);
    }

    #[tokio::test]
    async fn the_stream_sends_the_count_then_each_change() {
        let site = site().await;
        let w = world(&site, Visibility::Public).await;
        let cookie = sign_in(&site.store, &w.bob.id, "bob-token").await;
        notify(&site, &w, Reason::Mention).await;

        let request = Request::builder()
            .method("GET")
            .uri("/notifications/stream")
            .header("cookie", cookie.as_str())
            .body(Body::empty())
            .expect("request");
        let res = site.router.handle(request).await;
        assert_eq!(res.status().as_u16(), 200);
        assert_eq!(
            res.headers().get("content-type").unwrap(),
            "text/event-stream"
        );
        assert_eq!(res.headers().get("x-accel-buffering").unwrap(), "no");

        let mut frames = res.into_body().into_data_stream();
        let first = frames.next().await.expect("a frame").expect("data");
        assert_eq!(&first[..], b"event: unread\ndata: 1\n\n");

        // Marking the row read pushes the new count to the open stream.
        let read = post(&site, "/notifications/mark_all_read", &cookie, "").await;
        assert_eq!(read.status, 303);
        let next = frames.next().await.expect("a frame").expect("data");
        assert_eq!(&next[..], b"event: unread\ndata: 0\n\n");
    }

    #[tokio::test]
    async fn header_bell_shows_the_unread_count() {
        let site = site().await;
        let w = world(&site, Visibility::Public).await;
        let cookie = sign_in(&site.store, &w.bob.id, "bob-token").await;

        let none = get(&site, "/", Some(&cookie)).await;
        assert!(none.body.contains("href=\"/notifications\""));
        assert!(!none.body.contains("bell-count"));

        notify(&site, &w, Reason::Mention).await;
        let one = get(&site, "/", Some(&cookie)).await;
        assert!(one.body.contains("<span class=\"bell-count\">1</span>"));
    }

    #[tokio::test]
    async fn subscription_toggle_on_an_issue() {
        let site = site().await;
        let w = world(&site, Visibility::Public).await;
        let cookie = sign_in(&site.store, &w.bob.id, "bob-token").await;

        let res = post(
            &site,
            "/alice/demo/issues/1/subscription",
            &cookie,
            "&action=subscribe",
        )
        .await;
        assert_eq!(res.status, 303);
        assert_eq!(res.location.as_deref(), Some("/alice/demo/issues/1"));
        let sub = site
            .store
            .subscription(&w.bob.id, &w.issue.id)
            .await
            .expect("sub");
        assert_eq!(sub, Some((SubState::Subscribed, Reason::Subscribed)));

        let page = get(&site, "/alice/demo/issues/1", Some(&cookie)).await;
        assert!(page.body.contains("because you subscribed to this."));
        assert!(page.body.contains("value=\"unsubscribe\""));

        post(
            &site,
            "/alice/demo/issues/1/subscription",
            &cookie,
            "&action=unsubscribe",
        )
        .await;
        let sub = site
            .store
            .subscription(&w.bob.id, &w.issue.id)
            .await
            .expect("sub");
        assert_eq!(sub, Some((SubState::Ignored, Reason::Subscribed)));
    }

    #[tokio::test]
    async fn pull_subscription_404_for_unknown_number() {
        let site = site().await;
        let w = world(&site, Visibility::Public).await;
        let cookie = sign_in(&site.store, &w.bob.id, "bob-token").await;
        let res = post(
            &site,
            "/alice/demo/pulls/1/subscription",
            &cookie,
            "&action=subscribe",
        )
        .await;
        assert_eq!(res.status, 404);
    }

    #[tokio::test]
    async fn watch_form_sets_the_level() {
        let site = site().await;
        let w = world(&site, Visibility::Public).await;
        let cookie = sign_in(&site.store, &w.bob.id, "bob-token").await;

        let res = post(&site, "/alice/demo/watch", &cookie, "&level=watching").await;
        assert_eq!(res.status, 303);
        assert_eq!(
            site.store
                .watch_level(&w.bob.id, &w.repo.id)
                .await
                .expect("level"),
            WatchLevel::Watching
        );

        let header = get(&site, "/alice/demo/issues/1", Some(&cookie)).await;
        assert!(header.body.contains("action=\"/alice/demo/watch\""));
        assert!(header.body.contains("<details class=\"menu watch-menu\">"));
        assert!(!header.body.contains("<select"));
        // One form per level, and only the watched level is marked current.
        assert_eq!(header.body.matches("name=\"level\"").count(), 3);
        assert_eq!(
            header
                .body
                .matches("class=\"watch-option is-current\"")
                .count(),
            1
        );
        assert_eq!(header.body.matches("class=\"watch-option\"").count(), 2);
        assert!(header.body.contains("1 watching"));

        let bad = post(&site, "/alice/demo/watch", &cookie, "&level=loud").await;
        assert_eq!(bad.status, 400);
    }

    #[tokio::test]
    async fn anonymous_viewer_is_redirected_to_login() {
        let site = site().await;
        let _w = world(&site, Visibility::Public).await;
        let res = get(&site, "/notifications", None).await;
        assert_eq!(res.status, 303);
        assert!(res.location.unwrap_or_default().starts_with("/login"));
    }

    #[tokio::test]
    async fn row_for_a_repo_the_viewer_lost_access_to_is_hidden() {
        let site = site().await;
        let w = world(&site, Visibility::Private).await;
        let cookie = sign_in(&site.store, &w.bob.id, "bob-token").await;
        site.store
            .set_collaborator(&w.repo.id, &w.bob.id, Role::Read)
            .await
            .expect("collaborator");
        notify(&site, &w, Reason::Mention).await;
        assert_eq!(site.store.unread_count(&w.bob.id).await.expect("count"), 1);

        site.store
            .remove_collaborator(&w.repo.id, &w.bob.id)
            .await
            .expect("remove");
        let page = get(&site, "/notifications?all=1", Some(&cookie)).await;
        assert_eq!(page.status, 200);
        assert!(!page.body.contains("alice/demo #1 First"));
        assert_eq!(site.store.unread_count(&w.bob.id).await.expect("count"), 0);
    }

    #[tokio::test]
    async fn discussion_reply_shows_in_inbox_and_open_marks_read() {
        let site = site().await;
        let w = world(&site, Visibility::Public).await;
        // The world's issue takes number 1, so the discussion is #2.
        let discussion = site
            .store
            .create_discussion(
                &w.repo.id,
                &w.alice.id,
                forge_core::discussions::DiscussionCategory::General,
                "Ideas",
                "",
            )
            .await
            .expect("discussion");
        site.store
            .add_discussion_comment(&discussion.id, &w.bob.id, None, "+1", false)
            .await
            .expect("reply");
        let alice = sign_in(&site.store, &w.alice.id, "alice-token").await;
        let bob = sign_in(&site.store, &w.bob.id, "bob-token").await;
        let open = format!("/notifications/discussions/{}/open", discussion.id);

        let page = get(&site, "/notifications", Some(&alice)).await;
        assert_eq!(page.status, 200);
        assert!(page.body.contains("Ideas"));
        assert!(page.body.contains(&open));
        assert_eq!(
            site.store.unread_count(&w.alice.id).await.expect("count"),
            1
        );

        // Bob has no row for the discussion, so he cannot open it.
        let other = get(&site, &open, Some(&bob)).await;
        assert_eq!(other.status, 404);

        let res = get(&site, &open, Some(&alice)).await;
        assert_eq!(res.status, 303);
        assert_eq!(res.location.as_deref(), Some("/alice/demo/discussions/2"));
        assert_eq!(
            site.store.unread_count(&w.alice.id).await.expect("count"),
            0
        );
    }

    #[tokio::test]
    async fn discussion_rows_are_marked_read_by_mark_all() {
        let site = site().await;
        let w = world(&site, Visibility::Public).await;
        let discussion = site
            .store
            .create_discussion(
                &w.repo.id,
                &w.alice.id,
                forge_core::discussions::DiscussionCategory::General,
                "Ideas",
                "",
            )
            .await
            .expect("discussion");
        site.store
            .add_discussion_comment(&discussion.id, &w.bob.id, None, "+1", false)
            .await
            .expect("reply");
        let alice = sign_in(&site.store, &w.alice.id, "alice-token").await;
        assert_eq!(
            site.store.unread_count(&w.alice.id).await.expect("count"),
            1
        );

        let res = post(&site, "/notifications/mark_all_read", &alice, "").await;
        assert_eq!(res.status, 303);
        assert_eq!(
            site.store.unread_count(&w.alice.id).await.expect("count"),
            0
        );
    }

    #[tokio::test]
    async fn advisory_alerts_show_for_managers_and_open_marks_read() {
        let site = site().await;
        let w = world(&site, Visibility::Public).await;
        // Alice owns the repo, so she manages its alerts. Bob is a reader with a stale row.
        site.store
            .notify_new_alerts(&w.alice.id, &w.repo.id, 2)
            .await
            .expect("alerts");
        site.store
            .notify_new_alerts(&w.bob.id, &w.repo.id, 1)
            .await
            .expect("alerts");
        let alice = sign_in(&site.store, &w.alice.id, "alice-token").await;
        let bob = sign_in(&site.store, &w.bob.id, "bob-token").await;
        let open = format!("/notifications/advisories/{}/open", w.repo.id);

        let page = get(&site, "/notifications", Some(&alice)).await;
        assert_eq!(page.status, 200);
        assert!(page.body.contains("Security alerts"));
        assert!(page.body.contains("2 new alerts"));
        assert!(page.body.contains(&open));
        assert_eq!(
            site.store.unread_count(&w.alice.id).await.expect("count"),
            1
        );

        // Bob cannot manage the alerts: his row is hidden and marked read, as lost-access rows are.
        let bob_page = get(&site, "/notifications", Some(&bob)).await;
        assert_eq!(bob_page.status, 200);
        assert!(!bob_page.body.contains(&open));
        assert_eq!(site.store.unread_count(&w.bob.id).await.expect("count"), 0);
        let bob_open = get(&site, &open, Some(&bob)).await;
        assert_eq!(bob_open.status, 404);

        let res = get(&site, &open, Some(&alice)).await;
        assert_eq!(res.status, 303);
        assert_eq!(res.location.as_deref(), Some("/alice/demo/security"));
        assert_eq!(
            site.store.unread_count(&w.alice.id).await.expect("count"),
            0
        );
    }

    #[tokio::test]
    async fn advisory_rows_are_marked_read_by_mark_read_and_mark_all() {
        let site = site().await;
        let w = world(&site, Visibility::Public).await;
        let alice = sign_in(&site.store, &w.alice.id, "alice-token").await;
        let read = format!("/notifications/advisories/{}/read", w.repo.id);

        site.store
            .notify_new_alerts(&w.alice.id, &w.repo.id, 1)
            .await
            .expect("alerts");
        assert_eq!(
            site.store.unread_count(&w.alice.id).await.expect("count"),
            1
        );
        let res = post(&site, &read, &alice, "").await;
        assert_eq!(res.status, 303);
        assert_eq!(
            site.store.unread_count(&w.alice.id).await.expect("count"),
            0
        );

        site.store
            .notify_new_alerts(&w.alice.id, &w.repo.id, 3)
            .await
            .expect("alerts");
        assert_eq!(
            site.store.unread_count(&w.alice.id).await.expect("count"),
            1
        );
        let res = post(&site, "/notifications/mark_all_read", &alice, "").await;
        assert_eq!(res.status, 303);
        assert_eq!(
            site.store.unread_count(&w.alice.id).await.expect("count"),
            0
        );
    }
}