//! Notifications: `dev.knotless.listNotifications`, `getUnreadCount` and //! `updateSeen`, for the account a service token or the pages' session //! cookie names. They are worked out from the index as asked for: what //! others did that concerns the asker, and the invitations it has yet to //! answer. Only when it last saw them is kept. //! Those of a private repository come from this server's index of its //! space, where its owner lets it keep one, to its members alone. use super::*; use crate::index::Notified; use crate::viewer::Viewer; use anyhow::Context; use axum::Extension; use std::collections::HashSet; /// The most notifications `getUnreadCount` counts. const UNREAD_MAX: u32 = 100; /// The account asking, and the index, or why not: no one asks, or this /// server keeps no index to find notifications in. pub(super) fn asking(app: &App, viewer: Viewer) -> Result<(String, Arc)> { let Some(did) = viewer.0 else { let why = "ask as an account: with a service token, or logged in to the pages"; return Err(Shown(StatusCode::UNAUTHORIZED, why.into())); }; let index = app.index.clone().ok_or(Shown( StatusCode::NOT_FOUND, "this server keeps no index".into(), ))?; Ok((did, index)) } #[derive(Deserialize)] pub struct ListParams { limit: Option, cursor: Option, kind: Option, } /// `dev.knotless.listNotifications#notification`. #[derive(Serialize)] #[serde(rename_all = "camelCase")] pub struct NotificationView { uri: String, author: ActorView, reason: String, #[serde(skip_serializing_if = "Option::is_none")] reason_subject: Option, #[serde(skip_serializing_if = "Option::is_none")] repo: Option, #[serde(skip_serializing_if = "Option::is_none")] title: Option, #[serde(skip_serializing_if = "Option::is_none")] verdict: Option, #[serde(skip_serializing_if = "std::ops::Not::not")] mentioned: bool, #[serde(skip_serializing_if = "Option::is_none")] line_comments: Option, #[serde(skip_serializing_if = "Option::is_none")] role: Option, #[serde(skip_serializing_if = "Vec::is_empty")] checks: Vec, #[serde(skip_serializing_if = "Option::is_none")] text: Option, #[serde(skip_serializing_if = "std::ops::Not::not")] reviewed: bool, #[serde(skip_serializing_if = "Option::is_none")] subject_state: Option, is_read: bool, created_at: String, } #[derive(Serialize)] #[serde(rename_all = "camelCase")] pub struct ListOutput { notifications: Vec, #[serde(skip_serializing_if = "Option::is_none")] cursor: Option, #[serde(skip_serializing_if = "Option::is_none")] seen_at: Option, } /// `dev.knotless.listNotifications`. pub async fn list_notifications( State(app): State>, Extension(viewer): Extension, Query(p): Query, ) -> Result { let (did, index) = asking(&app, viewer)?; let limit = p.limit.unwrap_or(50).clamp(1, 100); let at = Page { limit, after: p.cursor.as_deref(), kind: p.kind.as_deref(), }; let (found, cursor) = page(&app, &index, &did, at).await?; let seen_at = index .seen(&did) .map_err(anyhow::Error::from) .map_err(AppError::internal)?; let views = found.into_iter().map(|n| view(&app, n, seen_at.as_deref())); let notifications = futures::future::join_all(views).await; Ok(answer(ListOutput { notifications, cursor, seen_at, })) } /// Which page of notifications: at most `limit`, after where the one /// before ended, of one `kind` or all. #[derive(Clone, Copy)] pub(super) struct Page<'a> { pub(super) limit: u32, pub(super) after: Option<&'a str>, pub(super) kind: Option<&'a str>, } /// A page of `did`'s notifications, as [`Index::notifications`] gives /// them: from the index, and from this server's index of the spaces of /// private repositories, of those `did` can still read, in one order. pub(super) async fn page( app: &App, index: &Index, did: &str, Page { limit, after, kind }: Page<'_>, ) -> Result<(Vec, Option)> { let orgs = super::orgs::joined(app, did); let mut subs = index .subscriptions(did) .context("reading what it chose to hear of") .map_err(AppError::internal)?; // Its own repositories it chose nothing for, public and private, it // hears all of, each once. for kept in std::iter::once(index).chain(app.spaces.iter().map(|s| &**s)) { let owned = kept.owned_watches(did, &subs); let owned = owned.context("reading its repositories"); subs.extend(owned.map_err(AppError::internal)?); } let (mut found, cursor) = index .notifications_as(did, &orgs, &subs, limit, after, kind) .context("reading notifications") .map_err(AppError::internal)?; let mut more = cursor.is_some(); // There is one wherever there is a gateway, and so a login. for spaces in app.spaces.iter() { let (private, cursor) = spaces .notifications_as(did, &orgs, &subs, limit, after, kind) .context("reading private notifications") .map_err(AppError::internal)?; more |= cursor.is_some(); let private = index .heard(did, private) .context("reading who hears of what") .map_err(AppError::internal)?; let mut private = readable(app, spaces, did, &orgs, private).await; private.iter_mut().for_each(|n| n.private = true); found.extend(private); } found.sort_by(|a, b| (&b.at, &b.uri).cmp(&(&a.at, &a.uri))); more |= found.len() > limit as usize; found.truncate(limit as usize); let cursor = found.last().map(|n| format!("{} {}", n.at, n.uri)); Ok((found, cursor.filter(|_| more))) } /// Those of `found` about a private repository `did`, the one asking, can /// read: their own; one its record in `spaces`, this server's index of it, /// says it is shared with them, its owner's word; one of an organization /// among `orgs`, theirs; or one they prove they read here. One they were /// taken out of is no longer theirs to hear of. One about no repository, /// as an invitation is, stays. async fn readable( app: &App, spaces: &Index, did: &str, orgs: &[String], found: Vec, ) -> Vec { let repos: HashSet<(String, String)> = found.iter().filter_map(|n| n.repo.clone()).collect(); let mut reads = HashSet::new(); for (owner, name) in repos { let record = spaces.repo(&forge::repo_uri(&owner, &name)).ok().flatten(); let record = record.and_then(|r| serde_json::from_value::(r).ok()); let shared = record.is_some_and(|r| r.is_shared_with(did)); let theirs = owner == did || orgs.contains(&owner); let read = theirs || shared || Repo::find(app, &owner, &name).await.is_ok(); reads.extend(read.then_some((owner, name))); } found .into_iter() .filter(|n| n.repo.as_ref().is_none_or(|repo| reads.contains(repo))) .collect() } async fn view(app: &App, n: Notified, seen: Option<&str>) -> NotificationView { let repo = match n.repo { Some((did, name)) => Some(RepoViewBasic { handle: handle(app, &did).await, did, name, }), None => None, }; NotificationView { author: actor(app, &n.author).await, is_read: seen.is_some_and(|seen| n.at.as_str() <= seen), uri: n.uri, reason: n.reason, reason_subject: n.subject, repo, title: n.title, verdict: n.verdict, mentioned: n.mentioned, line_comments: n.line_comments.filter(|n| *n > 0), role: n.role, checks: n.checks, text: n.text, reviewed: n.reviewed, subject_state: n.state, created_at: n.at, } } #[derive(Serialize)] pub struct UnreadOutput { count: usize, } /// `dev.knotless.getUnreadCount`. pub async fn get_unread_count( State(app): State>, Extension(viewer): Extension, ) -> Result { let (did, index) = asking(&app, viewer)?; let all = Page { limit: UNREAD_MAX, after: None, kind: None, }; let (found, _) = page(&app, &index, &did, all).await?; let seen = index .seen(&did) .map_err(anyhow::Error::from) .map_err(AppError::internal)?; let unread = |n: &&Notified| seen.as_deref().is_none_or(|seen| n.at.as_str() > seen); let count = found.iter().filter(unread).count(); Ok(answer(UnreadOutput { count })) } #[derive(Deserialize)] #[serde(rename_all = "camelCase")] pub struct SeenInput { seen_at: String, } #[derive(Serialize)] #[serde(rename_all = "camelCase")] pub struct SeenOutput { seen_at: String, } /// `dev.knotless.updateSeen`. Asked with the pages' cookie rather than a /// token, only from the site itself. pub async fn update_seen( State(app): State>, Extension(viewer): Extension, headers: HeaderMap, axum::Json(input): axum::Json, ) -> Result { let token = headers.contains_key(header::AUTHORIZATION); if !token && !crate::login::same_site(&headers) { return Err(Shown(StatusCode::FORBIDDEN, "not from this site".into())); } let (did, index) = asking(&app, viewer)?; if input.seen_at.parse::().is_err() { let why = format!("{:?} is not a datetime", input.seen_at); return Err(Shown(StatusCode::BAD_REQUEST, why)); } let seen = index.see(&did, &input.seen_at); let seen_at = seen .map_err(anyhow::Error::from) .map_err(AppError::internal)?; Ok(answer(SeenOutput { seen_at })) }