Notifications by thread: participating versus watching, filters by kind and repository, read, unread and done per thread, and opening a row marks it read eb0c44a9ad37

-- Notifications grouped by thread, filtered by repository, split into

-- participating and watching, and set aside when done (hub spec §10).

ALTER TABLE notifications ADD COLUMN repo_id INTEGER REFERENCES repositories(id) ON DELETE CASCADE;

-- The page the notification is about, without a fragment: one row per thread in the inbox.

ALTER TABLE notifications ADD COLUMN thread TEXT NOT NULL DEFAULT '';

ALTER TABLE notifications ADD COLUMN reason TEXT NOT NULL DEFAULT 'participating' CHECK (reason IN ('participating', 'watching'));

ALTER TABLE notifications ADD COLUMN done INTEGER NOT NULL DEFAULT 0;

UPDATE notifications SET thread = CASE WHEN instr(url, '#') > 0 THEN substr(url, 1, instr(url, '#') - 1) ELSE url END;

UPDATE notifications SET repo_id = (

    SELECT r.id FROM repositories r

    WHERE substr(notifications.url, 1, length('/' || r.owner || '/' || r.name || '/')) = '/' || r.owner || '/' || r.name || '/'

);

CREATE INDEX notifications_by_thread ON notifications (account_id, done, thread, at_ms DESC);

    pub repo_id: Option<i64>,

    pub thread: String,

    pub reason: String,

    pub done: bool,

}

/// Why someone is told: they take part in the thread, or only watch the repository.

#[derive(Clone, Copy, Debug, PartialEq, Eq)]

pub enum Reason {

    Participating,

    Watching,

}

impl Reason {

    pub fn as_str(self) -> &'static str {

        match self {

            Reason::Participating => "participating",

            Reason::Watching => "watching",

        }

    }

/// The thread a notification belongs to: its page, without a fragment.

pub fn thread_of(url: &str) -> &str {

    url.split('#').next().unwrap_or(url)

}

/// A notification to record.

pub struct New<'a> {

    pub account_id: i64,

    pub repo_id: Option<i64>,

    pub reason: Reason,

    pub kind: &'a str,

    pub title: &'a str,

    pub url: &'a str,

    pub actor: &'a str,

}

