Merge branch 'notifications' into 'main'

Draft: support for `/client/v3/notifications` endpoint

See merge request matrix/grapevine!180

book/changelog.md

@@ -324,3 +324,5 @@ This will be the first release of Grapevine since it was forked from Conduit
     ([!124](https://gitlab.computer.surgery/matrix/grapevine/-/merge_requests/124))
 26. Allow adding canonical aliases from remote servers.
     ([!158](https://gitlab.computer.surgery/matrix/grapevine/-/merge_requests/158))
+27. Added implementation for the notification endpoint.
+    ([!180](https://gitlab.computer.surgery/matrix/grapevine/-/merge_requests/180))

src/api/client_server.rs

@@ -11,6 +11,7 @@ mod keys;
 mod media;
 mod membership;
 mod message;
+mod notifications;
 mod profile;
 mod push;
 mod read_marker;
@@ -45,6 +46,7 @@ pub(crate) use keys::*;
 pub(crate) use media::*;
 pub(crate) use membership::*;
 pub(crate) use message::*;
+pub(crate) use notifications::*;
 pub(crate) use profile::*;
 pub(crate) use push::*;
 pub(crate) use read_marker::*;

src/api/client_server/notifications.rs

@@ -0,0 +1,43 @@
+use ruma::api::client::{error::ErrorKind, push::get_notifications};
+
+use crate::{services, Ar, Error, Ra, Result};
+
+/// # `GET /_matrix/client/r0/notifications`
+///
+/// Gets all events that we have been notified about.
+#[allow(clippy::unused_async)]
+pub(crate) async fn get_notifications_route(
+    body: Ar<get_notifications::v3::Request>,
+) -> Result<Ra<get_notifications::v3::Response>> {
+    let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+
+    let from = body
+        .from
+        .as_ref()
+        .map(|from| {
+            from.parse().map_err(|_| {
+                Error::BadRequest(
+                    ErrorKind::InvalidParam,
+                    "failed to parse from ",
+                )
+            })
+        })
+        .transpose()?;
+
+    let highlight = body
+        .only
+        .as_ref()
+        .is_some_and(|only| only.to_lowercase() == "highlight");
+
+    let (notifications, next_token) = services().rooms.user.get_notifications(
+        sender_user,
+        from,
+        body.limit,
+        highlight,
+    )?;
+
+    Ok(Ra(get_notifications::v3::Response {
+        next_token,
+        notifications,
+    }))
+}

src/cli/serve.rs

@@ -488,6 +488,7 @@ fn client_routes() -> Router {
         .ruma_route(c2s::get_pushrule_actions_route)
         .ruma_route(c2s::set_pushrule_actions_route)
         .ruma_route(c2s::delete_pushrule_route)
+        .ruma_route(c2s::get_notifications_route)
         .ruma_route(c2s::get_room_event_route)
         .ruma_route(c2s::get_room_aliases_route)
         .ruma_route(c2s::get_filter_route)

src/database.rs

@@ -140,15 +140,14 @@ pub(crate) struct KeyValueDatabase {
     // LazyLoadedIds = UserId + DeviceId + RoomId + LazyLoadedUserId
     pub(super) lazyloadedids: Arc<dyn KvTree>,
 
-    // NotifyCount = u64
-    pub(super) userroomid_notificationcount: Arc<dyn KvTree>,
-
-    // HightlightCount = u64
-    pub(super) userroomid_highlightcount: Arc<dyn KvTree>,
-
     // LastNotificationRead = u64
     pub(super) roomuserid_lastnotificationread: Arc<dyn KvTree>,
 
+    // UserPduCountId = UserId + PduCount
+    pub(super) userpducountid_notification: Arc<dyn KvTree>,
+    // UserRoomPduCountId = UserId + RoomId + PduCount
+    pub(super) userroompducountid: Arc<dyn KvTree>,
+
     /// Remember the current state hash of a room.
     pub(super) roomid_shortstatehash: Arc<dyn KvTree>,
 
@@ -392,12 +391,11 @@ impl KeyValueDatabase {
 
             lazyloadedids: builder.open_tree("lazyloadedids")?,
 
-            userroomid_notificationcount: builder
-                .open_tree("userroomid_notificationcount")?,
-            userroomid_highlightcount: builder
-                .open_tree("userroomid_highlightcount")?,
             roomuserid_lastnotificationread: builder
-                .open_tree("userroomid_highlightcount")?,
+                .open_tree("roomuserid_lastnotificationread")?,
+            userpducountid_notification: builder
+                .open_tree("userpducountid_notifications")?,
+            userroompducountid: builder.open_tree("userroompducountid")?,
 
             statekey_shortstatekey: builder
                 .open_tree("statekey_shortstatekey")?,

src/database/abstraction.rs

@@ -42,6 +42,7 @@ pub(crate) trait KvTree: Send + Sync {
     ) -> Box<dyn Iterator<Item = (Vec<u8>, Vec<u8>)> + 'a>;
 
     fn increment(&self, key: &[u8]) -> Result<Vec<u8>>;
+    #[allow(dead_code)]
     fn increment_batch(
         &self,
         iter: &mut dyn Iterator<Item = Vec<u8>>,

src/database/key_value/globals.rs

@@ -51,11 +51,16 @@ impl service::globals::Data for KeyValueDatabase {
         futures.push(self.userroomid_joined.watch_prefix(&userid_prefix));
         futures.push(self.userroomid_invitestate.watch_prefix(&userid_prefix));
         futures.push(self.userroomid_leftstate.watch_prefix(&userid_prefix));
+
+        let localpart_bytes = user_id.localpart().as_bytes().to_vec();
+        let mut localpart_prefix = localpart_bytes.clone();
+        localpart_prefix.push(0xFF);
+
+        // Return when we store a notification
         futures.push(
-            self.userroomid_notificationcount.watch_prefix(&userid_prefix),
+            self.userpducountid_notification.watch_prefix(&localpart_prefix),
         );
-        futures
+        futures.push(self.userroompducountid.watch_prefix(&localpart_prefix));
-            .push(self.userroomid_highlightcount.watch_prefix(&userid_prefix));
 
         // Events for rooms we are in
         for room_id in services()

src/database/key_value/rooms/timeline.rs

@@ -1,8 +1,7 @@
 use std::{mem::size_of, sync::Arc};
 
 use ruma::{
-    api::client::error::ErrorKind, CanonicalJsonObject, EventId, OwnedUserId,
+    api::client::error::ErrorKind, CanonicalJsonObject, EventId, RoomId, UserId,
-    RoomId, UserId,
 };
 use service::rooms::timeline::PduCount;
 
@@ -255,34 +254,6 @@ impl service::rooms::timeline::Data for KeyValueDatabase {
                 }),
         ))
     }
-
-    fn increment_notification_counts(
-        &self,
-        room_id: &RoomId,
-        notifies: Vec<OwnedUserId>,
-        highlights: Vec<OwnedUserId>,
-    ) -> Result<()> {
-        let mut notifies_batch = Vec::new();
-        let mut highlights_batch = Vec::new();
-        for user in notifies {
-            let mut userroom_id = user.as_bytes().to_vec();
-            userroom_id.push(0xFF);
-            userroom_id.extend_from_slice(room_id.as_bytes());
-            notifies_batch.push(userroom_id);
-        }
-        for user in highlights {
-            let mut userroom_id = user.as_bytes().to_vec();
-            userroom_id.push(0xFF);
-            userroom_id.extend_from_slice(room_id.as_bytes());
-            highlights_batch.push(userroom_id);
-        }
-
-        self.userroomid_notificationcount
-            .increment_batch(&mut notifies_batch.into_iter())?;
-        self.userroomid_highlightcount
-            .increment_batch(&mut highlights_batch.into_iter())?;
-        Ok(())
-    }
 }
 
 /// Returns the `count` of this pdu's id.

src/database/key_value/rooms/user.rs

@@ -1,9 +1,19 @@
-use ruma::{OwnedRoomId, OwnedUserId, RoomId, UserId};
+use ruma::{
+    api::client::{
+        error::ErrorKind, push::get_notifications::v3::Notification,
+    },
+    push::Action,
+    EventId, MilliSecondsSinceUnixEpoch, OwnedRoomId, OwnedUserId, RoomId,
+    UInt, UserId,
+};
 
 use crate::{
     database::KeyValueDatabase,
-    service::{self, rooms::short::ShortStateHash},
+    service::{
-    services, utils, Error, Result,
+        self,
+        rooms::{short::ShortStateHash, timeline::PduCount},
+    },
+    services, utils, Error, PduEvent, Result,
 };
 
 impl service::rooms::user::Data for KeyValueDatabase {
@@ -12,22 +22,14 @@ impl service::rooms::user::Data for KeyValueDatabase {
         user_id: &UserId,
         room_id: &RoomId,
     ) -> Result<()> {
-        let mut userroom_id = user_id.as_bytes().to_vec();
-        userroom_id.push(0xFF);
-        userroom_id.extend_from_slice(room_id.as_bytes());
         let mut roomuser_id = room_id.as_bytes().to_vec();
         roomuser_id.push(0xFF);
         roomuser_id.extend_from_slice(user_id.as_bytes());
 
-        self.userroomid_notificationcount
+        let next_count = services().globals.next_count()?;
-            .insert(&userroom_id, &0_u64.to_be_bytes())?;
-        self.userroomid_highlightcount
-            .insert(&userroom_id, &0_u64.to_be_bytes())?;
 
-        self.roomuserid_lastnotificationread.insert(
+        self.roomuserid_lastnotificationread
-            &roomuser_id,
+            .insert(&roomuser_id, &next_count.to_be_bytes())?;
-            &services().globals.next_count()?.to_be_bytes(),
-        )?;
 
         Ok(())
     }
@@ -37,18 +39,34 @@ impl service::rooms::user::Data for KeyValueDatabase {
         user_id: &UserId,
         room_id: &RoomId,
     ) -> Result<u64> {
-        let mut userroom_id = user_id.as_bytes().to_vec();
+        let pdu_count = self.last_notification_read(user_id, room_id)?;
-        userroom_id.push(0xFF);
-        userroom_id.extend_from_slice(room_id.as_bytes());
 
-        self.userroomid_notificationcount.get(&userroom_id)?.map_or(
+        let mut key = user_id.localpart().as_bytes().to_vec();
-            Ok(0),
+        key.push(0xFF);
-            |bytes| {
+
-                utils::u64_from_bytes(&bytes).map_err(|_| {
+        key.extend_from_slice(&pdu_count.to_be_bytes());
-                    Error::bad_database("Invalid notification count in db.")
+
-                })
+        let mut it = self.userpducountid_notification.iter_from(&key, false);
-            },
+
-        )
+        it.try_fold(0, |mut n, (_, value)| {
+            let (flags, value) = value.split_at(2);
+
+            let event_id = EventId::parse(
+                utils::string_from_bytes(
+                    value.split(|b| *b == 0xFF).nth(1).unwrap(),
+                )
+                .unwrap(),
+            )
+            .unwrap();
+
+            let pdu = services().rooms.timeline.get_pdu(&event_id)?;
+
+            if flags[0] != 0 && !pdu.map_or(true, |pdu| !pdu.is_redacted()) {
+                n += 1;
+            }
+
+            Ok(n)
+        })
     }
 
     fn highlight_count(
@@ -56,18 +74,34 @@ impl service::rooms::user::Data for KeyValueDatabase {
         user_id: &UserId,
         room_id: &RoomId,
     ) -> Result<u64> {
-        let mut userroom_id = user_id.as_bytes().to_vec();
+        let pdu_count = self.last_notification_read(user_id, room_id)?;
-        userroom_id.push(0xFF);
-        userroom_id.extend_from_slice(room_id.as_bytes());
 
-        self.userroomid_highlightcount.get(&userroom_id)?.map_or(
+        let mut key = user_id.localpart().as_bytes().to_vec();
-            Ok(0),
+        key.push(0xFF);
-            |bytes| {
+
-                utils::u64_from_bytes(&bytes).map_err(|_| {
+        key.extend_from_slice(&pdu_count.to_be_bytes());
-                    Error::bad_database("Invalid highlight count in db.")
+
-                })
+        let mut it = self.userpducountid_notification.iter_from(&key, false);
-            },
+
-        )
+        it.try_fold(0, |mut n, (_, value)| {
+            let (flags, value) = value.split_at(2);
+
+            let event_id = EventId::parse(
+                utils::string_from_bytes(
+                    value.split(|b| *b == 0xFF).nth(1).unwrap(),
+                )
+                .unwrap(),
+            )
+            .unwrap();
+
+            let pdu = services().rooms.timeline.get_pdu(&event_id)?;
+
+            if flags[1] != 0 && !pdu.map_or(true, |pdu| !pdu.is_redacted()) {
+                n += 1;
+            }
+
+            Ok(n)
+        })
     }
 
     fn last_notification_read(
@@ -93,6 +127,144 @@ impl service::rooms::user::Data for KeyValueDatabase {
             .unwrap_or(0))
     }
 
+    fn store_push_action(
+        &self,
+        pdu: &PduEvent,
+        user_id: &UserId,
+        notify: bool,
+        highlight: bool,
+        actions: &[Action],
+    ) -> Result<()> {
+        let Some(PduCount::Normal(pdu_count)) =
+            services().rooms.timeline.get_pdu_count(&pdu.event_id)?
+        else {
+            return Err(Error::BadRequest(
+                ErrorKind::InvalidParam,
+                "Event does not exist.",
+            ));
+        };
+
+        let mut key = user_id.localpart().as_bytes().to_vec();
+        key.push(0xFF);
+
+        key.extend_from_slice(&pdu_count.to_be_bytes());
+
+        let (notify, is_highlight) = (u8::from(notify), u8::from(highlight));
+
+        let mut value = vec![notify, is_highlight];
+
+        value.extend_from_slice(pdu.room_id.as_bytes());
+        value.push(0xFF);
+
+        value.extend_from_slice(pdu.event_id.as_bytes());
+        value.push(0xFF);
+
+        // TODO: timestamp at which the event notification was sent
+        let ts = serde_json::to_vec(&MilliSecondsSinceUnixEpoch::now())
+            .expect("ts should serialize");
+
+        value.extend_from_slice(&ts);
+        value.push(0xFF);
+
+        let actions =
+            serde_json::to_vec(actions).expect("actions should serialize");
+
+        value.extend_from_slice(&actions);
+
+        self.userpducountid_notification.insert(&key, &value)?;
+
+        let mut key = user_id.localpart().as_bytes().to_vec();
+        key.push(0xFF);
+
+        key.extend_from_slice(pdu.room_id.as_bytes());
+
+        key.extend_from_slice(&pdu_count.to_be_bytes());
+
+        self.userroompducountid.insert(&key, &[])
+    }
+
+    fn get_notifications(
+        &self,
+        user_id: &UserId,
+        from: Option<u64>,
+        limit: Option<UInt>,
+        highlight: bool,
+    ) -> Result<(Vec<Notification>, Option<String>)> {
+        let mut key = user_id.localpart().as_bytes().to_vec();
+
+        key.push(0xFF);
+        key.extend_from_slice(&from.unwrap_or(u64::MAX).to_be_bytes());
+
+        let limit = limit.and_then(|n| n.try_into().ok()).unwrap_or(50);
+
+        let it = self.userpducountid_notification.iter_from(&key, true);
+
+        let mut notifications = Vec::new();
+
+        let mut next_token = None;
+
+        for (key, value) in it {
+            let pdu_count = utils::u64_from_bytes(
+                key.split(|b| *b == 0xFF).nth(1).unwrap(),
+            )
+            .unwrap();
+
+            if notifications.len() == limit {
+                next_token = Some(format!("{pdu_count}"));
+
+                break;
+            }
+
+            let (flags, value) = value.split_at(2);
+
+            let (notify, is_highlight) = (flags[0] != 0, flags[1] != 0);
+
+            if !notify || highlight && !is_highlight {
+                continue;
+            }
+
+            let mut it = value.split(|b| *b == 0xFF);
+
+            let room_id = RoomId::parse(
+                utils::string_from_bytes(it.next().unwrap()).unwrap(),
+            )
+            .unwrap();
+
+            let event_id = EventId::parse(
+                utils::string_from_bytes(it.next().unwrap()).unwrap(),
+            )
+            .unwrap();
+
+            let pdu = services().rooms.timeline.get_pdu(&event_id)?;
+
+            // do not bother with missing or redacted PDUs
+            let Some(pdu) = pdu.filter(|pdu| !pdu.is_redacted()) else {
+                continue;
+            };
+
+            let ts: MilliSecondsSinceUnixEpoch =
+                serde_json::from_slice(it.next().unwrap()).unwrap();
+
+            let actions: Vec<Action> =
+                serde_json::from_slice(it.next().unwrap()).unwrap();
+
+            notifications.push((
+                key,
+                Notification {
+                    actions,
+                    event: pdu.to_sync_room_event(),
+                    profile_tag: None,
+                    read: pdu_count
+                        < self.last_notification_read(user_id, &room_id)?,
+                    room_id,
+                    ts,
+                },
+            ));
+        }
+
+        Ok((notifications.into_iter().map(|(_, n)| n).collect(), next_token))
+    }
+
     fn associate_token_shortstatehash(
         &self,
         room_id: &RoomId,

src/service/pusher.rs

@@ -177,13 +177,15 @@ impl Service {
             .transpose()?
             .unwrap_or_default();
 
-        for action in self.get_actions(
+        let actions = Service::get_actions(
             user,
             &ruleset,
             &power_levels,
             &pdu.to_sync_room_event(),
             &pdu.room_id,
-        )? {
+        )?;
+
+        for action in actions {
             let n = match action {
                 Action::Notify => true,
                 Action::SetTweak(tweak) => {
@@ -210,9 +212,8 @@ impl Service {
         Ok(())
     }
 
-    #[tracing::instrument(skip(self, user, ruleset, pdu))]
+    #[tracing::instrument(skip(user, ruleset, pdu))]
     pub(crate) fn get_actions<'a>(
-        &self,
         user: &UserId,
         ruleset: &'a Ruleset,
         power_levels: &RoomPowerLevelsEventContent,
@@ -311,12 +312,6 @@ impl Service {
                         .rooms
                         .state_accessor
                         .get_name(&event.room_id)?;
-
-                    self.send_request(
-                        &http.url,
-                        send_event_notification::v1::Request::new(notifi),
-                    )
-                    .await?;
                 }
 
                 Ok(())

src/service/rooms/timeline.rs

@@ -36,6 +36,7 @@ use crate::{
         appservice::NamespaceRegex,
         globals::{marker, SigningKeys},
         pdu::{EventHash, PduBuilder},
+        pusher,
         rooms::state::ExtractVersion,
     },
     services,
@@ -440,13 +441,15 @@ impl Service {
             let mut highlight = false;
             let mut notify = false;
 
-            for action in services().pusher.get_actions(
+            let actions = pusher::Service::get_actions(
                 user,
                 &rules_for_user,
                 &power_levels,
                 &sync_pdu,
                 &pdu.room_id,
-            )? {
+            )?;
+
+            for action in actions {
                 match action {
                     Action::Notify => notify = true,
                     Action::SetTweak(Tweak::Highlight(true)) => {
@@ -464,17 +467,19 @@ impl Service {
                 highlights.push(user.clone());
             }
 
+            if notify || highlight {
+                // Store push action
+                services()
+                    .rooms
+                    .user
+                    .store_push_action(pdu, user, notify, highlight, actions)?;
+            }
+
             for push_key in services().pusher.get_pushkeys(user) {
                 services().sending.send_push_pdu(&pdu_id, user, push_key?)?;
             }
         }
 
-        self.db.increment_notification_counts(
-            &pdu.room_id,
-            notifies,
-            highlights,
-        )?;
-
         match pdu.kind {
             TimelineEventType::RoomRedaction => {
                 let room_version_id = services()

src/service/rooms/timeline/data.rs

@@ -1,6 +1,6 @@
 use std::sync::Arc;
 
-use ruma::{CanonicalJsonObject, EventId, OwnedUserId, RoomId, UserId};
+use ruma::{CanonicalJsonObject, EventId, RoomId, UserId};
 
 use super::PduCount;
 use crate::{service::rooms::timeline::PduId, PduEvent, Result};
@@ -91,11 +91,4 @@ pub(crate) trait Data: Send + Sync {
         room_id: &RoomId,
         from: PduCount,
     ) -> Result<Box<dyn Iterator<Item = Result<(PduCount, PduEvent)>> + 'a>>;
-
-    fn increment_notification_counts(
-        &self,
-        room_id: &RoomId,
-        notifies: Vec<OwnedUserId>,
-        highlights: Vec<OwnedUserId>,
-    ) -> Result<()>;
 }

src/service/rooms/user/data.rs

@@ -1,6 +1,9 @@
-use ruma::{OwnedRoomId, OwnedUserId, RoomId, UserId};
+use ruma::{
+    api::client::push::get_notifications, push, OwnedRoomId, OwnedUserId,
+    RoomId, UInt, UserId,
+};
 
-use crate::{service::rooms::short::ShortStateHash, Result};
+use crate::{service::rooms::short::ShortStateHash, PduEvent, Result};
 
 pub(crate) trait Data: Send + Sync {
     fn reset_notification_counts(
@@ -28,6 +31,24 @@ pub(crate) trait Data: Send + Sync {
         room_id: &RoomId,
     ) -> Result<u64>;
 
+    fn store_push_action(
+        &self,
+        pdu: &PduEvent,
+        user_id: &UserId,
+        notify: bool,
+        highlight: bool,
+        actions: &[push::Action],
+    ) -> Result<()>;
+
+    /// Get all unread or notifying events for a user since ``pdu_count``
+    fn get_notifications(
+        &self,
+        user_id: &UserId,
+        from: Option<u64>,
+        limit: Option<UInt>,
+        highlight: bool,
+    ) -> Result<(Vec<get_notifications::v3::Notification>, Option<String>)>;
+
     fn associate_token_shortstatehash(
         &self,
         room_id: &RoomId,