Start work on storing notifications in the database

crates/collab/migrations.sqlite/20221109000000_test_schema.sql

@@ -312,3 +312,22 @@ CREATE TABLE IF NOT EXISTS "observed_channel_messages" (
 );
 
 CREATE UNIQUE INDEX "index_observed_channel_messages_user_and_channel_id" ON "observed_channel_messages" ("user_id", "channel_id");
+
+CREATE TABLE "notification_kinds" (
+    "id" INTEGER PRIMARY KEY NOT NULL,
+    "name" VARCHAR NOT NULL,
+);
+
+CREATE UNIQUE INDEX "index_notification_kinds_on_name" ON "notification_kinds" ("name");
+
+CREATE TABLE "notifications" (
+    "id" INTEGER PRIMARY KEY AUTOINCREMENT,
+    "created_at" TIMESTAMP NOT NULL default now,
+    "recipent_id" INTEGER NOT NULL REFERENCES users (id) ON DELETE CASCADE,
+    "kind" INTEGER NOT NULL REFERENCES notification_kinds (id),
+    "is_read" BOOLEAN NOT NULL DEFAULT FALSE,
+    "entity_id_1" INTEGER,
+    "entity_id_2" INTEGER
+);
+
+CREATE INDEX "index_notifications_on_recipient_id" ON "notifications" ("recipient_id");

crates/collab/migrations/20231004130100_create_notifications.sql

@@ -0,0 +1,18 @@
+CREATE TABLE "notification_kinds" (
+    "id" INTEGER PRIMARY KEY NOT NULL,
+    "name" VARCHAR NOT NULL,
+);
+
+CREATE UNIQUE INDEX "index_notification_kinds_on_name" ON "notification_kinds" ("name");
+
+CREATE TABLE notifications (
+    "id" SERIAL PRIMARY KEY,
+    "created_at" TIMESTAMP NOT NULL DEFAULT now(),
+    "recipent_id" INTEGER NOT NULL REFERENCES users (id) ON DELETE CASCADE,
+    "kind" INTEGER NOT NULL REFERENCES notification_kinds (id),
+    "is_read" BOOLEAN NOT NULL DEFAULT FALSE
+    "entity_id_1" INTEGER,
+    "entity_id_2" INTEGER
+);
+
+CREATE INDEX "index_notifications_on_recipient_id" ON "notifications" ("recipient_id");

crates/collab/src/db.rs