pub async fn add(pool: &Pool, n: New<'_>) -> sqlx::Result<()> {

    let New { account_id, repo_id, reason, kind, title, url, actor } = n;

    // A new event brings a thread that was set aside back to the inbox.

    sqlx::query("UPDATE notifications SET done = 0 WHERE account_id = ? AND thread = ? AND done = 1")

        .bind(account_id)

        .bind(thread_of(url))

        .execute(pool)

        .await?;

    sqlx::query("INSERT INTO notifications (account_id, at_ms, kind, title, url, actor, repo_id, thread, reason) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)")

        .bind(repo_id)

        .bind(thread_of(url))

        .bind(reason.as_str())

/// One inbox row: a thread, shown by its newest notification.

#[derive(Clone, Debug, FromRow)]

pub struct Thread {

    /// The newest notification's id.

    pub id: i64,

    pub at_ms: i64,

    pub kind: String,

    pub title: String,

    pub url: String,

    pub actor: String,

    pub thread: String,

    pub repo_id: Option<i64>,

    pub repo: Option<String>,

    pub count: i64,

    pub unread: i64,

    pub participating: bool,

    pub done: bool,

}

/// What the inbox shows.

#[derive(Clone, Debug, Default)]

pub struct Filter {

    pub done: bool,

    pub unread_only: bool,

    /// `participating` or `watching`.

    pub reason: Option<String>,

    /// A kind or its family (`issue` matches `issue.closed`).

    pub kind: Option<String>,

    pub repo_id: Option<i64>,

}

pub async fn threads(pool: &Pool, account_id: i64, f: &Filter, limit: i64) -> sqlx::Result<Vec<Thread>> {

    // SQLite takes bare columns from the row holding MAX(at_ms): the newest.

    let mut sql = String::from(

        "SELECT n.id, MAX(n.at_ms) AS at_ms, n.kind, n.title, n.url, n.actor, n.thread, n.repo_id, \

         (SELECT r.owner || '/' || r.name FROM repositories r WHERE r.id = n.repo_id) AS repo, \

         COUNT(*) AS count, SUM(n.read = 0) AS unread, SUM(n.reason = 'participating') > 0 AS participating, n.done \

         FROM notifications n WHERE n.account_id = ? AND n.done = ?",

    );

    if f.kind.is_some() {

        sql.push_str(" AND (n.kind = ? OR n.kind LIKE ? || '.%')");

    }

    if f.repo_id.is_some() {

        sql.push_str(" AND n.repo_id = ?");

    }

    sql.push_str(" GROUP BY n.thread HAVING 1");

    if f.unread_only {

        sql.push_str(" AND SUM(n.read = 0) > 0");

    }

    match f.reason.as_deref() {

        Some("participating") => sql.push_str(" AND SUM(n.reason = 'participating') > 0"),

        Some("watching") => sql.push_str(" AND SUM(n.reason = 'participating') = 0"),

        _ => {}

    }

    sql.push_str(" ORDER BY at_ms DESC LIMIT ?");

    let mut q = sqlx::query_as::<_, Thread>(&sql).bind(account_id).bind(f.done);

    if let Some(k) = &f.kind {

        q = q.bind(k).bind(k);

    }

    if let Some(r) = f.repo_id {

        q = q.bind(r);

    }

    q.bind(limit).fetch_all(pool).await

}

/// Mark a thread read or unread, done or back in the inbox.

pub async fn set_thread(pool: &Pool, account_id: i64, thread: &str, read: Option<bool>, done: Option<bool>) -> sqlx::Result<()> {

    if let Some(r) = read {

        if r {

            sqlx::query("UPDATE notifications SET read = 1 WHERE account_id = ? AND thread = ?")

                .bind(account_id)

                .bind(thread)

                .execute(pool)

                .await?;

        } else {

            // Unread: the newest notification of the thread only.

            sqlx::query("UPDATE notifications SET read = 0 WHERE id = (SELECT id FROM notifications WHERE account_id = ? AND thread = ? ORDER BY at_ms DESC, id DESC LIMIT 1)")

                .bind(account_id)

                .bind(thread)

                .execute(pool)

                .await?;

        }

    }

    if let Some(d) = done {

        let read_too = if d { ", read = 1" } else { "" };

        sqlx::query(&format!("UPDATE notifications SET done = ?{read_too} WHERE account_id = ? AND thread = ?"))

            .bind(d)

            .bind(account_id)

            .bind(thread)

            .execute(pool)

            .await?;

    }

    Ok(())

}

/// The notification with this id, if it is the account's.

pub async fn get(pool: &Pool, account_id: i64, id: i64) -> sqlx::Result<Option<Notification>> {

    sqlx::query_as("SELECT * FROM notifications WHERE account_id = ? AND id = ?").bind(account_id).bind(id).fetch_optional(pool).await

}

    sqlx::query_scalar("SELECT COUNT(*) FROM notifications WHERE account_id = ? AND read = 0 AND done = 0")

        .bind(account_id)

        .fetch_one(pool)

        .await

    // Participants: the thread's subscribers, the mentioned, direct targets.

    // Watchers of the repository who are none of those are only watching.

    let mut participants: BTreeSet<i64> = BTreeSet::new();

            participants.extend(subs);

            participants.insert(a.id);

    participants.extend(ev.direct.iter().copied());

    let mut targets: Vec<(i64, ndb::Reason)> = participants.iter().map(|id| (*id, ndb::Reason::Participating)).collect();

    if let Ok(w) = ndb::watchers(pool, ev.repo_id).await {

        targets.extend(w.into_iter().filter(|id| !participants.contains(id)).map(|id| (id, ndb::Reason::Watching)));

    }

    targets.retain(|(id, _)| *id != ev.actor.id);

    let repo_id = (ev.repo_id > 0).then_some(ev.repo_id);

    for (id, reason) in targets {

        ndb::add(

            pool,

            ndb::New { account_id: id, repo_id, reason, kind: ev.kind, title: &ev.title, url: &ev.url, actor: &ev.actor.username },

        )

        .await

        .ok();

    threads: Vec<ndb::Thread>,

    q: InboxQuery,

    /// Kind families and repositories among the threads in the inbox, with counts.

    kinds: Vec<(String, usize)>,

    repos: Vec<(String, usize)>,

}

/// `?view=unread|done&reason=participating|watching&kind=issue&repo=owner/name`.

#[derive(Clone, Debug, Default, Deserialize)]

pub struct InboxQuery {

    #[serde(default)]

    pub view: String,

    #[serde(default)]

    pub reason: String,

    #[serde(default)]

    pub kind: String,

    #[serde(default)]

    pub repo: String,

}

impl InboxQuery {

    /// This query with one key changed, as a URL (empty values drop out).

    pub fn with(&self, key: &str, value: &str) -> String {

        let mut q = self.clone();

        match key {

            "view" => q.view = value.into(),

            "reason" => q.reason = value.into(),

            "kind" => q.kind = value.into(),

            "repo" => q.repo = value.into(),

            _ => {}

        }

        let parts: Vec<String> = [("view", &q.view), ("reason", &q.reason), ("kind", &q.kind), ("repo", &q.repo)]

            .iter()

            .filter(|(_, v)| !v.is_empty())

            .map(|(k, v)| format!("{k}={}", super::settings::urlencode(v)))

            .collect();

        if parts.is_empty() {

            "/notifications".into()

        } else {

            format!("/notifications?{}", parts.join("&"))

        }

    }

    /// This query as it is, for forms that come back to it.

    pub fn here(&self) -> String {

        self.with("", "")

    }

/// `issue.closed` → `issue`.

fn family(kind: &str) -> &str {

    kind.split('.').next().unwrap_or(kind)

}

async fn inbox_page(

    app: &Shared,

    shell: Shell,

    me: &accounts::Account,

    q: InboxQuery,

    notice: Option<String>,

) -> Result<Page<InboxPage>, Error> {

    let e = |e| Error::from_db(shell.clone(), e);

    let repo_id = match q.repo.split_once('/') {

        Some((o, n)) => db::repos::get(&app.db, o, n).await.map_err(e)?.map(|r| r.id).or(Some(-1)),

        None => None,

    };

    let filter = ndb::Filter {

        done: q.view == "done",

        unread_only: q.view == "unread",

        reason: (!q.reason.is_empty()).then(|| q.reason.clone()),

        kind: (!q.kind.is_empty()).then(|| q.kind.clone()),

        repo_id,

    };

    let threads = ndb::threads(&app.db, me.id, &filter, 200).await.map_err(e)?;

    // Facets over the whole inbox, so the choices do not vanish as you pick.

    let all = ndb::threads(&app.db, me.id, &ndb::Filter { done: filter.done, ..Default::default() }, 1000).await.map_err(e)?;

    let mut kinds: std::collections::BTreeMap<String, usize> = Default::default();

    let mut repos: std::collections::BTreeMap<String, usize> = Default::default();

    for t in &all {

        *kinds.entry(family(&t.kind).to_string()).or_default() += 1;

        if let Some(r) = &t.repo {

            *repos.entry(r.clone()).or_default() += 1;

        }

    }

    let unread = ndb::unread(&app.db, me.id).await.map_err(e)?;

    Ok(Page(InboxPage {

        shell,

        threads,

        unread,

        email_on: fresh.map(|a| a.email_notifications).unwrap_or(false),

        notice,

        q,

        kinds: kinds.into_iter().collect(),

        repos: repos.into_iter().collect(),

    }))

pub async fn inbox(State(app): State<Shared>, viewer: Viewer, Query(q): Query<InboxQuery>) -> Result<Page<InboxPage>, Error> {

    inbox_page(&app, shell, &me.account, q, None).await

    inbox_page(&app, shell, &me.account, InboxQuery::default(), None).await

}

#[derive(Deserialize)]

pub struct ThreadForm {

    #[serde(rename = "_csrf")]

    csrf_token: String,

    thread: String,

    /// `read`, `unread`, `done` or `undone`.

    action: String,

    /// Where to go back to: the inbox with its filters.

    #[serde(default)]

    back: String,

/// `POST /notifications/thread`.

pub async fn thread(State(app): State<Shared>, viewer: Viewer, Form(f): Form<ThreadForm>) -> Result<Response, Error> {

    let shell = Shell::new(&app, "/notifications", &viewer);

    let me = require(&viewer, &shell)?;

    csrf(&viewer, &f.csrf_token, &shell)?;

    let (read, done) = match f.action.as_str() {

        "read" => (Some(true), None),

        "unread" => (Some(false), None),

        "done" => (None, Some(true)),

        "undone" => (None, Some(false)),

        _ => return Err(Error::bad(shell, "unknown action")),

    };

    ndb::set_thread(&app.db, me.account.id, &f.thread, read, done).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    let back = if f.back.starts_with("/notifications") { f.back } else { "/notifications".into() };

    Ok(Redirect::to(&back).into_response())

}

/// `GET /notifications/open/{id}`: mark the thread read and go to it.

pub async fn open(State(app): State<Shared>, viewer: Viewer, axum::extract::Path(id): axum::extract::Path<i64>) -> Result<Response, Error> {

    let shell = Shell::new(&app, "/notifications", &viewer);

    let me = require(&viewer, &shell)?;

    let n = ndb::get(&app.db, me.account.id, id)

        .await

        .map_err(|e| Error::from_db(shell.clone(), e))?

        .ok_or_else(|| Error::not_found(shell.clone()))?;

    ndb::set_thread(&app.db, me.account.id, &n.thread, Some(true), None).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    // Only local pages: a notification's URL is a path on this hub.

    let to = if n.url.starts_with('/') && !n.url.starts_with("//") { n.url } else { "/notifications".into() };

    Ok(Redirect::to(&to).into_response())

}

        InboxQuery::default(),

                        let title = format!("{}#{}: {} (closed by {})", row.slug(), i.number, i.title, &hash[..12.min(hash.len())]);

                        let url = format!("/{}/issue/{}", row.slug(), i.number);

                            ndb::New {

                                account_id: id,

                                repo_id: Some(row.id),

                                reason: ndb::Reason::Participating,

                                kind: "issue.closed",

                                title: &title,

                                url: &url,

                                actor: "",

                            },

        .route("/notifications/thread", post(inbox::thread))

        .route("/notifications/open/{id}", get(inbox::open))

    oxygenhub::db::notify::add(

        &app.db,

        oxygenhub::db::notify::New {

            account_id: bob_id,

            repo_id: None,

            reason: oxygenhub::db::notify::Reason::Participating,

            kind: "mention",

            title: "andy/demo#1: Bug",

            url: "/andy/demo/issue/1",

            actor: "andy",

        },

    )

    .await

    .unwrap();

    oxygenhub::db::notify::add(

        &app.db,

        oxygenhub::db::notify::New {

            account_id: bob_id,

            repo_id: None,

            reason: oxygenhub::db::notify::Reason::Participating,

            kind: "review",

            title: "andy/demo#2: Feat",

            url: "/andy/demo/proposal/2",

            actor: "andy",

        },

    )

    .await

    .unwrap();

    // Bob takes part in #1 (mentioned, assigned, commented) and only watches #3.

    let part = get(app, "/notifications?reason=participating", Some(&bob)).await.body;

    assert!(part.contains("andy/demo#1") && !part.contains("andy/demo#3: Feat"), "{part}");

    let watch = get(app, "/notifications?reason=watching", Some(&bob)).await.body;

    assert!(watch.contains("andy/demo#3: Feat") && !watch.contains("Crash on start"), "{watch}");

    let props = get(app, "/notifications?kind=proposal", Some(&bob)).await.body;

    assert!(props.contains("andy/demo#3: Feat") && !props.contains("andy/demo#1"), "kind filter");

    assert!(get(app, "/notifications?repo=andy/demo", Some(&bob)).await.body.contains("andy/demo#3: Feat"));

    assert!(get(app, "/notifications?repo=nobody/none", Some(&bob)).await.body.contains("Nothing matches these filters"));

    // One row per thread, however many events it had.

    let inbox = get(app, "/notifications", Some(&bob)).await.body;

    assert_eq!(inbox.matches("data-row-thread=\"/andy/demo/issue/1\"").count(), 1, "{inbox}");

    assert!(inbox.contains(" updates"), "the count of a busy thread");

    // Opening a thread marks it read and goes there.

    let bob_id = db::accounts::account_by_name(&app.db, "bob").await.unwrap().unwrap().id;

    let newest: i64 = sqlx::query_scalar(

        "SELECT id FROM notifications WHERE account_id = ? AND thread = '/andy/demo/proposal/3' ORDER BY id DESC LIMIT 1",

    )

    .bind(bob_id)

    .fetch_one(&app.db)

    .await

    .unwrap();

    let before = db::notify::unread(&app.db, bob_id).await.unwrap();

    let r = get(app, &format!("/notifications/open/{newest}"), Some(&bob)).await;

    assert_eq!(r.location.as_deref(), Some("/andy/demo/proposal/3"));

    assert!(db::notify::unread(&app.db, bob_id).await.unwrap() < before);

    assert_eq!(get(app, &format!("/notifications/open/{newest}"), Some(&andy)).await.status, StatusCode::NOT_FOUND, "only your own");

    // Done sets a thread aside until something new happens in it.

    let csrf_b = csrf_of(&get(app, "/notifications", Some(&bob)).await.body);

    let r = post(

        app,

        "/notifications/thread",

        Some(&bob),

        &format!("_csrf={csrf_b}&thread=%2Fandy%2Fdemo%2Fissue%2F1&action=done&back=%2Fnotifications"),

    )

    .await;

    assert_eq!(r.location.as_deref(), Some("/notifications"));

    assert!(!get(app, "/notifications", Some(&bob)).await.body.contains("Crash on start"));

    assert!(get(app, "/notifications?view=done", Some(&bob)).await.body.contains("Crash on start"));

    post(app, "/andy/demo/issue/1/comment", Some(&andy), &format!("_csrf={csrf_a}&body=one+more+thing")).await;

    assert!(get(app, "/notifications", Some(&bob)).await.body.contains("Crash on start"), "back with the new comment");

    let r = post(

        app,

        "/notifications/thread",

        Some(&bob),

        &format!("_csrf={csrf_b}&thread=%2Fandy%2Fdemo%2Fissue%2F1&action=read&back=https%3A%2F%2Fevil.example"),

    )

    .await;

    assert_eq!(r.location.as_deref(), Some("/notifications"), "only back to the inbox");

The inbox shows one row per **thread** (the page a notification is about, such as an issue or a proposal), with its newest event, how many updates it had and how many are new. Each notification records why it was sent: **participating** (the account opened, commented on, was assigned, mentioned in or asked to review the thread) or **watching** (it only watches the repository); a thread counts as participating if any of its notifications does. The inbox filters by view (inbox, unread, done), by reason, by kind (`issue`, `proposal`, `release`, …) and by repository, with counts over the whole inbox. A thread can be marked read or unread, and **done**, which sets it aside (read) until something new happens in it. Opening a row marks its thread read and goes to the page.

<div class="sectioned inbox">

<nav class="section-nav inbox-nav" aria-label="Filters">

  <a href="{{ q.with("view", "") }}"{% if q.view.is_empty() %} class="on" aria-current="page"{% endif %}>Inbox</a>

  <a href="{{ q.with("view", "unread") }}"{% if q.view == "unread" %} class="on" aria-current="page"{% endif %}>Unread{% if unread > 0 %} <span class="count-chip">{{ unread }}</span>{% endif %}</a>

  <a href="{{ q.with("view", "done") }}"{% if q.view == "done" %} class="on" aria-current="page"{% endif %}>Done</a>

  <h3>Why</h3>

  <a href="{{ q.with("reason", "") }}"{% if q.reason.is_empty() %} class="on"{% endif %}>Everything</a>

  <a href="{{ q.with("reason", "participating") }}"{% if q.reason == "participating" %} class="on"{% endif %} title="You opened, commented, were assigned, mentioned or asked to review">Participating</a>

  <a href="{{ q.with("reason", "watching") }}"{% if q.reason == "watching" %} class="on"{% endif %} title="You only watch the repository">Watching</a>

  {% if !kinds.is_empty() %}<h3>Kind</h3>

  <a href="{{ q.with("kind", "") }}"{% if q.kind.is_empty() %} class="on"{% endif %}>All kinds</a>

  {% for (k, n) in kinds %}<a href="{{ q.with("kind", k) }}"{% if &q.kind == k %} class="on"{% endif %}>{{ k }} <span class="small mute">{{ n }}</span></a>{% endfor %}{% endif %}

  {% if !repos.is_empty() %}<h3>Repository</h3>

  <a href="{{ q.with("repo", "") }}"{% if q.repo.is_empty() %} class="on"{% endif %}>All repositories</a>

  {% for (r, n) in repos %}<a href="{{ q.with("repo", r) }}"{% if &q.repo == r %} class="on"{% endif %}>{{ r }} <span class="small mute">{{ n }}</span></a>{% endfor %}{% endif %}

</nav>

<div class="sections">

{% if threads.is_empty() %}

<p class="empty">{% if q.view == "done" %}Nothing set aside.{% else if q.view == "unread" %}All read.{% else if !q.reason.is_empty() || !q.kind.is_empty() || !q.repo.is_empty() %}Nothing matches these filters.{% else %}Nothing yet. Watch a repository or take part in a thread and its events land here.{% endif %}</p>

{% else %}

<ul class="rows threads">

{% for t in threads %}

  <li{% if t.unread > 0 %} class="unread"{% endif %} data-row-thread="{{ t.thread }}">

    <div>

      <a class="name" href="/notifications/open/{{ t.id }}">{% if t.unread > 0 %}<span class="sig sig-valid" title="unread">●</span> {% endif %}{{ t.title }}</a>

      <p class="small">{% if let Some(r) = t.repo %}<span class="mute">{{ r }}</span> · {% endif %}{{ t.kind }} · {{ t.actor }} · <time title="{{ t.at_ms|when_i }}">{{ t.at_ms|ago_i }}</time>{% if t.count > 1 %} · {{ t.count }} updates{% if t.unread > 0 && t.unread < t.count %}, {{ t.unread }} new{% endif %}{% endif %} · <span class="tag{% if !t.participating %} dim{% endif %}">{% if t.participating %}participating{% else %}watching{% endif %}</span></p>

    </div>

    <div class="right">

      <form class="tiny" method="post" action="/notifications/thread"><input type="hidden" name="_csrf" value="{{ shell.csrf }}"><input type="hidden" name="thread" value="{{ t.thread }}"><input type="hidden" name="back" value="{{ q.here() }}">

        {% if t.unread > 0 %}<button name="action" value="read" title="Mark the thread read">read</button>{% else %}<button name="action" value="unread" title="Mark the thread unread">unread</button>{% endif %}

        {% if t.done %}<button name="action" value="undone" title="Back to the inbox">move to inbox</button>{% else %}<button name="action" value="done" title="Set the thread aside until something new happens">done</button>{% endif %}

      </form>

    </div>

  </li>

</div>

</div>

/* Inbox filters (hub spec §10). */

.inbox-nav h3 { margin: 14px 0 4px 12px; font-size: 11px; text-transform: uppercase; letter-spacing: .08em; color: var(--mute); }

.inbox .rows.threads .tag.dim { opacity: .7; }