refactor: use <_> instead of <_parameter_name>

src/database.rs

@@ -12,7 +12,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,
@@ -124,4 +126,77 @@ 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),
+        );
+
+        // TODO: only send for user they share a room with
+        futures.push(self.global_edus.presenceid_presence.watch_prefix(b""));
+
+        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/rooms.rs

@@ -666,7 +666,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 +677,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 +689,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 +704,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 +715,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 +726,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 +929,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/main.rs

@@ -86,7 +86,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,

src/ruma_wrapper.rs

@@ -13,7 +13,7 @@ use {
         http::Status,
         response::{self, Responder},
         tokio::io::AsyncReadExt,
-        Outcome::*,
+        outcome::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)