hanna/fabrica

feat(store): notifications schema and queries

885e0b2 · hanna committed on 2026-07-26

Add the notifications table (migration 0020) and store layer: create
(self-directed events are skipped), list, unread count, and mark-read /
mark-all-read.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Signed by hannaSSH key fingerprint: SHA256:4g9cWhkAAw8gwqhJUcVbSnGEvdGwklW+1Aa/fMUu59k

crates/store/src/lib.rs +2 −0

crates/store/src/notifications.rs +165 −0

migrations/postgres/0020_notifications.sql +13 −0

migrations/sqlite/0020_notifications.sql +13 −0

 mod labels;
 mod lfs;
 mod mirrors;
+mod notifications;
 mod releases;
 mod repos;
 mod sessions;
 pub use crate::issues::StateCounts;
 pub use crate::keys::NewKey;
 pub use crate::mirrors::{Mirror, MirrorDirection, NewMirror};
+pub use crate::notifications::{NewNotification, Notification};
 pub use crate::releases::{NewRelease, Release, ReleaseAsset};
 pub use crate::repos::{Collaborator, NewRepo};
 pub use crate::sessions::NewSession;

+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at https://mozilla.org/MPL/2.0/.
+//! Notifications: an event (an @mention, an assignment) directed at a user. The
+//! row stores foreign keys only; the web layer resolves actor/repo/issue at
+//! display time.
+use sqlx::any::AnyRow;
+use sqlx::{AssertSqlSafe, Row};
+use crate::{Store, StoreError, new_id, now_ms};
+/// A pending notification.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Notification {
+    /// ULID primary key.
+    pub id: String,
+    /// The recipient.
+    pub user_id: String,
+    /// Who triggered it, if still present.
+    pub actor_id: Option<String>,
+    /// The kind (`mention`, `assign`).
+    pub kind: String,
+    /// The repo it concerns, if any.
+    pub repo_id: Option<String>,
+    /// The issue/PR it concerns, if any.
+    pub issue_id: Option<String>,
+    /// When it was read, if it has been.
+    pub read_at: Option<i64>,
+    /// Creation time.
+    pub created_at: i64,
+}
+/// Fields to create a notification.
+#[derive(Debug, Clone)]
+pub struct NewNotification {
+    /// The recipient.
+    pub user_id: String,
+    /// Who triggered it.
+    pub actor_id: String,
+    /// The kind.
+    pub kind: String,
+    /// The repo it concerns.
+    pub repo_id: Option<String>,
+    /// The issue/PR it concerns.
+    pub issue_id: Option<String>,
+}
+const NOTIF_COLUMNS: &str = "id, user_id, actor_id, kind, repo_id, issue_id, read_at, created_at";
+impl Store {
+    fn map_notification(row: &AnyRow) -> Result<Notification, sqlx::Error> {
+        Ok(Notification {
+            id: row.try_get("id")?,
+            user_id: row.try_get("user_id")?,
+            actor_id: row.try_get("actor_id")?,
+            kind: row.try_get("kind")?,
+            repo_id: row.try_get("repo_id")?,
+            issue_id: row.try_get("issue_id")?,
+            read_at: row.try_get("read_at")?,
+            created_at: row.try_get("created_at")?,
+        })
+    }
+    /// Create a notification. A self-directed notification (actor == recipient)
+    /// is skipped — you are not notified of your own actions.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`StoreError::Query`] on a database failure.
+    pub async fn create_notification(&self, new: NewNotification) -> Result<(), StoreError> {
+        if new.user_id == new.actor_id {
+            return Ok(());
+        }
+        sqlx::query(
+            "INSERT INTO notifications \
+             (id, user_id, actor_id, kind, repo_id, issue_id, read_at, created_at) \
+             VALUES ($1, $2, $3, $4, $5, $6, $7, $8)",
+        )
+        .bind(new_id())
+        .bind(&new.user_id)
+        .bind(&new.actor_id)
+        .bind(&new.kind)
+        .bind(&new.repo_id)
+        .bind(&new.issue_id)
+        .bind(Option::<i64>::None)
+        .bind(now_ms())
+        .execute(&self.pool)
+        .await?;
+        Ok(())
+    }
+    /// A user's notifications, newest first, capped at `limit`.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`StoreError::Query`] on a database failure.
+    pub async fn list_notifications(
+        &self,
+        user_id: &str,
+        limit: i64,
+    ) -> Result<Vec<Notification>, StoreError> {
+        let sql = format!(
+            "SELECT {NOTIF_COLUMNS} FROM notifications WHERE user_id = $1 \
+             ORDER BY created_at DESC LIMIT $2"
+        );
+        let rows = sqlx::query(AssertSqlSafe(sql))
+            .bind(user_id)
+            .bind(limit)
+            .fetch_all(&self.pool)
+            .await?;
+        rows.iter()
+            .map(Self::map_notification)
+            .collect::<Result<_, _>>()
+            .map_err(Into::into)
+    }
+    /// The number of unread notifications for a user.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`StoreError::Query`] on a database failure.
+    pub async fn count_unread_notifications(&self, user_id: &str) -> Result<i64, StoreError> {
+        let n: i64 = sqlx::query(
+            "SELECT COUNT(*) AS n FROM notifications WHERE user_id = $1 AND read_at IS NULL",
+        )
+        .bind(user_id)
+        .fetch_one(&self.pool)
+        .await?
+        .try_get("n")?;
+        Ok(n)
+    }
+    /// Mark one notification read (scoped to its owner).
+    ///
+    /// # Errors
+    ///
+    /// Returns [`StoreError::Query`] on a database failure.
+    pub async fn mark_notification_read(&self, id: &str, user_id: &str) -> Result<(), StoreError> {
+        sqlx::query(
+            "UPDATE notifications SET read_at = $1 WHERE id = $2 AND user_id = $3 AND read_at IS NULL",
+        )
+        .bind(now_ms())
+        .bind(id)
+        .bind(user_id)
+        .execute(&self.pool)
+        .await?;
+        Ok(())
+    }
+    /// Mark all of a user's notifications read.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`StoreError::Query`] on a database failure.
+    pub async fn mark_all_notifications_read(&self, user_id: &str) -> Result<(), StoreError> {
+        sqlx::query("UPDATE notifications SET read_at = $1 WHERE user_id = $2 AND read_at IS NULL")
+            .bind(now_ms())
+            .bind(user_id)
+            .execute(&self.pool)
+            .await?;
+        Ok(())
+    }
+}

