feat: handle typing events

6 years ago · 3debb6203c
7 changed files with 150 additions and 29 deletions
--- a/README.md
+++ b/README.md
@ -22,7 +22,8 @@ A Matrix Homeserver that's faster than others.
				@@ -22,7 +22,8 @@ A Matrix Homeserver that's faster than others.
 - [x] Join rooms, lookup room ids
 - [x] Basic Riot web support
 - [x] Riot room discovery
- [ ] Riot read receipts
+- [x] Riot read receipts
+- [x] Typing indications
 - [ ] Riot presence
 - [ ] Password hashing
 - [ ] Proper room creation
--- a/Rocket.toml
+++ b/Rocket.toml
@ -2,6 +2,6 @@
				@@ -2,6 +2,6 @@
 address = "0.0.0.0"
 port = 14004

-#[global.tls]
-#certs = "/etc/letsencrypt/live/matrixtesting.koesters.xyz/fullchain.pem"
-#key = "/etc/letsencrypt/live/matrixtesting.koesters.xyz/privkey.pem"
+[global.tls]
+certs = "/etc/letsencrypt/live/matrixtesting.koesters.xyz/fullchain.pem"
+key = "/etc/letsencrypt/live/matrixtesting.koesters.xyz/privkey.pem"
--- a/src/client_server.rs
+++ b/src/client_server.rs
@ -27,6 +27,7 @@ use ruma_client_api::{
				@@ -27,6 +27,7 @@ use ruma_client_api::{
        state::{create_state_event_for_empty_key, create_state_event_for_key},
        sync::sync_events,
        thirdparty::get_protocols,
+        typing::create_typing_event,
    },
    unversioned::get_supported_versions,
 };
