johnandersen777.bsky.social/knotless

Switch repository

Loading…

//! 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<Index>)> {
    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<u32>,
    cursor: Option<String>,
    kind: Option<String>,
}

/// `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<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    repo: Option<RepoViewBasic>,
    #[serde(skip_serializing_if = "Option::is_none")]
    title: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    verdict: Option<String>,
    #[serde(skip_serializing_if = "std::ops::Not::not")]
    mentioned: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    line_comments: Option<u32>,
    #[serde(skip_serializing_if = "Option::is_none")]
    role: Option<String>,
    #[serde(skip_serializing_if = "Vec::is_empty")]
    checks: Vec<knotless::records::FailedCheck>,
    #[serde(skip_serializing_if = "Option::is_none")]
    text: Option<String>,
    #[serde(skip_serializing_if = "std::ops::Not::not")]
    reviewed: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    subject_state: Option<String>,
    is_read: bool,
    created_at: String,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ListOutput {
    notifications: Vec<NotificationView>,
    #[serde(skip_serializing_if = "Option::is_none")]
    cursor: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    seen_at: Option<String>,
}

/// `dev.knotless.listNotifications`.
pub async fn list_notifications(
    State(app): State<Arc<App>>,
    Extension(viewer): Extension<Viewer>,
    Query(p): Query<ListParams>,
) -> Result<Response> {
    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<Notified>, Option<String>)> {
    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<Notified>,
) -> Vec<Notified> {
    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::<RepoRecord>(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<Arc<App>>,
    Extension(viewer): Extension<Viewer>,
) -> Result<Response> {
    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<Arc<App>>,
    Extension(viewer): Extension<Viewer>,
    headers: HeaderMap,
    axum::Json(input): axum::Json<SeenInput>,
) -> Result<Response> {
    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::<jiff::Timestamp>().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 }))
}