+-- Notifications: an actor did something (mention, assignment) that a user should
+-- see. Rendered by resolving the linked actor/repo/issue at display time.
+CREATE TABLE notifications (
+    id         TEXT PRIMARY KEY,
+    user_id    TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+    actor_id   TEXT REFERENCES users(id) ON DELETE SET NULL,
+    kind       TEXT NOT NULL,
+    repo_id    TEXT REFERENCES repos(id) ON DELETE CASCADE,
+    issue_id   TEXT REFERENCES issues(id) ON DELETE CASCADE,
+    read_at    BIGINT,
+    created_at BIGINT NOT NULL
+);
+CREATE INDEX idx_notifications_user ON notifications (user_id, read_at);

+-- Notifications: an actor did something (mention, assignment) that a user should
+-- see. Rendered by resolving the linked actor/repo/issue at display time.
+CREATE TABLE notifications (
+    id         TEXT PRIMARY KEY,
+    user_id    TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+    actor_id   TEXT REFERENCES users(id) ON DELETE SET NULL,
+    kind       TEXT NOT NULL,
+    repo_id    TEXT REFERENCES repos(id) ON DELETE CASCADE,
+    issue_id   TEXT REFERENCES issues(id) ON DELETE CASCADE,
+    read_at    BIGINT,
+    created_at BIGINT NOT NULL
+);
+CREATE INDEX idx_notifications_user ON notifications (user_id, read_at);