feat: WIP relationships and threads

src/service/media/mod.rs

@@ -8,7 +8,7 @@ use image::imageops::FilterType;
 
 use tokio::{
     fs::File,
-    io::{AsyncReadExt, AsyncWriteExt},
+    io::{AsyncReadExt, AsyncWriteExt, BufReader},
 };
 
 pub struct FileMeta {
@@ -70,7 +70,9 @@ impl Service {
         {
             let path = services().globals.get_media_file(&key);
             let mut file = Vec::new();
-            File::open(path).await?.read_to_end(&mut file).await?;
+            BufReader::new(File::open(path).await?)
+                .read_to_end(&mut file)
+                .await?;
 
             Ok(Some(FileMeta {
                 content_disposition,

src/service/mod.rs

@@ -97,6 +97,7 @@ impl Services {
                     db,
                     lasttimelinecount_cache: Mutex::new(HashMap::new()),
                 },
+                threads: rooms::threads::Service { db },
                 user: rooms::user::Service { db },
             },
             transaction_ids: transaction_ids::Service { db },

src/service/pdu.rs

@@ -1,9 +1,9 @@
 use crate::Error;
 use ruma::{
     events::{
-        room::member::RoomMemberEventContent, AnyEphemeralRoomEvent, AnyStateEvent,
-        AnyStrippedStateEvent, AnySyncStateEvent, AnySyncTimelineEvent, AnyTimelineEvent,
-        StateEvent, TimelineEventType,
+        room::member::RoomMemberEventContent, AnyEphemeralRoomEvent, AnyMessageLikeEvent,
+        AnyStateEvent, AnyStrippedStateEvent, AnySyncStateEvent, AnySyncTimelineEvent,
+        AnyTimelineEvent, StateEvent, TimelineEventType,
     },
     serde::Raw,
     state_res, CanonicalJsonObject, CanonicalJsonValue, EventId, MilliSecondsSinceUnixEpoch,
@@ -175,6 +175,30 @@ impl PduEvent {
         serde_json::from_value(json).expect("Raw::from_value always works")
     }
 
+    #[tracing::instrument(skip(self))]
+    pub fn to_message_like_event(&self) -> Raw<AnyMessageLikeEvent> {
+        let mut json = json!({
+            "content": self.content,
+            "type": self.kind,
+            "event_id": self.event_id,
+            "sender": self.sender,
+            "origin_server_ts": self.origin_server_ts,
+            "room_id": self.room_id,
+        });
+
+        if let Some(unsigned) = &self.unsigned {
+            json["unsigned"] = json!(unsigned);
+        }
+        if let Some(state_key) = &self.state_key {
+            json["state_key"] = json!(state_key);
+        }
+        if let Some(redacts) = &self.redacts {
+            json["redacts"] = json!(redacts);
+        }
+
+        serde_json::from_value(json).expect("Raw::from_value always works")
+    }
+
     #[tracing::instrument(skip(self))]
     pub fn to_state_event(&self) -> Raw<AnyStateEvent> {
         let mut json = json!({

src/service/pusher/mod.rs

@@ -162,9 +162,7 @@ impl Service {
             &pdu.room_id,
         )? {
             let n = match action {
-                Action::DontNotify => false,
-                // TODO: Implement proper support for coalesce
-                Action::Notify | Action::Coalesce => true,
+                Action::Notify => true,
                 Action::SetTweak(tweak) => {
                     tweaks.push(tweak.clone());
                     continue;

src/service/rooms/mod.rs

@@ -13,6 +13,7 @@ pub mod state;
 pub mod state_accessor;
 pub mod state_cache;
 pub mod state_compressor;
+pub mod threads;
 pub mod timeline;
 pub mod user;
 
@@ -32,6 +33,7 @@ pub trait Data:
     + state_cache::Data
     + state_compressor::Data
     + timeline::Data
+    + threads::Data
     + user::Data
 {
 }
@@ -53,5 +55,6 @@ pub struct Service {
     pub state_cache: state_cache::Service,
     pub state_compressor: state_compressor::Service,
     pub timeline: timeline::Service,
+    pub threads: threads::Service,
     pub user: user::Service,
 }

src/service/rooms/pdu_metadata/data.rs

@@ -4,6 +4,7 @@ use crate::Result;
 use ruma::{EventId, RoomId};
 
 pub trait Data: Send + Sync {
+    fn add_relation(&self, from: u64, to: u64) -> Result<()>;
     fn mark_as_referenced(&self, room_id: &RoomId, event_ids: &[Arc<EventId>]) -> Result<()>;
     fn is_event_referenced(&self, room_id: &RoomId, event_id: &EventId) -> Result<bool>;
     fn mark_event_soft_failed(&self, event_id: &EventId) -> Result<()>;

src/service/rooms/pdu_metadata/mod.rs

@@ -4,13 +4,20 @@ use std::sync::Arc;
 pub use data::Data;
 use ruma::{EventId, RoomId};
 
-use crate::Result;
+use crate::{services, Result};
 
 pub struct Service {
     pub db: &'static dyn Data,
 }
 
 impl Service {
+    #[tracing::instrument(skip(self, from, to))]
+    pub fn add_relation(&self, from: &EventId, to: &EventId) -> Result<()> {
+        let from = services().rooms.short.get_or_create_shorteventid(from)?;
+        let to = services().rooms.short.get_or_create_shorteventid(to)?;
+        self.db.add_relation(from, to)
+    }
+
     #[tracing::instrument(skip(self, room_id, event_ids))]
     pub fn mark_as_referenced(&self, room_id: &RoomId, event_ids: &[Arc<EventId>]) -> Result<()> {
         self.db.mark_as_referenced(room_id, event_ids)

src/service/rooms/threads/data.rs

@@ -0,0 +1,15 @@
+use crate::{PduEvent, Result};
+use ruma::{api::client::threads::get_threads::v1::IncludeThreads, OwnedUserId, RoomId, UserId};
+
+pub trait Data: Send + Sync {
+    fn threads_until<'a>(
+        &'a self,
+        user_id: &'a UserId,
+        room_id: &'a RoomId,
+        until: u64,
+        include: &'a IncludeThreads,
+    ) -> Result<Box<dyn Iterator<Item = Result<(u64, PduEvent)>> + 'a>>;
+
+    fn update_participants(&self, root_id: &[u8], participants: &[OwnedUserId]) -> Result<()>;
+    fn get_participants(&self, root_id: &[u8]) -> Result<Option<Vec<OwnedUserId>>>;
+}

src/service/rooms/threads/mod.rs

@@ -0,0 +1,119 @@
+mod data;
+use std::sync::Arc;
+
+pub use data::Data;
+use ruma::{
+    api::client::{error::ErrorKind, threads::get_threads::v1::IncludeThreads},
+    events::{relation::BundledThread, StateEventType},
+    uint, CanonicalJsonValue, EventId, OwnedUserId, RoomId, UserId,
+};
+use serde::Deserialize;
+use serde_json::json;
+
+use crate::{services, utils, Error, PduEvent, Result};
+
+use super::timeline::PduCount;
+
+pub struct Service {
+    pub db: &'static dyn Data,
+}
+
+impl Service {
+    pub fn threads_until<'a>(
+        &'a self,
+        user_id: &'a UserId,
+        room_id: &'a RoomId,
+        until: u64,
+        include: &'a IncludeThreads,
+    ) -> Result<impl Iterator<Item = Result<(u64, PduEvent)>> + 'a> {
+        self.db.threads_until(user_id, room_id, until, include)
+    }
+
+    pub fn add_to_thread<'a>(&'a self, root_event_id: &EventId, pdu: &PduEvent) -> Result<()> {
+        let root_id = &services()
+            .rooms
+            .timeline
+            .get_pdu_id(root_event_id)?
+            .ok_or_else(|| {
+                Error::BadRequest(
+                    ErrorKind::InvalidParam,
+                    "Invalid event id in thread message",
+                )
+            })?;
+
+        let root_pdu = services()
+            .rooms
+            .timeline
+            .get_pdu_from_id(root_id)?
+            .ok_or_else(|| {
+                Error::BadRequest(ErrorKind::InvalidParam, "Thread root pdu not found")
+            })?;
+
+        let mut root_pdu_json = services()
+            .rooms
+            .timeline
+            .get_pdu_json_from_id(root_id)?
+            .ok_or_else(|| {
+                Error::BadRequest(ErrorKind::InvalidParam, "Thread root pdu not found")
+            })?;
+
+        if let CanonicalJsonValue::Object(unsigned) = root_pdu_json
+            .entry("unsigned".to_owned())
+            .or_insert_with(|| CanonicalJsonValue::Object(Default::default()))
+        {
+            if let Some(mut relations) = unsigned
+                .get("m.relations")
+                .and_then(|r| r.as_object())
+                .and_then(|r| r.get("m.thread"))
+                .and_then(|relations| {
+                    serde_json::from_value::<BundledThread>(relations.clone().into()).ok()
+                })
+            {
+                // Thread already existed
+                relations.count += uint!(1);
+                relations.latest_event = pdu.to_message_like_event();
+
+                let content = serde_json::to_value(relations).expect("to_value always works");
+
+                unsigned.insert(
+                    "m.relations".to_owned(),
+                    json!({ "m.thread": content })
+                        .try_into()
+                        .expect("thread is valid json"),
+                );
+            } else {
+                // New thread
+                let relations = BundledThread {
+                    latest_event: pdu.to_message_like_event(),
+                    count: uint!(1),
+                    current_user_participated: true,
+                };
+
+                let content = serde_json::to_value(relations).expect("to_value always works");
+
+                unsigned.insert(
+                    "m.relations".to_owned(),
+                    json!({ "m.thread": content })
+                        .try_into()
+                        .expect("thread is valid json"),
+                );
+            }
+
+            services()
+                .rooms
+                .timeline
+                .replace_pdu(root_id, &root_pdu_json, &root_pdu)?;
+        }
+
+        let mut users = Vec::new();
+        if let Some(userids) = self.db.get_participants(&root_id)? {
+            users.extend_from_slice(&userids);
+            users.push(pdu.sender.clone());
+        } else {
+            users.push(root_pdu.sender);
+            users.push(pdu.sender.clone());
+        }
+
+        self.db.update_participants(root_id, &users)
+    }
+}

src/service/rooms/timeline/data.rs

@@ -57,7 +57,12 @@ pub trait Data: Send + Sync {
     ) -> Result<()>;
 
     /// Removes a pdu and creates a new one with the same id.
-    fn replace_pdu(&self, pdu_id: &[u8], pdu: &PduEvent) -> Result<()>;
+    fn replace_pdu(
+        &self,
+        pdu_id: &[u8],
+        pdu_json: &CanonicalJsonObject,
+        pdu: &PduEvent,
+    ) -> Result<()>;
 
     /// Returns an iterator over all events and their tokens in a room that happened before the
     /// event with id `until` in reverse-chronological order.

src/service/rooms/timeline/mod.rs

@@ -18,13 +18,13 @@ use ruma::{
     events::{
         push_rules::PushRulesEvent,
         room::{
-            create::RoomCreateEventContent, member::MembershipState,
+            create::RoomCreateEventContent, encrypted::Relation, member::MembershipState,
             power_levels::RoomPowerLevelsEventContent,
         },
         GlobalAccountDataEventType, StateEventType, TimelineEventType,
     },
     push::{Action, Ruleset, Tweak},
-    serde::Base64,
+    serde::{Base64, JsonObject},
     state_res,
     state_res::{Event, RoomVersion},
     uint, user_id, CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, OwnedRoomId,
@@ -197,8 +197,13 @@ impl Service {
 
     /// Removes a pdu and creates a new one with the same id.
     #[tracing::instrument(skip(self))]
-    fn replace_pdu(&self, pdu_id: &[u8], pdu: &PduEvent) -> Result<()> {
-        self.db.replace_pdu(pdu_id, pdu)
+    pub fn replace_pdu(
+        &self,
+        pdu_id: &[u8],
+        pdu_json: &CanonicalJsonObject,
+        pdu: &PduEvent,
+    ) -> Result<()> {
+        self.db.replace_pdu(pdu_id, pdu_json, pdu)
     }
 
     /// Creates a new persisted data unit and adds it to a room.
@@ -352,9 +357,7 @@ impl Service {
                 &pdu.room_id,
             )? {
                 match action {
-                    Action::DontNotify => notify = false,
-                    // TODO: Implement proper support for coalesce
-                    Action::Notify | Action::Coalesce => notify = true,
+                    Action::Notify => notify = true,
                     Action::SetTweak(Tweak::Highlight(true)) => {
                         highlight = true;
                     }
@@ -457,6 +460,50 @@ impl Service {
             _ => {}
         }
 
+        // Update Relationships
+        #[derive(Deserialize)]
+        struct ExtractRelatesTo {
+            #[serde(rename = "m.relates_to")]
+            relates_to: Relation,
+        }
+
+        #[derive(Clone, Debug, Deserialize)]
+        struct ExtractEventId {
+            event_id: OwnedEventId,
+        }
+        #[derive(Clone, Debug, Deserialize)]
+        struct ExtractRelatesToEventId {
+            #[serde(rename = "m.relates_to")]
+            relates_to: ExtractEventId,
+        }
+
+        if let Ok(content) = serde_json::from_str::<ExtractRelatesToEventId>(pdu.content.get()) {
+            services()
+                .rooms
+                .pdu_metadata
+                .add_relation(&pdu.event_id, &content.relates_to.event_id)?;
+        }
+
+        if let Ok(content) = serde_json::from_str::<ExtractRelatesTo>(pdu.content.get()) {
+            match content.relates_to {
+                Relation::Reply { in_reply_to } => {
+                    // We need to do it again here, because replies don't have
+                    // event_id as a top level field
+                    services()
+                        .rooms
+                        .pdu_metadata
+                        .add_relation(&pdu.event_id, &in_reply_to.event_id)?;
+                }
+                Relation::Thread(thread) => {
+                    services()
+                        .rooms
+                        .threads
+                        .add_to_thread(&thread.event_id, pdu)?;
+                }
+                _ => {} // TODO: Aggregate other types
+            }
+        }
+
         for appservice in services().appservice.all()? {
             if services()
                 .rooms
@@ -957,12 +1004,17 @@ impl Service {
     /// Replace a PDU with the redacted form.
     #[tracing::instrument(skip(self, reason))]
     pub fn redact_pdu(&self, event_id: &EventId, reason: &PduEvent) -> Result<()> {
+        // TODO: Don't reserialize, keep original json
         if let Some(pdu_id) = self.get_pdu_id(event_id)? {
             let mut pdu = self
                 .get_pdu_from_id(&pdu_id)?
                 .ok_or_else(|| Error::bad_database("PDU ID points to invalid PDU."))?;
             pdu.redact(reason)?;
-            self.replace_pdu(&pdu_id, &pdu)?;
+            self.replace_pdu(
+                &pdu_id,
+                &utils::to_canonical_object(&pdu).expect("PDU is an object"),
+                &pdu,
+            )?;
         }
         // If event does not exist, just noop
         Ok(())