@@ -20,7 +20,7 @@ use rpc::{
 };
 use sea_orm::{
     entity::prelude::*,
-    sea_query::{Alias, Expr, OnConflict, Query},
+    sea_query::{Alias, Expr, OnConflict},
     ActiveValue, Condition, ConnectionTrait, DatabaseConnection, DatabaseTransaction, DbErr,
     FromQueryResult, IntoActiveModel, IsolationLevel, JoinType, QueryOrder, QuerySelect, Statement,
     TransactionTrait,

crates/collab/src/db/ids.rs

@@ -80,3 +80,4 @@ id_type!(SignupId);
 id_type!(UserId);
 id_type!(ChannelBufferCollaboratorId);
 id_type!(FlagId);
+id_type!(NotificationId);

crates/collab/src/db/queries.rs

@@ -5,6 +5,7 @@ pub mod buffers;
 pub mod channels;
 pub mod contacts;
 pub mod messages;
+pub mod notifications;
 pub mod projects;
 pub mod rooms;
 pub mod servers;

crates/collab/src/db/queries/access_tokens.rs

@@ -1,4 +1,5 @@
 use super::*;
+use sea_orm::sea_query::Query;
 
 impl Database {
     pub async fn create_access_token(

crates/collab/src/db/queries/notifications.rs

@@ -0,0 +1,140 @@
+use super::*;
+use rpc::{Notification, NotificationEntityKind, NotificationKind};
+
+impl Database {
+    pub async fn ensure_notification_kinds(&self) -> Result<()> {
+        self.transaction(|tx| async move {
+            notification_kind::Entity::insert_many(NotificationKind::all().map(|kind| {
+                notification_kind::ActiveModel {
+                    id: ActiveValue::Set(kind as i32),
+                    name: ActiveValue::Set(kind.to_string()),
+                }
+            }))
+            .on_conflict(OnConflict::new().do_nothing().to_owned())
+            .exec(&*tx)
+            .await?;
+            Ok(())
+        })
+        .await
+    }
+
+    pub async fn get_notifications(
+        &self,
+        recipient_id: UserId,
+        limit: usize,
+    ) -> Result<proto::AddNotifications> {
+        self.transaction(|tx| async move {
+            let mut result = proto::AddNotifications::default();
+
+            let mut rows = notification::Entity::find()
+                .filter(notification::Column::RecipientId.eq(recipient_id))
+                .order_by_desc(notification::Column::Id)
+                .limit(limit as u64)
+                .stream(&*tx)
+                .await?;
+
+            let mut user_ids = Vec::new();
+            let mut channel_ids = Vec::new();
+            let mut message_ids = Vec::new();
+            while let Some(row) = rows.next().await {
+                let row = row?;
+
+                let Some(kind) = NotificationKind::from_i32(row.kind) else {
+                    continue;
+                };
+                let Some(notification) = Notification::from_fields(
+                    kind,
+                    [
+                        row.entity_id_1.map(|id| id as u64),
+                        row.entity_id_2.map(|id| id as u64),
+                        row.entity_id_3.map(|id| id as u64),
+                    ],
+                ) else {
+                    continue;
+                };
+
+                // Gather the ids of all associated entities.
+                let (_, associated_entities) = notification.to_fields();
+                for entity in associated_entities {
+                    let Some((id, kind)) = entity else {
+                        break;
+                    };
+                    match kind {
+                        NotificationEntityKind::User => &mut user_ids,
+                        NotificationEntityKind::Channel => &mut channel_ids,
+                        NotificationEntityKind::ChannelMessage => &mut message_ids,
+                    }
+                    .push(id);
+                }
+
+                result.notifications.push(proto::Notification {
+                    kind: row.kind as u32,
+                    timestamp: row.created_at.assume_utc().unix_timestamp() as u64,
+                    is_read: row.is_read,
+                    entity_id_1: row.entity_id_1.map(|id| id as u64),
+                    entity_id_2: row.entity_id_2.map(|id| id as u64),
+                    entity_id_3: row.entity_id_3.map(|id| id as u64),
+                });
+            }
+
+            let users = user::Entity::find()
+                .filter(user::Column::Id.is_in(user_ids))
+                .all(&*tx)
+                .await?;
+            let channels = channel::Entity::find()
+                .filter(user::Column::Id.is_in(channel_ids))
+                .all(&*tx)
+                .await?;
+            let messages = channel_message::Entity::find()
+                .filter(user::Column::Id.is_in(message_ids))
+                .all(&*tx)
+                .await?;
+
+            for user in users {
+                result.users.push(proto::User {
+                    id: user.id.to_proto(),
+                    github_login: user.github_login,
+                    avatar_url: String::new(),
+                });
+            }
+            for channel in channels {
+                result.channels.push(proto::Channel {
+                    id: channel.id.to_proto(),
+                    name: channel.name,
+                });
+            }
+            for message in messages {
+                result.messages.push(proto::ChannelMessage {
+                    id: message.id.to_proto(),
+                    body: message.body,
+                    timestamp: message.sent_at.assume_utc().unix_timestamp() as u64,
+                    sender_id: message.sender_id.to_proto(),
+                    nonce: None,
+                });
+            }
+
+            Ok(result)
+        })
+        .await
+    }
+
+    pub async fn create_notification(
+        &self,
+        recipient_id: UserId,
+        notification: Notification,
+        tx: &DatabaseTransaction,
+    ) -> Result<()> {
+        let (kind, associated_entities) = notification.to_fields();
+        notification::ActiveModel {
+            recipient_id: ActiveValue::Set(recipient_id),
+            kind: ActiveValue::Set(kind as i32),
+            entity_id_1: ActiveValue::Set(associated_entities[0].map(|(id, _)| id as i32)),
+            entity_id_2: ActiveValue::Set(associated_entities[1].map(|(id, _)| id as i32)),
+            entity_id_3: ActiveValue::Set(associated_entities[2].map(|(id, _)| id as i32)),
+            ..Default::default()
+        }
+        .save(&*tx)
+        .await?;
+        Ok(())
+    }
+}

crates/collab/src/db/tables.rs

@@ -12,6 +12,8 @@ pub mod contact;
 pub mod feature_flag;
 pub mod follower;
 pub mod language_server;
+pub mod notification;
+pub mod notification_kind;
 pub mod observed_buffer_edits;
 pub mod observed_channel_messages;
 pub mod project;

crates/collab/src/db/tables/notification.rs

@@ -0,0 +1,29 @@
+use crate::db::{NotificationId, UserId};
+use sea_orm::entity::prelude::*;
+use time::PrimitiveDateTime;
+
+#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)]
+#[sea_orm(table_name = "notifications")]
+pub struct Model {
+    #[sea_orm(primary_key)]
+    pub id: NotificationId,
+    pub recipient_id: UserId,
+    pub kind: i32,
+    pub is_read: bool,
+    pub created_at: PrimitiveDateTime,
+    pub entity_id_1: Option<i32>,
+    pub entity_id_2: Option<i32>,
+    pub entity_id_3: Option<i32>,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {
+    #[sea_orm(
+        belongs_to = "super::user::Entity",
+        from = "Column::RecipientId",
+        to = "super::user::Column::Id"
+    )]
+    Recipient,
+}
+
+impl ActiveModelBehavior for ActiveModel {}

crates/collab/src/db/tables/notification_kind.rs

@@ -0,0 +1,14 @@
+use sea_orm::entity::prelude::*;
+
+#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)]
+#[sea_orm(table_name = "notification_kinds")]
+pub struct Model {
+    #[sea_orm(primary_key)]
+    pub id: i32,
+    pub name: String,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {}
+
+impl ActiveModelBehavior for ActiveModel {}