feat(notifications): an in-app inbox for mentions, replies, tasks and shares

CHANGELOG.md

@@ -8,6 +8,42 @@ so this log can be regenerated from history (e.g. with `git-cliff`).
 
 ## [Unreleased]
 
+### Added
+- **Notifications and an inbox.** An `@mention` in a comment used to fire a
+  `pg_notify('comment_mentions', …)` that no process in the codebase ever
+  listened for, into a `MSG_MENTION` frame the frontend reserved and never
+  received. Mentioning a colleague did nothing they would ever see. Five events
+  now write to a `notifications` table — mentions, replies in threads you are
+  part of, task assignment, overdue tasks, and documents shared with you — read
+  through an Inbox in the sidebar with an unread badge and a `/notifications`
+  page. Delivery is a 30-second poll that also refreshes on window focus; the
+  table is shaped as an outbox (`emailed_at`) so email can arrive later without
+  a migration. Idempotency is a unique index on `(user_id, dedupe_key)` rather
+  than application logic, which is what lets the overdue sweep run on every
+  replica with no leader election.
+- **Two checklist items with identical text assigned to the same person notify
+  once.** The notification's dedupe key is content-addressed precisely so that
+  reordering a checklist does not re-notify everyone; the cost is that identical
+  text collides. The assignee still sees both items on `/tasks`.
+- **Self-assignment is not suppressed when you assign yourself by typing.** The
+  guard that drops self-notifications needs to know who acted, but the path a
+  live edit takes does not carry that: WebSocket updates are persisted with no
+  user id, and the channel feeding the reindex worker carries only a document
+  id. Editing a checklist item that assigns you a task will notify you about it;
+  the guard does work on the import paths.
+
+### Fixed
+- **Members whose display name contains a space could not be mentioned when
+  opening a thread or replying.** Comment mentions were resolved server-side
+  by matching `@(\w+)` against member display names, so `@Christian Hüning`
+  captured `Christian`, matched nobody, and silently notified no one. The
+  mention picker now sends the user ids it resolved, unioned with whatever the
+  regex still matches so a hand-typed second name is never dropped; the regex
+  remains as the sole path for comments written before this release. Editing
+  an existing comment does not yet carry a `mentions` field, so adding
+  `@Christian Hüning` in an edit still notifies nobody — only the create path
+  is fixed here.
+
 ## [0.5.0] - 2026-09-06
 
 The editor moves from Tiptap 2 to Tiptap 3. The visible result should be nothing

crates/knot-obs/src/metrics.rs

@@ -48,6 +48,12 @@ pub fn init(addr: &str) -> Result<(), MetricsError> {
     );
     describe_counter!("knot_room_snapshots_total", "Snapshots written to storage");
 
+    // Notifications
+    describe_counter!(
+        "knot_notifications_emitted_total",
+        "Notifications written to the inbox, by kind"
+    );
+
     // Storage / pool
     describe_gauge!("knot_db_pool_size", "Total connections in the pool");
     describe_gauge!("knot_db_pool_idle", "Idle connections in the pool");

crates/knot-server/src/lib.rs

@@ -12,10 +12,10 @@ use knot_auth::{Hasher, Throttle};
 use knot_config::Config;
 use knot_docs::AclCache;
 use knot_storage::{
-    BlobMeta, BlobStore, CommentStore, DocStore, GrantStore, MarkdownCacheStore, PgBytesStore,
-    PgCommentStore, PgDocStore, PgGrantStore, PgMarkdownCache, PgSearchStore, PgSessionStore,
-    PgShareTokenStore, PgUserStore, PgWorkspaceStore, Pool, SearchStore, SessionStore,
-    ShareTokenStore, UserStore, WorkspaceStore,
+    BlobMeta, BlobStore, CommentStore, DocStore, GrantStore, MarkdownCacheStore, NotificationStore,
+    PgBytesStore, PgCommentStore, PgDocStore, PgGrantStore, PgMarkdownCache, PgNotificationStore,
+    PgSearchStore, PgSessionStore, PgShareTokenStore, PgUserStore, PgWorkspaceStore, Pool,
+    SearchStore, SessionStore, ShareTokenStore, UserStore, WorkspaceStore,
 };
 use tower_http::services::{ServeDir, ServeFile};
 use uuid::Uuid;
@@ -30,6 +30,7 @@ pub mod board_room_shim;
 pub mod comments_listener;
 pub mod http_error;
 pub mod metrics;