@ -468,7 +469,7 @@ pub fn set_read_marker_route(
				@@ -468,7 +469,7 @@ pub fn set_read_marker_route(
        user_receipts.insert(
            user_id.clone(),
            ruma_events::receipt::Receipt {
-                ts: Some(utils::millis_since_unix_epoch()),
+                ts: Some(utils::millis_since_unix_epoch().try_into().unwrap()),
            },
        );
        let mut receipt_content = HashMap::new();
@ -491,6 +492,38 @@ pub fn set_read_marker_route(
				@@ -491,6 +492,38 @@ pub fn set_read_marker_route(
    MatrixResult(Ok(set_read_marker::Response))
 }

+#[put(
+    "/_matrix/client/r0/rooms/<_room_id>/typing/<_user_id>",
+    data = "<body>"
+)]
+pub fn create_typing_event_route(
+    data: State<Data>,
+    body: Ruma<create_typing_event::Request>,
+    _room_id: String,
+    _user_id: String,
+) -> MatrixResult<create_typing_event::Response> {
+    let user_id = body.user_id.clone().expect("user is authenticated");
+    let edu = EduEvent::Typing(ruma_events::typing::TypingEvent {
+        content: ruma_events::typing::TypingEventContent {
+            user_ids: vec![user_id.clone()],
+        },
+        room_id: None, // None because it can be inferred
+    });
+
+    if body.typing {
+        data.roomactive_add(
+            edu,
+            &body.room_id,
+            body.timeout.map(|d| d.as_millis() as u64).unwrap_or(30000)
+                + utils::millis_since_unix_epoch().try_into().unwrap_or(0),
+        );
+    } else {
+        data.roomactive_remove(edu, &body.room_id);
+    }
+
+    MatrixResult(Ok(create_typing_event::Response))
+}
+
 #[post("/_matrix/client/r0/createRoom", data = "<body>")]
 pub fn create_room_route(
    data: State<Data>,
@ -745,6 +778,8 @@ pub fn sync_route(
				@@ -745,6 +778,8 @@ pub fn sync_route(
    for room_id in joined_roomids {
        let pdus = { data.pdus_since(&room_id, since) };
        let room_events = pdus.into_iter().map(|pdu| pdu.to_room_event()).collect();
+        let mut edus = data.roomlatests_since(&room_id, since);
+        edus.extend_from_slice(&data.roomactives_in(&room_id));

        joined_rooms.insert(
            room_id.clone().try_into().unwrap(),
@ -765,9 +800,7 @@ pub fn sync_route(
				@@ -765,9 +800,7 @@ pub fn sync_route(
                    events: room_events,
                },
                state: sync_events::State { events: Vec::new() },
-                ephemeral: sync_events::Ephemeral {
-                    events: data.roomlatests_since(&room_id, since),
-                },
+                ephemeral: sync_events::Ephemeral { events: edus },
            },
        );
    }
--- a/src/data.rs
+++ b/src/data.rs
@ -316,7 +316,7 @@ impl Data {
				@@ -316,7 +316,7 @@ impl Data {
            room_id: room_id.clone(),
            sender: sender.clone(),
            origin: self.hostname.clone(),
-            origin_server_ts: utils::millis_since_unix_epoch(),
+            origin_server_ts: utils::millis_since_unix_epoch().try_into().unwrap(),
            kind: event_type,
            content,
            state_key,
@ -415,8 +415,7 @@ impl Data {
				@@ -415,8 +415,7 @@ impl Data {
    }

    pub fn roomlatest_update(&self, user_id: &UserId, room_id: &RoomId, event: EduEvent) {
-        let mut prefix = vec![b'd'];
-        prefix.extend_from_slice(room_id.to_string().as_bytes());
+        let mut prefix = room_id.to_string().as_bytes().to_vec();
        prefix.push(0xff);

        // Start with last
@ -475,8 +474,7 @@ impl Data {
				@@ -475,8 +474,7 @@ impl Data {
    pub fn roomlatests_since(&self, room_id: &RoomId, since: u64) -> Vec<EduEvent> {
        let mut room_latests = Vec::new();

-        let mut prefix = vec![b'd'];
-        prefix.extend_from_slice(room_id.to_string().as_bytes());
+        let mut prefix = room_id.to_string().as_bytes().to_vec();
        prefix.push(0xff);

        let mut current = prefix.clone();
@ -499,6 +497,102 @@ impl Data {
				@@ -499,6 +497,102 @@ impl Data {
        room_latests
    }

+    pub fn roomactive_add(&self, event: EduEvent, room_id: &RoomId, timeout: u64) {
+        let mut prefix = room_id.to_string().as_bytes().to_vec();
+        prefix.push(0xff);
+
+        let mut current = prefix.clone();
+
+        while let Some((key, _)) = self.db.roomactiveid_roomactive.get_gt(&current).unwrap() {
+            if key.starts_with(&prefix)
+                && utils::u64_from_bytes(key.split(|&c| c == 0xff).nth(1).unwrap())
+                    > utils::millis_since_unix_epoch().try_into().unwrap()
+            {
+                current = key.to_vec();
+                self.db.roomactiveid_roomactive.remove(&current).unwrap();
+            } else {
+                break;
+            }
+        }
+
+        // Increment the last index and use that
+        let index = utils::u64_from_bytes(
+            &self
+                .db
+                .pduid_pdu
+                .update_and_fetch(b"n", utils::increment)
+                .unwrap()
+                .unwrap(),
+        );
+
+        let mut room_active_id = prefix;
+        room_active_id.extend_from_slice(&timeout.to_be_bytes());
+        room_active_id.push(0xff);
+        room_active_id.extend_from_slice(&index.to_be_bytes());
+
+        self.db
+            .roomactiveid_roomactive
+            .insert(room_active_id, &*serde_json::to_string(&event).unwrap())
+            .unwrap();
+    }
+
+    pub fn roomactive_remove(&self, event: EduEvent, room_id: &RoomId) {
+        let mut prefix = room_id.to_string().as_bytes().to_vec();
+        prefix.push(0xff);
+
+        let mut current = prefix.clone();
+
+        let json = serde_json::to_string(&event).unwrap();
+
+        while let Some((key, value)) = self.db.roomactiveid_roomactive.get_gt(&current).unwrap() {
+            if key.starts_with(&prefix) {
+                current = key.to_vec();
+                if value == json.as_bytes() {
+                    self.db.roomactiveid_roomactive.remove(&current).unwrap();
+                    break;
+                }
+            } else {
+                break;
+            }
+        }
+    }
+
+    /// Returns a vector of the most recent read_receipts in a room that happened after the event with id `since`.
+    pub fn roomactives_in(&self, room_id: &RoomId) -> Vec<EduEvent> {
+        let mut room_actives = Vec::new();
+
+        let mut prefix = room_id.to_string().as_bytes().to_vec();
+        prefix.push(0xff);
+
+        let mut current = prefix.clone();
+        current.extend_from_slice(&utils::millis_since_unix_epoch().to_be_bytes());
+
+        while let Some((key, value)) = self.db.roomactiveid_roomactive.get_gt(&current).unwrap() {
+            if key.starts_with(&prefix) {
+                current = key.to_vec();
+                room_actives.push(
+                    serde_json::from_slice::<EventResult<_>>(&value)
+                        .expect("room_active in db is valid")
+                        .into_result()
+                        .expect("room_active in db is valid"),
+                );
+            } else {
+                break;
+            }
+        }
+
+        if room_actives.is_empty() {
+            return vec![EduEvent::Typing(ruma_events::typing::TypingEvent {
+                content: ruma_events::typing::TypingEventContent {
+                    user_ids: Vec::new(),
+                },
+                room_id: None, // None because it can be inferred
+            })];
+        } else {
+            room_actives
+        }
+    }
+
    pub fn debug(&self) {
        self.db.debug();
    }
--- a/src/database.rs
+++ b/src/database.rs
@ -64,7 +64,7 @@ pub struct Database {
				@@ -64,7 +64,7 @@ pub struct Database {
    pub userid_roomids: MultiValue,
    // EDUs:
    pub roomlatestid_roomlatest: sled::Tree, // Read Receipts, RoomLatestId = RoomId + Since + UserId TODO: Types
-    pub timeofremoval_roomrelevants: MultiValue, // Typing
+    pub roomactiveid_roomactive: sled::Tree, // Typing, RoomActiveId = TimeoutTime + Since
    pub globalallid_globalall: sled::Tree,   // ToDevice, GlobalAllId = UserId + Since
    pub globallatestid_globallatest: sled::Tree, // Presence, GlobalLatestId = Since + Type + UserId
    _db: sled::Db,
@ -103,9 +103,7 @@ impl Database {
				@@ -103,9 +103,7 @@ impl Database {
            roomid_userids: MultiValue(db.open_tree("roomid_userids").unwrap()),
            userid_roomids: MultiValue(db.open_tree("userid_roomids").unwrap()),
            roomlatestid_roomlatest: db.open_tree("roomlatestid_roomlatest").unwrap(),
-            timeofremoval_roomrelevants: MultiValue(
-                db.open_tree("timeofremoval_roomrelevants").unwrap(),
-            ),
+            roomactiveid_roomactive: db.open_tree("roomactiveid_roomactive").unwrap(),
            globalallid_globalall: db.open_tree("globalallid_globalall").unwrap(),
            globallatestid_globallatest: db.open_tree("globallatestid_globallatest").unwrap(),
            _db: db,
@ -201,7 +199,7 @@ impl Database {
				@@ -201,7 +199,7 @@ impl Database {
                String::from_utf8_lossy(&v),
            );
        }
-        println!("\n# RoomLatestId -> RoomLatest");
+        println!("\n# RoomLatestId -> RoomLatest:");
        for (k, v) in self.roomlatestid_roomlatest.iter().map(|r| r.unwrap()) {
            println!(
                "{:?} -> {:?}",
@ -209,12 +207,8 @@ impl Database {
				@@ -209,12 +207,8 @@ impl Database {
                String::from_utf8_lossy(&v),
            );
        }
-        println!("\n# TimeOfRemoval -> RoomRelevants Id:");
-        for (k, v) in self
-            .timeofremoval_roomrelevants
-            .iter_all()
-            .map(|r| r.unwrap())
-        {
+        println!("\n# RoomActiveId -> RoomActives:");
+        for (k, v) in self.roomactiveid_roomactive.iter().map(|r| r.unwrap()) {
            println!(
                "{:?} -> {:?}",
                String::from_utf8_lossy(&k),
--- a/src/main.rs
+++ b/src/main.rs
@ -40,6 +40,7 @@ fn setup_rocket(data: Data) -> rocket::Rocket {
				@@ -40,6 +40,7 @@ fn setup_rocket(data: Data) -> rocket::Rocket {
                client_server::get_keys_route,
                client_server::upload_keys_route,
                client_server::set_read_marker_route,
+                client_server::create_typing_event_route,
                client_server::create_room_route,
                client_server::get_alias_route,
                client_server::join_room_by_id_route,
--- a/src/utils.rs
+++ b/src/utils.rs
@ -4,13 +4,11 @@ use std::{
				@@ -4,13 +4,11 @@ use std::{
    time::{SystemTime, UNIX_EPOCH},
 };

-pub fn millis_since_unix_epoch() -> js_int::UInt {
-    (SystemTime::now()
+pub fn millis_since_unix_epoch() -> u64 {
+    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap()
-        .as_millis() as u64)
-        .try_into()
-        .expect("time millis are <= MAX_SAFE_UINT")
+        .as_millis() as u64
 }

 pub fn increment(old: Option<&[u8]>) -> Option<Vec<u8>> {