Merge pull request 'Long polling and more' (#162) from longpoll into master

Reviewed-on: https://git.koesters.xyz/timo/conduit/pulls/162

Cargo.lock

@@ -214,9 +214,9 @@ checksum = "2e8c087f005730276d1096a652e92a8bacee2e2472bcc9715a74d2bec38b5820"
 
 [[package]]
 name = "bytemuck"
-version = "1.3.0"
+version = "1.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d40636046a60a45ee5185e885a3ccb771f7a2065fb7cbcc2a7ecfd9896d1c365"
+checksum = "db7a1029718df60331e557c9e83a55523c955e5dd2a7bfeffad6bbd50b538ae9"
 
 [[package]]
 name = "byteorder"
@@ -275,6 +275,7 @@ dependencies = [
  "serde_json",
  "sled",
  "thiserror",
+ "tokio",
 ]
 
 [[package]]
@@ -813,9 +814,9 @@ dependencies = [
 
 [[package]]
 name = "image"
-version = "0.23.7"
+version = "0.23.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a2397fc43bd5648b7117aabb3c5e62d0e62c194826ec77b0b4d0c41e62744635"
+checksum = "543904170510c1b5fb65140485d84de4a57fddb2ed685481e9020ce3d2c9f64c"
 dependencies = [
  "bytemuck",
  "byteorder",
@@ -881,9 +882,9 @@ dependencies = [
 
 [[package]]
 name = "js-sys"
-version = "0.3.42"
+version = "0.3.44"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "52732a3d3ad72c58ad2dc70624f9c17b46ecd0943b9a4f1ee37c4c18c5d983e2"
+checksum = "85a7e2c92a4804dd459b86c339278d0fe87cf93757fae222c3fa3ae75458bc73"
 dependencies = [
  "wasm-bindgen",
 ]
@@ -915,9 +916,9 @@ checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646"
 
 [[package]]
 name = "libc"
-version = "0.2.73"
+version = "0.2.74"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bd7d4bd64732af4bf3a67f367c27df8520ad7e230c5817b8ff485864d80242b9"
+checksum = "a2f02823cf78b754822df5f7f268fb59822e7296276d3e069d8e8cb26a14bd10"
 
 [[package]]
 name = "lock_api"
@@ -1238,18 +1239,18 @@ checksum = "d4fd5641d01c8f18a23da7b6fe29298ff4b55afcccdf78973b24cf3175fee32e"
 
 [[package]]
 name = "pin-project"
-version = "0.4.22"
+version = "0.4.23"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "12e3a6cdbfe94a5e4572812a0201f8c0ed98c1c452c7b8563ce2276988ef9c17"
+checksum = "ca4433fff2ae79342e497d9f8ee990d174071408f28f726d6d83af93e58e48aa"
 dependencies = [
  "pin-project-internal",
 ]
 
 [[package]]
 name = "pin-project-internal"
-version = "0.4.22"
+version = "0.4.23"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6a0ffd45cf79d88737d7cc85bfd5d2894bee1139b356e616fe85dc389c61aaf7"
+checksum = "2c0e815c3ee9a031fdf5af21c10aa17c573c9c6a566328d99e3936c34e36461f"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -1304,9 +1305,9 @@ checksum = "237a5ed80e274dbc66f86bd59c1e25edc039660be53194b5fe0a482e0f2612ea"
 
 [[package]]
 name = "proc-macro-hack"
-version = "0.5.16"
+version = "0.5.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7e0456befd48169b9f13ef0f0ad46d492cf9d2dbb918bcf38e01eed4ce3ec5e4"
+checksum = "99c605b9a0adc77b7211c6b1f722dcb613d68d66859a44f3d485a6da332b0598"
 
 [[package]]
 name = "proc-macro-nested"
@@ -1484,7 +1485,7 @@ dependencies = [
 [[package]]
 name = "rocket"
 version = "0.5.0-dev"
-source = "git+https://github.com/SergioBenitez/Rocket.git?rev=8d779caa22c63b15a6c3ceb75d8f6d4971b2eb67#8d779caa22c63b15a6c3ceb75d8f6d4971b2eb67"
+source = "git+https://github.com/timokoesters/Rocket.git?branch=empty_parameters#f6d40ecd5d871d97837b3116eb670fb3c06d95b9"
 dependencies = [
  "async-trait",
  "atomic",
@@ -1509,7 +1510,7 @@ dependencies = [
 [[package]]
 name = "rocket_codegen"
 version = "0.5.0-dev"
-source = "git+https://github.com/SergioBenitez/Rocket.git?rev=8d779caa22c63b15a6c3ceb75d8f6d4971b2eb67#8d779caa22c63b15a6c3ceb75d8f6d4971b2eb67"
+source = "git+https://github.com/timokoesters/Rocket.git?branch=empty_parameters#f6d40ecd5d871d97837b3116eb670fb3c06d95b9"
 dependencies = [
  "devise",
  "glob",
@@ -1521,7 +1522,7 @@ dependencies = [
 [[package]]
 name = "rocket_http"
 version = "0.5.0-dev"
-source = "git+https://github.com/SergioBenitez/Rocket.git?rev=8d779caa22c63b15a6c3ceb75d8f6d4971b2eb67#8d779caa22c63b15a6c3ceb75d8f6d4971b2eb67"
+source = "git+https://github.com/timokoesters/Rocket.git?branch=empty_parameters#f6d40ecd5d871d97837b3116eb670fb3c06d95b9"
 dependencies = [
  "cookie",
  "http",
@@ -1543,7 +1544,7 @@ dependencies = [
 [[package]]
 name = "ruma"
 version = "0.1.0"
-source = "git+https://github.com/ruma/ruma?rev=e047c647ddcb368e7eb1e05ae8823a9494273457#e047c647ddcb368e7eb1e05ae8823a9494273457"
+source = "git+https://github.com/ruma/ruma?rev=d5d2d1d893fa12d27960e4c58d6c09b215d06e95#d5d2d1d893fa12d27960e4c58d6c09b215d06e95"
 dependencies = [
  "ruma-api",
  "ruma-client-api",
@@ -1558,7 +1559,7 @@ dependencies = [
 [[package]]
 name = "ruma-api"
 version = "0.16.1"
-source = "git+https://github.com/ruma/ruma?rev=e047c647ddcb368e7eb1e05ae8823a9494273457#e047c647ddcb368e7eb1e05ae8823a9494273457"
+source = "git+https://github.com/ruma/ruma?rev=d5d2d1d893fa12d27960e4c58d6c09b215d06e95#d5d2d1d893fa12d27960e4c58d6c09b215d06e95"
 dependencies = [
  "http",
  "percent-encoding",
@@ -1573,7 +1574,7 @@ dependencies = [
 [[package]]
 name = "ruma-api-macros"
 version = "0.16.1"
-source = "git+https://github.com/ruma/ruma?rev=e047c647ddcb368e7eb1e05ae8823a9494273457#e047c647ddcb368e7eb1e05ae8823a9494273457"
+source = "git+https://github.com/ruma/ruma?rev=d5d2d1d893fa12d27960e4c58d6c09b215d06e95#d5d2d1d893fa12d27960e4c58d6c09b215d06e95"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -1583,7 +1584,7 @@ dependencies = [
 [[package]]
 name = "ruma-client-api"
 version = "0.9.0"
-source = "git+https://github.com/ruma/ruma?rev=e047c647ddcb368e7eb1e05ae8823a9494273457#e047c647ddcb368e7eb1e05ae8823a9494273457"
+source = "git+https://github.com/ruma/ruma?rev=d5d2d1d893fa12d27960e4c58d6c09b215d06e95#d5d2d1d893fa12d27960e4c58d6c09b215d06e95"
 dependencies = [
  "http",
  "js_int",
@@ -1600,7 +1601,7 @@ dependencies = [
 [[package]]
 name = "ruma-common"
 version = "0.2.0"
-source = "git+https://github.com/ruma/ruma?rev=e047c647ddcb368e7eb1e05ae8823a9494273457#e047c647ddcb368e7eb1e05ae8823a9494273457"
+source = "git+https://github.com/ruma/ruma?rev=d5d2d1d893fa12d27960e4c58d6c09b215d06e95#d5d2d1d893fa12d27960e4c58d6c09b215d06e95"
 dependencies = [
  "js_int",
  "ruma-serde",
@@ -1612,7 +1613,7 @@ dependencies = [
 [[package]]
 name = "ruma-events"
 version = "0.21.3"
-source = "git+https://github.com/ruma/ruma?rev=e047c647ddcb368e7eb1e05ae8823a9494273457#e047c647ddcb368e7eb1e05ae8823a9494273457"
+source = "git+https://github.com/ruma/ruma?rev=d5d2d1d893fa12d27960e4c58d6c09b215d06e95#d5d2d1d893fa12d27960e4c58d6c09b215d06e95"
 dependencies = [
  "js_int",
  "ruma-common",
@@ -1627,7 +1628,7 @@ dependencies = [
 [[package]]
 name = "ruma-events-macros"
 version = "0.21.3"
-source = "git+https://github.com/ruma/ruma?rev=e047c647ddcb368e7eb1e05ae8823a9494273457#e047c647ddcb368e7eb1e05ae8823a9494273457"
+source = "git+https://github.com/ruma/ruma?rev=d5d2d1d893fa12d27960e4c58d6c09b215d06e95#d5d2d1d893fa12d27960e4c58d6c09b215d06e95"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -1637,7 +1638,7 @@ dependencies = [
 [[package]]
 name = "ruma-federation-api"
 version = "0.0.2"
-source = "git+https://github.com/ruma/ruma?rev=e047c647ddcb368e7eb1e05ae8823a9494273457#e047c647ddcb368e7eb1e05ae8823a9494273457"
+source = "git+https://github.com/ruma/ruma?rev=d5d2d1d893fa12d27960e4c58d6c09b215d06e95#d5d2d1d893fa12d27960e4c58d6c09b215d06e95"
 dependencies = [
  "js_int",
  "ruma-api",
@@ -1652,7 +1653,7 @@ dependencies = [
 [[package]]
 name = "ruma-identifiers"
 version = "0.17.1"
-source = "git+https://github.com/ruma/ruma?rev=e047c647ddcb368e7eb1e05ae8823a9494273457#e047c647ddcb368e7eb1e05ae8823a9494273457"
+source = "git+https://github.com/ruma/ruma?rev=d5d2d1d893fa12d27960e4c58d6c09b215d06e95#d5d2d1d893fa12d27960e4c58d6c09b215d06e95"
 dependencies = [
  "rand",
  "serde",
@@ -1662,7 +1663,7 @@ dependencies = [
 [[package]]
 name = "ruma-identifiers-macros"
 version = "0.17.1"
-source = "git+https://github.com/ruma/ruma?rev=e047c647ddcb368e7eb1e05ae8823a9494273457#e047c647ddcb368e7eb1e05ae8823a9494273457"
+source = "git+https://github.com/ruma/ruma?rev=d5d2d1d893fa12d27960e4c58d6c09b215d06e95#d5d2d1d893fa12d27960e4c58d6c09b215d06e95"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -1673,7 +1674,7 @@ dependencies = [
 [[package]]
 name = "ruma-serde"
 version = "0.2.2"
-source = "git+https://github.com/ruma/ruma?rev=e047c647ddcb368e7eb1e05ae8823a9494273457#e047c647ddcb368e7eb1e05ae8823a9494273457"
+source = "git+https://github.com/ruma/ruma?rev=d5d2d1d893fa12d27960e4c58d6c09b215d06e95#d5d2d1d893fa12d27960e4c58d6c09b215d06e95"
 dependencies = [
  "form_urlencoded",
  "itoa",
@@ -1685,7 +1686,7 @@ dependencies = [
 [[package]]
 name = "ruma-signatures"
 version = "0.6.0-dev.1"
-source = "git+https://github.com/ruma/ruma?rev=e047c647ddcb368e7eb1e05ae8823a9494273457#e047c647ddcb368e7eb1e05ae8823a9494273457"
+source = "git+https://github.com/ruma/ruma?rev=d5d2d1d893fa12d27960e4c58d6c09b215d06e95#d5d2d1d893fa12d27960e4c58d6c09b215d06e95"
 dependencies = [
  "base64 0.12.3",
  "ring",
@@ -1837,9 +1838,9 @@ dependencies = [
 
 [[package]]
 name = "serde_json"
-version = "1.0.56"
+version = "1.0.57"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3433e879a558dde8b5e8feb2a04899cf34fdde1fafb894687e52105fc1162ac3"
+checksum = "164eacbdb13512ec2745fb09d51fd5b22b0d65ed294a1dcf7285a360c80a675c"
 dependencies = [
  "itoa",
  "ryu",
@@ -2028,9 +2029,9 @@ checksum = "502d53007c02d7605a05df1c1a73ee436952781653da5d0bf57ad608f66932c1"
 
 [[package]]
 name = "syn"
-version = "1.0.35"
+version = "1.0.36"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fb7f4c519df8c117855e19dd8cc851e89eb746fe7a73f0157e0d95fdec5369b0"
+checksum = "4cdb98bcb1f9d81d07b536179c269ea15999b5d14ea958196413869445bb5250"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -2338,9 +2339,9 @@ checksum = "cccddf32554fecc6acb585f82a32a72e28b48f8c4c1883ddfeeeaa96f7d8e519"
 
 [[package]]
 name = "wasm-bindgen"
-version = "0.2.65"
+version = "0.2.67"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f3edbcc9536ab7eababcc6d2374a0b7bfe13a2b6d562c5e07f370456b1a8f33d"
+checksum = "f0563a9a4b071746dd5aedbc3a28c6fe9be4586fb3fbadb67c400d4f53c6b16c"
 dependencies = [
  "cfg-if",
  "serde",
@@ -2350,9 +2351,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-backend"
-version = "0.2.65"
+version = "0.2.67"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "89ed2fb8c84bfad20ea66b26a3743f3e7ba8735a69fe7d95118c33ec8fc1244d"
+checksum = "bc71e4c5efa60fb9e74160e89b93353bc24059999c0ae0fb03affc39770310b0"
 dependencies = [
  "bumpalo",
  "lazy_static",
@@ -2365,9 +2366,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-futures"
-version = "0.4.15"
+version = "0.4.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "41ad6e4e8b2b7f8c90b6e09a9b590ea15cb0d1dbe28502b5a405cd95d1981671"
+checksum = "95f8d235a77f880bcef268d379810ea6c0af2eacfa90b1ad5af731776e0c4699"
 dependencies = [
  "cfg-if",
  "js-sys",
@@ -2377,9 +2378,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-macro"
-version = "0.2.65"
+version = "0.2.67"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "eb071268b031a64d92fc6cf691715ca5a40950694d8f683c5bb43db7c730929e"
+checksum = "97c57cefa5fa80e2ba15641578b44d36e7a64279bc5ed43c6dbaf329457a2ed2"
 dependencies = [
  "quote",
  "wasm-bindgen-macro-support",
@@ -2387,9 +2388,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-macro-support"
-version = "0.2.65"
+version = "0.2.67"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cf592c807080719d1ff2f245a687cbadb3ed28b2077ed7084b47aba8b691f2c6"
+checksum = "841a6d1c35c6f596ccea1f82504a192a60378f64b3bb0261904ad8f2f5657556"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -2400,15 +2401,15 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-shared"
-version = "0.2.65"
+version = "0.2.67"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "72b6c0220ded549d63860c78c38f3bcc558d1ca3f4efa74942c536ddbbb55e87"
+checksum = "93b162580e34310e5931c4b792560108b10fd14d64915d7fff8ff00180e70092"
 
 [[package]]
 name = "web-sys"
-version = "0.3.42"
+version = "0.3.44"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8be2398f326b7ba09815d0b403095f34dd708579220d099caae89be0b32137b2"
+checksum = "dda38f4e5ca63eda02c059d243aa25b5f35ab98451e518c51612cd0f1bd19a47"
 dependencies = [
  "js-sys",
  "wasm-bindgen",

Cargo.toml

@@ -13,8 +13,12 @@ edition = "2018"
 
 [dependencies]
 # TODO: This can become optional as soon as proper configs are supported
-rocket = { git = "https://github.com/SergioBenitez/Rocket.git", rev = "8d779caa22c63b15a6c3ceb75d8f6d4971b2eb67", features = ["tls"], optional = false } # Used to handle requests
+#rocket = { git = "https://github.com/SergioBenitez/Rocket.git", rev = "8d779caa22c63b15a6c3ceb75d8f6d4971b2eb67", features = ["tls"] } # Used to handle requests
-ruma = { git = "https://github.com/ruma/ruma", features = ["rand", "client-api", "federation-api", "unstable-pre-spec", "unstable-synapse-quirks"], rev = "e047c647ddcb368e7eb1e05ae8823a9494273457" } # Used for matrix spec type definitions and helpers
+rocket = { git = "https://github.com/timokoesters/Rocket.git", branch = "empty_parameters", features = ["tls"] }
+
+tokio = "0.2.22" # Used for long polling
+ruma = { git = "https://github.com/ruma/ruma", features = ["rand", "client-api", "federation-api", "unstable-pre-spec", "unstable-synapse-quirks"], rev = "d5d2d1d893fa12d27960e4c58d6c09b215d06e95" } # Used for matrix spec type definitions and helpers
+#ruma = { path = "../ruma/ruma", features = ["rand", "client-api", "federation-api", "unstable-pre-spec", "unstable-synapse-quirks"] }
 sled = "0.32.0" # Used for storing data permanently
 log = "0.4.8" # Used for emitting log entries
 http = "0.2.1" # Used for rocket<->ruma conversions

src/database.rs

@@ -1,5 +1,4 @@
 pub(self) mod account_data;
-pub(self) mod global_edus;
 pub(self) mod globals;
 pub(self) mod key_backups;
 pub(self) mod media;
@@ -12,7 +11,9 @@ use directories::ProjectDirs;
 use log::info;
 use std::fs::remove_dir_all;
 
-use rocket::Config;
+use futures::StreamExt;
+use rocket::{futures, Config};
+use ruma::{DeviceId, UserId};
 
 pub struct Database {
     pub globals: globals::Globals,
@@ -20,7 +21,6 @@ pub struct Database {
     pub uiaa: uiaa::Uiaa,
     pub rooms: rooms::Rooms,
     pub account_data: account_data::AccountData,
-    pub global_edus: global_edus::GlobalEdus,
     pub media: media::Media,
     pub key_backups: key_backups::KeyBackups,
     pub _db: sled::Db,
@@ -75,6 +75,7 @@ impl Database {
                 userdeviceid_metadata: db.open_tree("userdeviceid_metadata")?,
                 token_userdeviceid: db.open_tree("token_userdeviceid")?,
                 onetimekeyid_onetimekeys: db.open_tree("onetimekeyid_onetimekeys")?,
+                userid_lastonetimekeyupdate: db.open_tree("userid_lastonetimekeyupdate")?,
                 keychangeid_userid: db.open_tree("devicekeychangeid_userid")?,
                 keyid_key: db.open_tree("keyid_key")?,
                 userid_masterkeyid: db.open_tree("userid_masterkeyid")?,
@@ -91,6 +92,8 @@ impl Database {
                     roomlatestid_roomlatest: db.open_tree("roomlatestid_roomlatest")?, // Read receipts
                     roomactiveid_userid: db.open_tree("roomactiveid_userid")?, // Typing notifs
                     roomid_lastroomactiveupdate: db.open_tree("roomid_lastroomactiveupdate")?,
+                    presenceid_presence: db.open_tree("presenceid_presence")?,
+                    userid_lastpresenceupdate: db.open_tree("userid_lastpresenceupdate")?,
                 },
                 pduid_pdu: db.open_tree("pduid_pdu")?,
                 eventid_pduid: db.open_tree("eventid_pduid")?,
@@ -110,9 +113,6 @@ impl Database {
             account_data: account_data::AccountData {
                 roomuserdataid_accountdata: db.open_tree("roomuserdataid_accountdata")?,
             },
-            global_edus: global_edus::GlobalEdus {
-                presenceid_presence: db.open_tree("presenceid_presence")?, // Presence
-            },
             media: media::Media {
                 mediaid_file: db.open_tree("mediaid_file")?,
             },
@@ -124,4 +124,74 @@ impl Database {
             _db: db,
         })
     }
+
+    pub async fn watch(&self, user_id: &UserId, device_id: &DeviceId) -> () {
+        let mut userid_prefix = user_id.to_string().as_bytes().to_vec();
+        userid_prefix.push(0xff);
+        let mut userdeviceid_prefix = userid_prefix.clone();
+        userdeviceid_prefix.extend_from_slice(device_id.as_bytes());
+        userdeviceid_prefix.push(0xff);
+
+        let mut futures = futures::stream::FuturesUnordered::new();
+
+        futures.push(self.users.keychangeid_userid.watch_prefix(b""));
+
+        // Return when *any* user changed his key
+        // TODO: only send for user they share a room with
+        futures.push(
+            self.users
+                .todeviceid_events
+                .watch_prefix(&userdeviceid_prefix),
+        );
+
+        futures.push(self.rooms.userroomid_joined.watch_prefix(&userid_prefix));
+        futures.push(self.rooms.userroomid_invited.watch_prefix(&userid_prefix));
+        futures.push(self.rooms.userroomid_left.watch_prefix(&userid_prefix));
+
+        // Events for rooms we are in
+        for room_id in self.rooms.rooms_joined(user_id).filter_map(|r| r.ok()) {
+            let mut roomid_prefix = room_id.to_string().as_bytes().to_vec();
+            roomid_prefix.push(0xff);
+
+            // PDUs
+            futures.push(self.rooms.pduid_pdu.watch_prefix(&roomid_prefix));
+
+            // EDUs
+            futures.push(
+                self.rooms
+                    .edus
+                    .roomid_lastroomactiveupdate
+                    .watch_prefix(&roomid_prefix),
+            );
+
+            futures.push(
+                self.rooms
+                    .edus
+                    .roomlatestid_roomlatest
+                    .watch_prefix(&roomid_prefix),
+            );
+
+            // Room account data
+            let mut roomuser_prefix = roomid_prefix.clone();
+            roomuser_prefix.extend_from_slice(&userid_prefix);
+
+            futures.push(
+                self.account_data
+                    .roomuserdataid_accountdata
+                    .watch_prefix(&roomuser_prefix),
+            );
+        }
+
+        let mut globaluserdata_prefix = vec![0xff];
+        globaluserdata_prefix.extend_from_slice(&userid_prefix);
+
+        futures.push(
+            self.account_data
+                .roomuserdataid_accountdata
+                .watch_prefix(&globaluserdata_prefix),
+        );
+
+        // Wait until one of them finds something
+        futures.next().await;
+    }
 }

src/database/account_data.rs

@@ -1,5 +1,6 @@
 use crate::{utils, Error, Result};
 use ruma::{
+    api::client::error::ErrorKind,
     events::{AnyEvent as EduEvent, EventType},
     Raw, RoomId, UserId,
 };
@@ -19,7 +20,7 @@ impl AccountData {
         room_id: Option<&RoomId>,
         user_id: &UserId,
         event_type: EventType,
-        event: &T,
+        data: &T,
         globals: &super::globals::Globals,
     ) -> Result<()> {
         let mut prefix = room_id
@@ -42,10 +43,16 @@ impl AccountData {
         key.push(0xff);
         key.extend_from_slice(event_type.to_string().as_bytes());
 
-        self.roomuserdataid_accountdata.insert(
+        let json = serde_json::to_value(data).expect("all types here can be serialized"); // TODO: maybe add error handling
-            key,
+        if json.get("type").is_none() || json.get("content").is_none() {
-            &*serde_json::to_string(&event).expect("Map::to_string always works"),
+            return Err(Error::BadRequest(
-        )?;
+                ErrorKind::InvalidParam,
+                "Account data doesn't have all required fields.",
+            ));
+        }
+
+        self.roomuserdataid_accountdata
+            .insert(key, &*json.to_string())?;
 
         Ok(())
     }
@@ -60,7 +67,7 @@ impl AccountData {
         self.find_event(room_id, user_id, &kind)
             .map(|r| {
                 let (_, v) = r?;
-                serde_json::from_slice(&v).map_err(|_| Error::BadDatabase("could not deserialize"))
+                serde_json::from_slice(&v).map_err(|_| Error::bad_database("could not deserialize"))
             })
             .transpose()
     }

src/database/global_edus.rs

@@ -1,62 +0,0 @@
-use crate::{Error, Result};
-use ruma::Raw;
-
-pub struct GlobalEdus {
-    //pub globalallid_globalall: sled::Tree, // ToDevice, GlobalAllId = UserId + Count
-    pub(super) presenceid_presence: sled::Tree, // Presence, PresenceId = Count + UserId
-}
-
-impl GlobalEdus {
-    /// Adds a global event which will be saved until a new event replaces it (e.g. presence updates).
-    pub fn update_presence(
-        &self,
-        presence: ruma::events::presence::PresenceEvent,
-        globals: &super::globals::Globals,
-    ) -> Result<()> {
-        // Remove old entry
-        if let Some(old) = self
-            .presenceid_presence
-            .iter()
-            .keys()
-            .rev()
-            .filter_map(|r| r.ok())
-            .find(|key| {
-                key.rsplit(|&b| b == 0xff)
-                    .next()
-                    .expect("rsplit always returns an element")
-                    == presence.sender.to_string().as_bytes()
-            })
-        {
-            // This is the old global_latest
-            self.presenceid_presence.remove(old)?;
-        }
-
-        let mut presence_id = globals.next_count()?.to_be_bytes().to_vec();
-        presence_id.push(0xff);
-        presence_id.extend_from_slice(&presence.sender.to_string().as_bytes());
-
-        self.presenceid_presence.insert(
-            presence_id,
-            &*serde_json::to_string(&presence).expect("PresenceEvent can be serialized"),
-        )?;
-
-        Ok(())
-    }
-
-    /// Returns an iterator over the most recent presence updates that happened after the event with id `since`.
-    pub fn presence_since(
-        &self,
-        since: u64,
-    ) -> Result<impl Iterator<Item = Result<Raw<ruma::events::presence::PresenceEvent>>>> {
-        let first_possible_edu = (since + 1).to_be_bytes().to_vec(); // +1 so we don't send the event at since
-
-        Ok(self
-            .presenceid_presence
-            .range(&*first_possible_edu..)
-            .filter_map(|r| r.ok())
-            .map(|(_, v)| {
-                Ok(serde_json::from_slice(&v)
-                    .map_err(|_| Error::bad_database("Invalid presence event in db."))?)
-            }))
-    }
-}

src/database/rooms.rs

@@ -611,44 +611,29 @@ impl Rooms {
         self.pdus_since(user_id, room_id, 0)
     }
 
-    /// Returns an iterator over all events in a room that happened after the event with id `since`.
+    /// Returns a double-ended iterator over all events in a room that happened after the event with id `since`
+    /// in chronological order.
     pub fn pdus_since(
         &self,
         user_id: &UserId,
         room_id: &RoomId,
         since: u64,
-    ) -> Result<impl Iterator<Item = Result<PduEvent>>> {
+    ) -> Result<impl DoubleEndedIterator<Item = Result<PduEvent>>> {
-        // Create the first part of the full pdu id
-        let mut pdu_id = room_id.to_string().as_bytes().to_vec();
-        pdu_id.push(0xff);
-        pdu_id.extend_from_slice(&(since).to_be_bytes());
-
-        self.pdus_since_pduid(user_id, room_id, &pdu_id)
-    }
-
-    /// Returns an iterator over all events in a room that happened after the event with id `since`.
-    pub fn pdus_since_pduid(
-        &self,
-        user_id: &UserId,
-        room_id: &RoomId,
-        pdu_id: &[u8],
-    ) -> Result<impl Iterator<Item = Result<PduEvent>>> {
-        // Create the first part of the full pdu id
         let mut prefix = room_id.to_string().as_bytes().to_vec();
         prefix.push(0xff);
 
+        // Skip the first pdu if it's exactly at since, because we sent that last time
+        let mut first_pdu_id = prefix.clone();
+        first_pdu_id.extend_from_slice(&(since + 1).to_be_bytes());
+
+        let mut last_pdu_id = prefix.clone();
+        last_pdu_id.extend_from_slice(&u64::MAX.to_be_bytes());
+
         let user_id = user_id.clone();
         Ok(self
             .pduid_pdu
-            .range(pdu_id..)
+            .range(first_pdu_id..last_pdu_id)
-            // Skip the first pdu if it's exactly at since, because we sent that last time
-            .skip(if self.pduid_pdu.get(pdu_id)?.is_some() {
-                1
-            } else {
-                0
-            })
             .filter_map(|r| r.ok())
-            .take_while(move |(k, _)| k.starts_with(&prefix))
             .map(move |(_, v)| {
                 let mut pdu = serde_json::from_slice::<PduEvent>(&v)
                     .map_err(|_| Error::bad_database("PDU in db is invalid."))?;
@@ -666,7 +651,7 @@ impl Rooms {
         user_id: &UserId,
         room_id: &RoomId,
         until: u64,
-    ) -> impl Iterator<Item = Result<(IVec, PduEvent)>> {
+    ) -> impl Iterator<Item = Result<(u64, PduEvent)>> {
         // Create the first part of the full pdu id
         let mut prefix = room_id.to_string().as_bytes().to_vec();
         prefix.push(0xff);
@@ -677,6 +662,7 @@ impl Rooms {
         let current: &[u8] = &current;
 
         let user_id = user_id.clone();
+        let prefixlen = prefix.len();
         self.pduid_pdu
             .range(..current)
             .rev()
@@ -688,7 +674,11 @@ impl Rooms {
                 if pdu.sender != user_id {
                     pdu.unsigned.remove("transaction_id");
                 }
-                Ok((k, pdu))
+                Ok((
+                    utils::u64_from_bytes(&k[prefixlen..])
+                        .map_err(|_| Error::bad_database("Invalid pdu id in db."))?,
+                    pdu,
+                ))
             })
     }
 
@@ -699,7 +689,7 @@ impl Rooms {
         user_id: &UserId,
         room_id: &RoomId,
         from: u64,
-    ) -> impl Iterator<Item = Result<(IVec, PduEvent)>> {
+    ) -> impl Iterator<Item = Result<(u64, PduEvent)>> {
         // Create the first part of the full pdu id
         let mut prefix = room_id.to_string().as_bytes().to_vec();
         prefix.push(0xff);
@@ -710,6 +700,7 @@ impl Rooms {
         let current: &[u8] = &current;
 
         let user_id = user_id.clone();
+        let prefixlen = prefix.len();
         self.pduid_pdu
             .range(current..)
             .filter_map(|r| r.ok())
@@ -720,7 +711,11 @@ impl Rooms {
                 if pdu.sender != user_id {
                     pdu.unsigned.remove("transaction_id");
                 }
-                Ok((k, pdu))
+                Ok((
+                    utils::u64_from_bytes(&k[prefixlen..])
+                        .map_err(|_| Error::bad_database("Invalid pdu id in db."))?,
+                    pdu,
+                ))
             })
     }
 
@@ -919,7 +914,7 @@ impl Rooms {
             })
     }
 
-    /// Returns an iterator over all left members of a room.
+    /// Returns an iterator over all rooms this user joined.
     pub fn rooms_joined(&self, user_id: &UserId) -> impl Iterator<Item = Result<RoomId>> {
         self.userroomid_joined
             .scan_prefix(user_id.to_string())

src/database/rooms/edus.rs

@@ -1,15 +1,25 @@
 use crate::{utils, Error, Result};
+use js_int::UInt;
 use ruma::{
-    events::{AnyEvent as EduEvent, SyncEphemeralRoomEvent},
+    events::{
+        presence::{PresenceEvent, PresenceEventContent},
+        AnyEvent as EduEvent, SyncEphemeralRoomEvent,
+    },
+    presence::PresenceState,
     Raw, RoomId, UserId,
 };
-use std::convert::TryFrom;
+use std::{
+    collections::HashMap,
+    convert::{TryFrom, TryInto},
+};
 
 pub struct RoomEdus {
     pub(in super::super) roomuserid_lastread: sled::Tree, // RoomUserId = Room + User
     pub(in super::super) roomlatestid_roomlatest: sled::Tree, // Read Receipts, RoomLatestId = RoomId + Count + UserId
     pub(in super::super) roomactiveid_userid: sled::Tree, // Typing, RoomActiveId = RoomId + TimeoutTime + Count
     pub(in super::super) roomid_lastroomactiveupdate: sled::Tree, // LastRoomActiveUpdate = Count
+    pub(in super::super) presenceid_presence: sled::Tree, // PresenceId = RoomId + Count + UserId
+    pub(in super::super) userid_lastpresenceupdate: sled::Tree, // LastPresenceUpdate = Count
 }
 
 impl RoomEdus {
@@ -263,4 +273,182 @@ impl RoomEdus {
             })?))
         })
     }
+
+    /// Adds a presence event which will be saved until a new event replaces it.
+    ///
+    /// Note: This method takes a RoomId because presence updates are always bound to rooms to
+    /// make sure users outside these rooms can't see them.
+    pub fn update_presence(
+        &self,
+        user_id: &UserId,
+        room_id: &RoomId,
+        presence: ruma::events::presence::PresenceEvent,
+        globals: &super::super::globals::Globals,
+    ) -> Result<()> {
+        // TODO: Remove old entry? Or maybe just wipe completely from time to time?
+
+        let count = globals.next_count()?.to_be_bytes();
+
+        let mut presence_id = room_id.to_string().as_bytes().to_vec();
+        presence_id.push(0xff);
+        presence_id.extend_from_slice(&count);
+        presence_id.push(0xff);
+        presence_id.extend_from_slice(&presence.sender.to_string().as_bytes());
+
+        self.presenceid_presence.insert(
+            presence_id,
+            &*serde_json::to_string(&presence).expect("PresenceEvent can be serialized"),
+        )?;
+
+        self.userid_lastpresenceupdate.insert(
+            &user_id.to_string().as_bytes(),
+            &utils::millis_since_unix_epoch().to_be_bytes(),
+        )?;
+
+        Ok(())
+    }
+
+    /// Resets the presence timeout, so the user will stay in their current presence state.
+    pub fn ping_presence(&self, user_id: &UserId) -> Result<()> {
+        self.userid_lastpresenceupdate.insert(
+            &user_id.to_string().as_bytes(),
+            &utils::millis_since_unix_epoch().to_be_bytes(),
+        )?;
+
+        Ok(())
+    }
+
+    /// Returns the timestamp of the last presence update of this user in millis since the unix epoch.
+    pub fn last_presence_update(&self, user_id: &UserId) -> Result<Option<u64>> {
+        self.userid_lastpresenceupdate
+            .get(&user_id.to_string().as_bytes())?
+            .map(|bytes| {
+                utils::u64_from_bytes(&bytes).map_err(|_| {
+                    Error::bad_database("Invalid timestamp in userid_lastpresenceupdate.")
+                })
+            })
+            .transpose()
+    }
+
+    /// Sets all users to offline who have been quiet for too long.
+    pub fn presence_maintain(
+        &self,
+        rooms: &super::Rooms,
+        globals: &super::super::globals::Globals,
+    ) -> Result<()> {
+        let current_timestamp = utils::millis_since_unix_epoch();
+
+        for (user_id_bytes, last_timestamp) in self
+            .userid_lastpresenceupdate
+            .iter()
+            .filter_map(|r| r.ok())
+            .filter_map(|(k, bytes)| {
+                Some((
+                    k,
+                    utils::u64_from_bytes(&bytes)
+                        .map_err(|_| {
+                            Error::bad_database("Invalid timestamp in userid_lastpresenceupdate.")
+                        })
+                        .ok()?,
+                ))
+            })
+            .take_while(|(_, timestamp)| current_timestamp - timestamp > 5 * 60_000)
+        // 5 Minutes
+        {
+            self.userid_lastpresenceupdate.remove(&user_id_bytes)?;
+
+            // Send new presence events to set the user offline
+            let count = globals.next_count()?.to_be_bytes();
+            let user_id = utils::string_from_bytes(&user_id_bytes)
+                .map_err(|_| {
+                    Error::bad_database("Invalid UserId bytes in userid_lastpresenceupdate.")
+                })?
+                .try_into()
+                .map_err(|_| Error::bad_database("Invalid UserId in userid_lastpresenceupdate."))?;
+            for room_id in rooms.rooms_joined(&user_id).filter_map(|r| r.ok()) {
+                let mut presence_id = room_id.to_string().as_bytes().to_vec();
+                presence_id.push(0xff);
+                presence_id.extend_from_slice(&count);
+                presence_id.push(0xff);
+                presence_id.extend_from_slice(&user_id_bytes);
+
+                self.presenceid_presence.insert(
+                    presence_id,
+                    &*serde_json::to_string(&PresenceEvent {
+                        content: PresenceEventContent {
+                            avatar_url: None,
+                            currently_active: None,
+                            displayname: None,
+                            last_active_ago: Some(
+                                last_timestamp.try_into().expect("time is valid"),
+                            ),
+                            presence: PresenceState::Offline,
+                            status_msg: None,
+                        },
+                        sender: user_id.clone(),
+                    })
+                    .expect("PresenceEvent can be serialized"),
+                )?;
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Returns an iterator over the most recent presence updates that happened after the event with id `since`.
+    pub fn presence_since(
+        &self,
+        room_id: &RoomId,
+        since: u64,
+        rooms: &super::Rooms,
+        globals: &super::super::globals::Globals,
+    ) -> Result<HashMap<UserId, PresenceEvent>> {
+        self.presence_maintain(rooms, globals)?;
+
+        let mut prefix = room_id.to_string().as_bytes().to_vec();
+        prefix.push(0xff);
+
+        let mut first_possible_edu = prefix.clone();
+        first_possible_edu.extend_from_slice(&(since + 1).to_be_bytes()); // +1 so we don't send the event at since
+        let mut hashmap = HashMap::new();
+
+        for (key, value) in self
+            .presenceid_presence
+            .range(&*first_possible_edu..)
+            .filter_map(|r| r.ok())
+            .take_while(|(key, _)| key.starts_with(&prefix))
+        {
+            let user_id = UserId::try_from(
+                utils::string_from_bytes(
+                    key.rsplit(|&b| b == 0xff)
+                        .next()
+                        .expect("rsplit always returns an element"),
+                )
+                .map_err(|_| Error::bad_database("Invalid UserId bytes in presenceid_presence."))?,
+            )
+            .map_err(|_| Error::bad_database("Invalid UserId in presenceid_presence."))?;
+
+            let mut presence = serde_json::from_slice::<PresenceEvent>(&value)
+                .map_err(|_| Error::bad_database("Invalid presence event in db."))?;
+
+            let current_timestamp: UInt = utils::millis_since_unix_epoch()
+                .try_into()
+                .expect("time is valid");
+
+            if presence.content.presence == PresenceState::Online {
+                // Don't set last_active_ago when the user is online
+                presence.content.last_active_ago = None;
+            } else {
+                // Convert from timestamp to duration
+                presence.content.last_active_ago = presence
+                    .content
+                    .last_active_ago
+                    .map(|timestamp| current_timestamp - timestamp);
+            }
+
+            hashmap.insert(user_id, presence);
+        }
+
+        Ok(hashmap)
+    }
 }

src/database/users.rs

@@ -9,7 +9,7 @@ use ruma::{
         },
     },
     events::{AnyToDeviceEvent, EventType},
-    DeviceId, Raw, UserId,
+    DeviceId, Raw, RoomId, UserId,
 };
 use std::{collections::BTreeMap, convert::TryFrom, mem, time::SystemTime};
 
@@ -22,7 +22,8 @@ pub struct Users {
     pub(super) token_userdeviceid: sled::Tree,
 
     pub(super) onetimekeyid_onetimekeys: sled::Tree, // OneTimeKeyId = UserId + AlgorithmAndDeviceId
-    pub(super) keychangeid_userid: sled::Tree,       // KeyChangeId = Count
+    pub(super) userid_lastonetimekeyupdate: sled::Tree, // LastOneTimeKeyUpdate = Count
+    pub(super) keychangeid_userid: sled::Tree,       // KeyChangeId = RoomId + Count
     pub(super) keyid_key: sled::Tree,                // KeyId = UserId + KeyId (depends on key type)
     pub(super) userid_masterkeyid: sled::Tree,
     pub(super) userid_selfsigningkeyid: sled::Tree,
@@ -270,6 +271,7 @@ impl Users {
         device_id: &DeviceId,
         one_time_key_key: &AlgorithmAndDeviceId,
         one_time_key_value: &OneTimeKey,
+        globals: &super::globals::Globals,
     ) -> Result<()> {
         let mut key = user_id.to_string().as_bytes().to_vec();
         key.push(0xff);
@@ -294,14 +296,32 @@ impl Users {
                 .expect("OneTimeKey::to_string always works"),
         )?;
 
+        self.userid_lastonetimekeyupdate.insert(
+            &user_id.to_string().as_bytes(),
+            &globals.next_count()?.to_be_bytes(),
+        )?;
+
         Ok(())
     }
 
+    pub fn last_one_time_keys_update(&self, user_id: &UserId) -> Result<u64> {
+        self
+            .userid_lastonetimekeyupdate
+            .get(&user_id.to_string().as_bytes())?
+            .map(|bytes| {
+                utils::u64_from_bytes(&bytes).map_err(|_| {
+                    Error::bad_database("Count in roomid_lastroomactiveupdate is invalid.")
+                })
+            })
+            .unwrap_or(Ok(0))
+    }
+
     pub fn take_one_time_key(
         &self,
         user_id: &UserId,
         device_id: &DeviceId,
         key_algorithm: &KeyAlgorithm,
+        globals: &super::globals::Globals,
     ) -> Result<Option<(AlgorithmAndDeviceId, OneTimeKey)>> {
         let mut prefix = user_id.to_string().as_bytes().to_vec();
         prefix.push(0xff);
@@ -311,6 +331,11 @@ impl Users {
         prefix.extend_from_slice(key_algorithm.to_string().as_bytes());
         prefix.push(b':');
 
+        self.userid_lastonetimekeyupdate.insert(
+            &user_id.to_string().as_bytes(),
+            &globals.next_count()?.to_be_bytes(),
+        )?;
+
         self.onetimekeyid_onetimekeys
             .scan_prefix(&prefix)
             .next()
@@ -371,6 +396,7 @@ impl Users {
         user_id: &UserId,
         device_id: &DeviceId,
         device_keys: &DeviceKeys,
+        rooms: &super::rooms::Rooms,
         globals: &super::globals::Globals,
     ) -> Result<()> {
         let mut userdeviceid = user_id.to_string().as_bytes().to_vec();
@@ -382,8 +408,14 @@ impl Users {
             &*serde_json::to_string(&device_keys).expect("DeviceKeys::to_string always works"),
         )?;
 
-        self.keychangeid_userid
+        let count = globals.next_count()?.to_be_bytes();
-            .insert(globals.next_count()?.to_be_bytes(), &*user_id.to_string())?;
+        for room_id in rooms.rooms_joined(&user_id) {
+            let mut key = room_id?.to_string().as_bytes().to_vec();
+            key.push(0xff);
+            key.extend_from_slice(&count);
+
+            self.keychangeid_userid.insert(key, &*user_id.to_string())?;
+        }
 
         Ok(())
     }
@@ -394,6 +426,7 @@ impl Users {
         master_key: &CrossSigningKey,
         self_signing_key: &Option<CrossSigningKey>,
         user_signing_key: &Option<CrossSigningKey>,
+        rooms: &super::rooms::Rooms,
         globals: &super::globals::Globals,
     ) -> Result<()> {
         // TODO: Check signatures
@@ -482,8 +515,14 @@ impl Users {
                 .insert(&*user_id.to_string(), user_signing_key_key)?;
         }
 
-        self.keychangeid_userid
+        let count = globals.next_count()?.to_be_bytes();
-            .insert(globals.next_count()?.to_be_bytes(), &*user_id.to_string())?;
+        for room_id in rooms.rooms_joined(&user_id) {
+            let mut key = room_id?.to_string().as_bytes().to_vec();
+            key.push(0xff);
+            key.extend_from_slice(&count);
+
+            self.keychangeid_userid.insert(key, &*user_id.to_string())?;
+        }
 
         Ok(())
     }
@@ -494,6 +533,7 @@ impl Users {
         key_id: &str,
         signature: (String, String),
         sender_id: &UserId,
+        rooms: &super::rooms::Rooms,
         globals: &super::globals::Globals,
     ) -> Result<()> {
         let mut key = target_id.to_string().as_bytes().to_vec();
@@ -525,19 +565,42 @@ impl Users {
                 .expect("CrossSigningKey::to_string always works"),
         )?;
 
+        // TODO: Should we notify about this change?
+        let count = globals.next_count()?.to_be_bytes();
+        for room_id in rooms.rooms_joined(&target_id) {
+            let mut key = room_id?.to_string().as_bytes().to_vec();
+            key.push(0xff);
+            key.extend_from_slice(&count);
+
             self.keychangeid_userid
-            .insert(globals.next_count()?.to_be_bytes(), &*target_id.to_string())?;
+                .insert(key, &*target_id.to_string())?;
+        }
 
         Ok(())
     }
 
-    pub fn keys_changed(&self, since: u64) -> impl Iterator<Item = Result<UserId>> {
+    pub fn keys_changed(
+        &self,
+        room_id: &RoomId,
+        from: u64,
+        to: Option<u64>,
+    ) -> impl Iterator<Item = Result<UserId>> {
+        let mut prefix = room_id.to_string().as_bytes().to_vec();
+        prefix.push(0xff);
+
+        let mut start = prefix.clone();
+        start.extend_from_slice(&(from + 1).to_be_bytes());
+
+        let mut end = prefix.clone();
+        end.extend_from_slice(&to.unwrap_or(u64::MAX).to_be_bytes());
+
         self.keychangeid_userid
-            .range((since + 1).to_be_bytes()..)
+            .range(start..end)
-            .values()
+            .filter_map(|r| r.ok())
-            .map(|bytes| {
+            .take_while(move |(k, _)| k.starts_with(&prefix))
+            .map(|(_, bytes)| {
                 Ok(
-                    UserId::try_from(utils::string_from_bytes(&bytes?).map_err(|_| {
+                    UserId::try_from(utils::string_from_bytes(&bytes).map_err(|_| {
                         Error::bad_database(
                             "User ID in devicekeychangeid_userid is invalid unicode.",
                         )

src/main.rs

@@ -28,6 +28,7 @@ fn setup_rocket() -> rocket::Rocket {
                 client_server::register_route,
                 client_server::get_login_route,
                 client_server::login_route,
+                client_server::whoami_route,
                 client_server::logout_route,
                 client_server::logout_all_route,
                 client_server::change_password_route,
@@ -86,7 +87,7 @@ fn setup_rocket() -> rocket::Rocket {
                 client_server::get_state_events_route,
                 client_server::get_state_events_for_key_route,
                 client_server::get_state_events_for_empty_key_route,
-                client_server::sync_route,
+                client_server::sync_events_route,
                 client_server::get_context_route,
                 client_server::get_message_events_route,
                 client_server::turn_server_route,
@@ -107,6 +108,7 @@ fn setup_rocket() -> rocket::Rocket {
                 client_server::options_route,
                 client_server::upload_signing_keys_route,
                 client_server::upload_signatures_route,
+                client_server::get_key_changes_route,
                 client_server::pushers_route,
                 client_server::set_pushers_route,
                 //server_server::well_known_server,

src/pdu.rs

@@ -79,39 +79,99 @@ impl PduEvent {
     }
 
     pub fn to_sync_room_event(&self) -> Raw<AnySyncRoomEvent> {
-        let json = serde_json::to_string(&self).expect("PDUs are always valid");
+        let mut json = json!({
-        serde_json::from_str::<AnySyncRoomEvent>(&json)
+            "content": self.content,
-            .map(Raw::from)
+            "type": self.kind,
-            .expect("AnySyncRoomEvent can always be built from a full PDU event")
+            "event_id": self.event_id,
+            "sender": self.sender,
+            "origin_server_ts": self.origin_server_ts,
+            "unsigned": self.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")
+    }
+
     pub fn to_room_event(&self) -> Raw<AnyRoomEvent> {
-        let json = serde_json::to_string(&self).expect("PDUs are always valid");
+        let mut json = json!({
-        serde_json::from_str::<AnyRoomEvent>(&json)
+            "content": self.content,
-            .map(Raw::from)
+            "type": self.kind,
-            .expect("AnyRoomEvent can always be built from a full PDU event")
+            "event_id": self.event_id,
+            "sender": self.sender,
+            "origin_server_ts": self.origin_server_ts,
+            "unsigned": self.unsigned,
+            "room_id": self.room_id,
+        });
+
+        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")
+    }
+
     pub fn to_state_event(&self) -> Raw<AnyStateEvent> {
-        let json = serde_json::to_string(&self).expect("PDUs are always valid");
+        let json = json!({
-        serde_json::from_str::<AnyStateEvent>(&json)
+            "content": self.content,
-            .map(Raw::from)
+            "type": self.kind,
-            .expect("AnyStateEvent can always be built from a full PDU event")
+            "event_id": self.event_id,
+            "sender": self.sender,
+            "origin_server_ts": self.origin_server_ts,
+            "unsigned": self.unsigned,
+            "room_id": self.room_id,
+            "state_key": self.state_key,
+        });
+
+        serde_json::from_value(json).expect("Raw::from_value always works")
     }
+
     pub fn to_sync_state_event(&self) -> Raw<AnySyncStateEvent> {
-        let json = serde_json::to_string(&self).expect("PDUs are always valid");
+        let json = json!({
-        serde_json::from_str::<AnySyncStateEvent>(&json)
+            "content": self.content,
-            .map(Raw::from)
+            "type": self.kind,
-            .expect("AnySyncStateEvent can always be built from a full PDU event")
+            "event_id": self.event_id,
+            "sender": self.sender,
+            "origin_server_ts": self.origin_server_ts,
+            "unsigned": self.unsigned,
+            "state_key": self.state_key,
+        });
+
+        serde_json::from_value(json).expect("Raw::from_value always works")
     }
+
     pub fn to_stripped_state_event(&self) -> Raw<AnyStrippedStateEvent> {
-        let json = serde_json::to_string(&self).expect("PDUs are always valid");
+        let json = json!({
-        serde_json::from_str::<AnyStrippedStateEvent>(&json)
+            "content": self.content,
-            .map(Raw::from)
+            "type": self.kind,
-            .expect("AnyStrippedStateEvent can always be built from a full PDU event")
+            "sender": self.sender,
+            "state_key": self.state_key,
+        });
+
+        serde_json::from_value(json).expect("Raw::from_value always works")
     }
+
     pub fn to_member_event(&self) -> Raw<StateEvent<MemberEventContent>> {
-        let json = serde_json::to_string(&self).expect("PDUs are always valid");
+        let json = json!({
-        serde_json::from_str::<StateEvent<MemberEventContent>>(&json)
+            "content": self.content,
-            .map(Raw::from)
+            "type": self.kind,
-            .expect("StateEvent<MemberEventContent> can always be built from a full PDU event")
+            "event_id": self.event_id,
+            "sender": self.sender,
+            "origin_server_ts": self.origin_server_ts,
+            "redacts": self.redacts,
+            "unsigned": self.unsigned,
+            "room_id": self.room_id,
+            "state_key": self.state_key,
+        });
+
+        serde_json::from_value(json).expect("Raw::from_value always works")
     }
 }

src/ruma_wrapper.rs

@@ -11,9 +11,9 @@ use {
             Data, FromDataFuture, FromTransformedData, Transform, TransformFuture, Transformed,
         },
         http::Status,
+        outcome::Outcome::*,
         response::{self, Responder},
         tokio::io::AsyncReadExt,
-        Outcome::*,
         Request, State,
     },
     ruma::api::Endpoint,
@@ -24,7 +24,7 @@ use {
 /// first.
 pub struct Ruma<T> {
     pub body: T,
-    pub user_id: Option<UserId>,
+    pub sender_id: Option<UserId>,
     pub device_id: Option<Box<DeviceId>>,
     pub json_body: Option<Box<serde_json::value::RawValue>>, // This is None when body is not a valid string
 }
@@ -94,7 +94,7 @@ impl<'a, T: Endpoint> FromTransformedData<'a> for Ruma<T> {
             match T::try_from(http_request) {
                 Ok(t) => Success(Ruma {
                     body: t,
-                    user_id,
+                    sender_id: user_id,
                     device_id,
                     // TODO: Can we avoid parsing it again? (We only need this for append_pdu)
                     json_body: utils::string_from_bytes(&body)

sytest/sytest-whitelist

@@ -36,7 +36,6 @@ Current state appears in timeline in private history
 Current state appears in timeline in private history with many messages before
 Deleted tags appear in an incremental v2 /sync
 Deleting a non-existent alias should return a 404
-Device messages over federation wake up /sync
 Device messages wake up /sync
 Events come down the correct room
 GET /device/{deviceId}
@@ -120,6 +119,6 @@ User in shared private room does appear in user directory
 User is offline if they set_presence=offline in their sync
 Users with sufficient power-level can delete other's aliases
 Version responds 200 OK with valid structure
-Wildcard device messages over federation wake up /sync
+We should see our own leave event when rejecting an invite, even if history_visibility is restricted (riot-web/3462)
 Wildcard device messages wake up /sync
 query for user with no keys returns empty key dict