+pub mod notifications_sweep;
 pub mod protocol;
 pub mod reindex;
 pub mod room;
@@ -59,6 +60,7 @@ pub struct AppState {
     pub boards: Option<Arc<dyn knot_storage::BoardStore>>,
     pub board_rooms: Option<Arc<knot_crdt::BoardRooms>>,
     pub tasks: Option<Arc<dyn knot_storage::TaskStore>>,
+    pub notifications: Option<Arc<dyn NotificationStore>>,
     pub hasher: Arc<Hasher>,
     pub throttle: Arc<Throttle>,
     pub session_key: Vec<u8>,
@@ -95,6 +97,7 @@ impl AppState {
             boards: None,
             board_rooms: None,
             tasks: None,
+            notifications: None,
             hasher: Arc::new(Hasher::new()),
             throttle: Arc::new(Throttle::new()),
             session_key: Vec::new(),
@@ -136,6 +139,8 @@ impl AppState {
             Arc::new(knot_storage::PgBoardStore::new(pool.clone()));
         let tasks: Arc<dyn knot_storage::TaskStore> =
             Arc::new(knot_storage::PgTaskStore::new(pool.clone()));
+        let notifications: Arc<dyn NotificationStore> =
+            Arc::new(PgNotificationStore::new(pool.clone()));
         Self {
             pool: Some(pool),
             users: Some(users),
@@ -156,6 +161,7 @@ impl AppState {
             boards: Some(boards),
             board_rooms: None,
             tasks: Some(tasks),
+            notifications: Some(notifications),
             hasher: Arc::new(Hasher::new()),
             throttle: Arc::new(Throttle::new()),
             session_key: Vec::new(),

crates/knot-server/src/main.rs

@@ -274,6 +274,10 @@ async fn run_server(cfg: Config) {
         let _handle = knot_server::comments_listener::spawn(pool, rooms);
         tracing::info!("comments listener spawned");
     }
+    if let Some(pool) = state.pool.clone() {
+        let _handle = knot_server::notifications_sweep::spawn(pool);
+        tracing::info!("notification sweep spawned");
+    }
 
     // Token shared with every collab socket; cancelled on SIGTERM so they
     // send a clean 1001 Close and drain instead of being severed mid-rollout.

crates/knot-server/src/notifications_sweep.rs

@@ -0,0 +1,116 @@
+//! Periodic notification work: overdue tasks, and retention.
+//!
+//! Runs on every replica with no leader election. Both halves are safe to
+//! run concurrently — the overdue emit is `ON CONFLICT DO NOTHING` against
+//! `notifications_dedupe`, and the prune (delegated to
+//! `PgNotificationStore::prune`) is idempotent by construction.
+
+use chrono::{DateTime, Utc};
+use knot_storage::{NotificationStore, NotificationStoreError, PgNotificationStore};
+use sqlx::PgPool;
+use tokio::task::JoinHandle;
+
+const INTERVAL: std::time::Duration = std::time::Duration::from_secs(15 * 60);
+
+#[derive(Debug, Default, PartialEq, Eq)]
+pub struct SweepOutcome {
+    pub due_emitted: u64,
+    pub pruned: u64,
+}
+
+/// Emit `task_due` for every open, assigned, task that went overdue in the
+/// last 7 days, then prune.
+///
+/// The dedupe key is content-addressed exactly like `task_assigned`'s
+/// (`task_due:<doc_id>:<sha256(text)[..16]>:<assignee>:<date>`, computed with
+/// the same SQL sha256 expression `PgTaskStore` uses at
+/// `crates/knot-storage/src/tasks.rs`) rather than keyed on `doc_tasks.id`.
+/// `doc_tasks.id` is `"<doc_id>:<item_index>"`, so inserting a checklist
+/// item above an overdue one shifts every later item's id — an id-keyed
+/// dedupe key would treat that reorder as a brand-new task and re-notify
+/// every overdue assignee the same day. The date suffix still makes the key
+/// a pure function of `(task content, assignee, day)`, so an overdue task
+/// notifies once a day rather than once every fifteen minutes.
+///
+/// The `t.due_at > $1 - interval '7 days'` floor exists so that on an
+/// existing deployment — where `doc_tasks` already has rows with `due_at`
+/// stretching back to whenever the workspace was created, and
+/// `notifications` starts out empty — the first sweep after this feature
+/// ships does not treat every task that has *ever* been overdue as newly
+/// overdue and fire a `task_due` burst for the entire historical backlog.
+/// Only work that went overdue recently is worth a push; anything older
+/// than that is still visible on `/tasks` without a notification.
+pub async fn run_once(pool: &PgPool, now: DateTime<Utc>) -> Result<SweepOutcome, sqlx::Error> {
+    // The date half of the dedupe key is computed here, in Rust, from `now`
+    // — not with `to_char(...)` in SQL. `to_char` renders using the
+    // connection's session `TimeZone` GUC, which nothing in this repo ever
+    // sets; today every replica happens to inherit the same server default
+    // so the key is stable, but that's incidental, not enforced. Binding a
+    // pre-formatted date string makes the key a pure function of `now`,
+    // matching the design's claim that correctness lives entirely in the
+    // unique index, not in ambient session state.
+    let date = now.format("%Y-%m-%d").to_string();
+    let emitted = sqlx::query(
+        "INSERT INTO notifications \
+           (workspace_id, user_id, actor_id, kind, doc_id, target_kind, target_id, dedupe_key, data) \
+         SELECT t.workspace_id, t.assignee_user_id, NULL, 'task_due', t.doc_id, 'task', t.id, \
+                'task_due:' || t.doc_id || ':' || \
+                  encode(substring(sha256(convert_to(t.text, 'UTF8')) from 1 for 8), 'hex') || \
+                  ':' || t.assignee_user_id || ':' || $2, \
+                jsonb_build_object('excerpt', t.text, 'due_at', t.due_at) \
+         FROM doc_tasks t \
+         WHERE t.assignee_user_id IS NOT NULL \
+           AND t.checked = false \
+           AND t.due_at IS NOT NULL \
+           AND t.due_at < $1 \
+           AND t.due_at > $1 - interval '7 days' \
+         ON CONFLICT (user_id, dedupe_key) DO NOTHING",
+    )
+    .bind(now)
+    .bind(&date)
+    .execute(pool)
+    .await?
+    .rows_affected();
+
+    // Retention is `PgNotificationStore::prune`'s rule, not a second copy of
+    // it — the store already implements "read rows die after 90 days, all
+    // rows after 180" for the inbox-mutation endpoints, and duplicating
+    // that DELETE here would leave two places to keep in sync.
+    // `prune` returns the store's own error type; unwrap to the `sqlx::Error`
+    // this function's signature carries — `NotificationStoreError` only
+    // ever wraps one.
+    let pruned = match PgNotificationStore::new(pool.clone()).prune(now).await {
+        Ok(n) => n,
+        Err(NotificationStoreError::Sqlx(e)) => return Err(e),
+    };
+
+    if emitted > 0 {
+        metrics::counter!("knot_notifications_emitted_total", "kind" => "task_due")
+            .increment(emitted);
+    }
+    Ok(SweepOutcome {
+        due_emitted: emitted,
+        pruned,
+    })
+}
+
+/// Spawn the 15-minute loop. One per process; every replica runs its own.
+pub fn spawn(pool: PgPool) -> JoinHandle<()> {
+    tokio::spawn(async move {
+        let mut ticker = tokio::time::interval(INTERVAL);
+        // The first tick fires immediately; skip it so a rolling restart
+        // doesn't have every pod sweep at once on boot.
+        ticker.tick().await;
+        loop {
+            ticker.tick().await;
+            match run_once(&pool, Utc::now()).await {
+                Ok(out) => tracing::debug!(
+                    due_emitted = out.due_emitted,
+                    pruned = out.pruned,
+                    "notification sweep"
+                ),
+                Err(e) => tracing::warn!(error=?e, "notification sweep failed"),
+            }
+        }
+    })
+}

crates/knot-server/src/reindex.rs

@@ -52,7 +52,14 @@ pub fn spawn(state: AppState, mut rx: mpsc::Receiver<Uuid>) {
                     }
                     let to_flush = std::mem::take(&mut pending);
                     for doc_id in to_flush {
-                        if let Err(e) = refresh_markdown_and_index(&state, doc_id).await {
+                        // The dirty-notification channel carries only a
+                        // doc-id (see the module docs above), so there is
+                        // no editor identity to forward here. That means a
+                        // `task_assigned` notification triggered purely by
+                        // this worker's periodic flush has no actor — it
+                        // is never suppressed as a "self-assignment" and
+                        // never attributed to anyone in particular.
+                        if let Err(e) = refresh_markdown_and_index(&state, doc_id, None).await {
                             tracing::warn!(error=?e, %doc_id, "reindex worker: refresh failed");
                         }
                     }