Merge branch 'correct-sendtxn' into pushers

It seems in order to test this I may also have to impl the email 3pid
route? I need to call the set_pusher route somehow.

Cargo.lock

@@ -44,9 +44,9 @@ checksum = "5f093eed78becd229346bf859eec0aa4dd7ddde0757287b2b4107a1f09c80002"
 
 [[package]]
 name = "async-trait"
-version = "0.1.42"
+version = "0.1.48"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8d3a45e77e34375a7923b1e8febb049bb011f064714a8e17a1a616fef01da13d"
+checksum = "36ea56748e10732c49404c153638a15ec3d6211ec5ff35d9bb20e13b93576adf"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -134,9 +134,9 @@ checksum = "bed57e2090563b83ba8f83366628ce535a7584c9afa4c9fc0612a03925c6df58"
 
 [[package]]
 name = "byteorder"
-version = "1.4.2"
+version = "1.4.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ae44d1a3d5a19df61dd0c8beb138458ac2a53a7ac09eba97d55592540004306b"
+checksum = "14c189c53d098945499cdfa7ecc63567cf3886b3332b312a5b4585d8d3a6a610"
 
 [[package]]
 name = "bytes"
@@ -711,21 +711,6 @@ dependencies = [
  "want",
 ]
 
-[[package]]
-name = "hyper-rustls"
-version = "0.22.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5f9f7a97316d44c0af9b0301e65010573a853a9fc97046d7331d7f6bc0fd5a64"
-dependencies = [
- "futures-util",
- "hyper",
- "log",
- "rustls",
- "tokio",
- "tokio-rustls",
- "webpki",
-]
-
 [[package]]
 name = "hyper-tls"
 version = "0.5.0"
@@ -769,9 +754,9 @@ dependencies = [
 
 [[package]]
 name = "indexmap"
-version = "1.6.1"
+version = "1.6.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4fb1fa934250de4de8aef298d81c729a7d33d8c239daa3a7575e6b92bfc7313b"
+checksum = "824845a0bf897a9042383849b02c1bc219c2383772efcd5c6f9766fa4b81aef3"
 dependencies = [
  "autocfg",
  "hashbrown",
@@ -877,9 +862,9 @@ checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646"
 
 [[package]]
 name = "libc"
-version = "0.2.87"
+version = "0.2.88"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "265d751d31d6780a3f956bb5b8022feba2d94eeee5a84ba64f4212eedca42213"
+checksum = "03b07a082330a35e43f63177cc01689da34fbffa0105e1246cf0311472cac73a"
 
 [[package]]
 name = "linked-hash-map"
@@ -1263,9 +1248,9 @@ dependencies = [
 
 [[package]]
 name = "pin-project-lite"
-version = "0.2.5"
+version = "0.2.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0cf491442e4b033ed1c722cb9f0df5fcfcf4de682466c46469c36bc47dc5548a"
+checksum = "dc0e1f259c92177c30a4c9d177246edd0a3568b25756a977d0632cf8fa37e905"
 
 [[package]]
 name = "pin-utils"
@@ -1453,14 +1438,13 @@ dependencies = [
 
 [[package]]
 name = "regex"
-version = "1.4.3"
+version = "1.4.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d9251239e129e16308e70d853559389de218ac275b515068abc96829d05b948a"
+checksum = "54fd1046a3107eb58f42de31d656fee6853e5d276c455fd943742dce89fc3dd3"
 dependencies = [
  "aho-corasick",
  "memchr",
  "regex-syntax",
- "thread_local",
 ]
 
 [[package]]
@@ -1475,9 +1459,9 @@ dependencies = [
 
 [[package]]
 name = "regex-syntax"
-version = "0.6.22"
+version = "0.6.23"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b5eb417147ba9860a96cfe72a0b93bf88fee1744b5636ec99ab20c1aa9376581"
+checksum = "24d5f089152e60f62d28b835fbff2cd2e8dc0baf1ac13343bef92ab7eed84548"
 
 [[package]]
 name = "remove_dir_all"
@@ -1490,9 +1474,9 @@ dependencies = [
 
 [[package]]
 name = "reqwest"
-version = "0.11.1"
+version = "0.11.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0460542b551950620a3648c6aa23318ac6b3cd779114bd873209e6e8b5eb1c34"
+checksum = "bf12057f289428dbf5c591c74bf10392e4a8003f993405a902f20117019022d4"
 dependencies = [
  "base64 0.13.0",
  "bytes",
@@ -1502,7 +1486,6 @@ dependencies = [
  "http",
  "http-body",
  "hyper",
- "hyper-rustls",
  "hyper-tls",
  "ipnet",
  "js-sys",
@@ -1512,17 +1495,14 @@ dependencies = [
  "native-tls",
  "percent-encoding",
  "pin-project-lite",
- "rustls",
  "serde",
  "serde_urlencoded",
  "tokio",
  "tokio-native-tls",
- "tokio-rustls",
  "url",
  "wasm-bindgen",
  "wasm-bindgen-futures",
  "web-sys",
- "webpki-roots",
  "winreg 0.7.0",
 ]
 
@@ -1897,9 +1877,9 @@ dependencies = [
 
 [[package]]
 name = "security-framework"
-version = "2.1.1"
+version = "2.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2dfd318104249865096c8da1dfabf09ddbb6d0330ea176812a62ec75e40c4166"
+checksum = "d493c5f39e02dfb062cd8f33301f90f9b13b650e8c1b1d0fd75c19dd64bff69d"
 dependencies = [
  "bitflags",
  "core-foundation",
@@ -1935,18 +1915,18 @@ checksum = "388a1df253eca08550bef6c72392cfe7c30914bf41df5269b68cbd6ff8f570a3"
 
 [[package]]
 name = "serde"
-version = "1.0.123"
+version = "1.0.124"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "92d5161132722baa40d802cc70b15262b98258453e85e5d1d365c757c73869ae"
+checksum = "bd761ff957cb2a45fbb9ab3da6512de9de55872866160b23c25f1a841e99d29f"
 dependencies = [
  "serde_derive",
 ]
 
 [[package]]
 name = "serde_derive"
-version = "1.0.123"
+version = "1.0.124"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9391c295d64fc0abb2c556bad848f33cb8296276b1ad2677d1ae1ace4f258f31"
+checksum = "1800f7693e94e186f5e25a28291ae1570da908aff7d97a095dec1e56ff99069b"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -2147,9 +2127,9 @@ checksum = "213701ba3370744dcd1a12960caa4843b3d68b4d1c0a5d575e0d65b2ee9d16c0"
 
 [[package]]
 name = "syn"
-version = "1.0.60"
+version = "1.0.63"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c700597eca8a5a762beb35753ef6b94df201c81cca676604f547495a0d7f0081"
+checksum = "8fd9bc7ccc2688b3344c2f48b9b546648b25ce0b20fc717ee7fa7981a8ca9717"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -2295,9 +2275,9 @@ checksum = "cda74da7e1a664f795bb1f8a87ec406fb89a02522cf6e50620d016add6dbbf5c"
 
 [[package]]
 name = "tokio"
-version = "1.2.0"
+version = "1.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e8190d04c665ea9e6b6a0dc45523ade572c088d2e6566244c1122671dbf4ae3a"
+checksum = "8d56477f6ed99e10225f38f9f75f872f29b8b8bd8c0b946f63345bb144e9eeda"
 dependencies = [
  "autocfg",
  "bytes",
@@ -2346,9 +2326,9 @@ dependencies = [
 
 [[package]]
 name = "tokio-util"
-version = "0.6.3"
+version = "0.6.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ebb7cb2f00c5ae8df755b252306272cd1790d39728363936e01827e11f0b017b"
+checksum = "ec31e5cc6b46e653cf57762f36f71d5e6386391d88a72fd6db4508f8f676fb29"
 dependencies = [
  "bytes",
  "futures-core",
@@ -2387,9 +2367,9 @@ dependencies = [
 
 [[package]]
 name = "tracing-attributes"
-version = "0.1.13"
+version = "0.1.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a8a9bd1db7706f2373a190b0d067146caa39350c486f3d455b0e33b431f94c07"
+checksum = "c42e6fa53307c8a17e4ccd4dc81cf5ec38db9209f59b222210375b54ee40d1e2"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -2441,9 +2421,9 @@ dependencies = [
 
 [[package]]
 name = "tracing-subscriber"
-version = "0.2.16"
+version = "0.2.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8ab8966ac3ca27126141f7999361cc97dd6fb4b71da04c02044fa9045d98bb96"
+checksum = "705096c6f83bf68ea5d357a6aa01829ddbdac531b357b45abeca842938085baa"
 dependencies = [
  "ansi_term",
  "chrono",
@@ -2699,15 +2679,6 @@ dependencies = [
  "untrusted",
 ]
 
-[[package]]
-name = "webpki-roots"
-version = "0.21.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "82015b7e0b8bad8185994674a13a93306bea76cf5a16c5a181382fd3a5ec2376"
-dependencies = [
- "webpki",
-]
-
 [[package]]
 name = "weezl"
 version = "0.1.4"

Cargo.toml

@@ -20,7 +20,7 @@ rocket = { git = "https://github.com/SergioBenitez/Rocket.git", rev = "93e62c86e
 # Used for matrix spec type definitions and helpers
 #ruma = { git = "https://github.com/ruma/ruma", features = ["rand", "appservice-api", "client-api", "federation-api", "unstable-pre-spec", "unstable-synapse-quirks", "unstable-exhaustive-types"], rev = "0a10afe6dacc2b7a50a8002c953d10b7fb4e37bc" }
 # ruma = { git = "https://github.com/DevinR528/ruma", features = ["rand", "client-api", "federation-api", "unstable-exhaustive-types", "unstable-pre-spec", "unstable-synapse-quirks"], branch = "verified-export" }
-ruma = { path = "../ruma/ruma", features = ["unstable-exhaustive-types", "rand", "client-api", "federation-api", "unstable-pre-spec", "unstable-synapse-quirks"] }
+ruma = { path = "../ruma/ruma", features = ["unstable-exhaustive-types", "rand", "client-api", "federation-api", "push-gateway-api", "unstable-pre-spec", "unstable-synapse-quirks"] }
 
 # Used when doing state resolution
 # state-res = { git = "https://github.com/timokoesters/state-res", branch = "timo-spec-comp", features = ["unstable-pre-spec"] }
@@ -50,7 +50,7 @@ rand = "0.8.3"
 # Used to hash passwords
 rust-argon2 = "0.8.3"
 # Used to send requests
-reqwest = { version = "0.11.1", features = ["rustls-tls"] }
+reqwest = { version = "0.11.1" }
 # Used for conduit::Error type
 thiserror = "1.0.24"
 # Used to generate thumbnails for images

src/client_server/account.rs

@@ -657,3 +657,17 @@ pub async fn deactivate_route(
     }
     .into())
 }
+
+/*/
+#[cfg_attr(
+    feature = "conduit_bin",
+    get("/_matrix/client/r0/account/3pid", data = "<body>")
+)]
+pub async fn third_party_route(
+    body: Ruma<account::add_3pid::Request<'_>>,
+) -> ConduitResult<account::add_3pid::Response> {
+    let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+
+    Ok(account::add_3pid::Response::default().into())
+}
+*/

src/client_server/push.rs

@@ -5,10 +5,11 @@ use ruma::{
         error::ErrorKind,
         r0::push::{
             delete_pushrule, get_pushers, get_pushrule, get_pushrule_actions, get_pushrule_enabled,
-            get_pushrules_all, set_pushrule, set_pushrule_actions, set_pushrule_enabled, RuleKind,
+            get_pushrules_all, set_pusher, set_pushrule, set_pushrule_actions,
+            set_pushrule_enabled, RuleKind,
         },
     },
-    events::EventType,
+    events::{push_rules, EventType},
     push::{
         ConditionalPushRuleInit, ContentPushRule, OverridePushRule, PatternedPushRuleInit,
         RoomPushRule, SenderPushRule, SimplePushRuleInit, UnderridePushRule,
@@ -31,7 +32,7 @@ pub async fn get_pushrules_all_route(
 
     let event = db
         .account_data
-        .get::<ruma::events::push_rules::PushRulesEvent>(None, &sender_user, EventType::PushRules)?
+        .get::<push_rules::PushRulesEvent>(None, &sender_user, EventType::PushRules)?
         .ok_or(Error::BadRequest(
             ErrorKind::NotFound,
             "PushRules event not found.",
@@ -56,7 +57,7 @@ pub async fn get_pushrule_route(
 
     let event = db
         .account_data
-        .get::<ruma::events::push_rules::PushRulesEvent>(None, &sender_user, EventType::PushRules)?
+        .get::<push_rules::PushRulesEvent>(None, &sender_user, EventType::PushRules)?
         .ok_or(Error::BadRequest(
             ErrorKind::NotFound,
             "PushRules event not found.",
@@ -122,7 +123,7 @@ pub async fn set_pushrule_route(
 
     let mut event = db
         .account_data
-        .get::<ruma::events::push_rules::PushRulesEvent>(None, &sender_user, EventType::PushRules)?
+        .get::<push_rules::PushRulesEvent>(None, &sender_user, EventType::PushRules)?
         .ok_or(Error::BadRequest(
             ErrorKind::NotFound,
             "PushRules event not found.",
@@ -269,7 +270,7 @@ pub async fn get_pushrule_actions_route(
 
     let mut event = db
         .account_data
-        .get::<ruma::events::push_rules::PushRulesEvent>(None, &sender_user, EventType::PushRules)?
+        .get::<push_rules::PushRulesEvent>(None, &sender_user, EventType::PushRules)?
         .ok_or(Error::BadRequest(
             ErrorKind::NotFound,
             "PushRules event not found.",
@@ -333,7 +334,7 @@ pub async fn set_pushrule_actions_route(
 
     let mut event = db
         .account_data
-        .get::<ruma::events::push_rules::PushRulesEvent>(None, &sender_user, EventType::PushRules)?
+        .get::<push_rules::PushRulesEvent>(None, &sender_user, EventType::PushRules)?
         .ok_or(Error::BadRequest(
             ErrorKind::NotFound,
             "PushRules event not found.",
@@ -437,7 +438,7 @@ pub async fn get_pushrule_enabled_route(
 
     let mut event = db
         .account_data
-        .get::<ruma::events::push_rules::PushRulesEvent>(None, &sender_user, EventType::PushRules)?
+        .get::<push_rules::PushRulesEvent>(None, &sender_user, EventType::PushRules)?
         .ok_or(Error::BadRequest(
             ErrorKind::NotFound,
             "PushRules event not found.",
@@ -602,7 +603,7 @@ pub async fn delete_pushrule_route(
 
     let mut event = db
         .account_data
-        .get::<ruma::events::push_rules::PushRulesEvent>(None, &sender_user, EventType::PushRules)?
+        .get::<push_rules::PushRulesEvent>(None, &sender_user, EventType::PushRules)?
         .ok_or(Error::BadRequest(
             ErrorKind::NotFound,
             "PushRules event not found.",
@@ -676,22 +677,38 @@ pub async fn delete_pushrule_route(
     Ok(delete_pushrule::Response.into())
 }
 
-#[cfg_attr(feature = "conduit_bin", get("/_matrix/client/r0/pushers"))]
-#[tracing::instrument]
-pub async fn get_pushers_route() -> ConduitResult<get_pushers::Response> {
+#[cfg_attr(
+    feature = "conduit_bin",
+    get("/_matrix/client/r0/pushers", data = "<body>")
+)]
+#[tracing::instrument(skip(db, body))]
+pub async fn get_pushers_route(
+    db: State<'_, Database>,
+    body: Ruma<get_pushers::Request>,
+) -> ConduitResult<get_pushers::Response> {
+    let sender = body.sender_user.as_ref().expect("authenticated endpoint");
+
     Ok(get_pushers::Response {
-        pushers: Vec::new(),
+        pushers: db.pusher.get_pusher(sender)?,
     }
     .into())
 }
 
-#[cfg_attr(feature = "conduit_bin", post("/_matrix/client/r0/pushers/set"))]
-#[tracing::instrument(skip(db))]
-pub async fn set_pushers_route(db: State<'_, Database>) -> ConduitResult<get_pushers::Response> {
+#[cfg_attr(
+    feature = "conduit_bin",
+    post("/_matrix/client/r0/pushers/set", data = "<body>")
+)]
+#[tracing::instrument(skip(db, body))]
+pub async fn set_pushers_route(
+    db: State<'_, Database>,
+    body: Ruma<set_pusher::Request>,
+) -> ConduitResult<set_pusher::Response> {
+    let sender = body.sender_user.as_ref().expect("authenticated endpoint");
+    let pusher = body.pusher.clone();
+
+    db.pusher.set_pusher(sender, pusher)?;
+
     db.flush().await?;
 
-    Ok(get_pushers::Response {
-        pushers: Vec::new(),
-    }
-    .into())
+    Ok(set_pusher::Response::default().into())
 }

src/client_server/read_marker.rs

@@ -3,7 +3,9 @@ use crate::{ConduitResult, Database, Error, Ruma};
 use ruma::{
     api::client::{
         error::ErrorKind,
-        r0::{read_marker::set_read_marker, receipt::create_receipt},
+        r0::{
+            capabilities::get_capabilities, read_marker::set_read_marker, receipt::create_receipt,
+        },
     },
     events::{AnyEphemeralRoomEvent, AnyEvent, EventType},
 };

src/client_server/sync.rs

@@ -706,7 +706,7 @@ pub async fn sync_events_route(
         let delay = tokio::time::sleep(duration);
         tokio::pin!(delay);
         tokio::select! {
-            _ = &mut delay => {}
+            _ = &mut delay, if delay.is_elapsed() => {}
             _ = watcher => {}
         }
     }

src/database.rs

@@ -4,6 +4,7 @@ pub mod appservice;
 pub mod globals;
 pub mod key_backups;
 pub mod media;
+pub mod pusher;
 pub mod rooms;
 pub mod sending;
 pub mod transaction_ids;
@@ -80,6 +81,7 @@ pub struct Database {
     pub sending: sending::Sending,
     pub admin: admin::Admin,
     pub appservice: appservice::Appservice,
+    pub pusher: pusher::PushData,
     pub _db: sled::Db,
 }
 
@@ -194,6 +196,7 @@ impl Database {
                 cached_registrations: Arc::new(RwLock::new(HashMap::new())),
                 id_appserviceregistrations: db.open_tree("id_appserviceregistrations")?,
             },
+            pusher: pusher::PushData::new(&db)?,
             _db: db,
         };
 

src/database/globals.rs

@@ -13,6 +13,7 @@ use trust_dns_resolver::TokioAsyncResolver;
 
 pub const COUNTER: &str = "c";
 
+pub type DestinationCache = Arc<RwLock<HashMap<Box<ServerName>, (String, Option<String>)>>>;
 type WellKnownMap = HashMap<Box<ServerName>, (String, Option<String>)>;
 
 #[derive(Clone)]
@@ -88,8 +89,8 @@ impl Globals {
                 Error::bad_config("Failed to set up trust dns resolver with system config.")
             })?,
             actual_destination_cache: Arc::new(RwLock::new(HashMap::new())),
-            jwt_decoding_key,
             servertimeout_signingkey,
+            jwt_decoding_key,
         })
     }
 

src/database/pusher.rs

@@ -0,0 +1,546 @@
+use crate::{Database, Error, PduEvent, Result};
+use log::{error, info, warn};
+use ruma::{
+    api::{
+        client::r0::push::{Pusher, PusherKind},
+        push_gateway::send_event_notification::{
+            self,
+            v1::{Device, Notification, NotificationCounts, NotificationPriority},
+        },
+        OutgoingRequest,
+    },
+    events::room::{
+        member::{MemberEventContent, MembershipState},
+        message::{MessageEventContent, MessageType, TextMessageEventContent},
+        power_levels::PowerLevelsEventContent,
+    },
+    events::EventType,
+    push::{Action, PushCondition, PushFormat, Ruleset, Tweak},
+    uint, UInt, UserId,
+};
+
+use std::{convert::TryFrom, fmt::Debug, time::Duration};
+
+#[derive(Debug, Clone)]
+pub struct PushData {
+    /// UserId + pushkey -> Pusher
+    pub(super) senderkey_pusher: sled::Tree,
+}
+
+impl PushData {
+    pub fn new(db: &sled::Db) -> Result<Self> {
+        Ok(Self {
+            senderkey_pusher: db.open_tree("senderkey_pusher")?,
+        })
+    }
+
+    pub fn set_pusher(&self, sender: &UserId, pusher: Pusher) -> Result<()> {
+        println!("CCCCCCCCCCCCCCCCCCCCCc");
+        dbg!(&pusher);
+        let mut key = sender.as_bytes().to_vec();
+        key.push(0xff);
+        key.extend_from_slice(pusher.pushkey.as_bytes());
+
+        // There are 2 kinds of pushers but the spec says: null deletes the pusher.
+        if pusher.kind.is_none() {
+            return self
+                .senderkey_pusher
+                .remove(key)
+                .map(|_| ())
+                .map_err(Into::into);
+        }
+
+        self.senderkey_pusher.insert(
+            dbg!(key),
+            &*serde_json::to_string(&pusher).expect("Pusher is valid JSON string"),
+        )?;
+
+        Ok(())
+    }
+
+    pub fn get_pusher(&self, sender: &UserId) -> Result<Vec<Pusher>> {
+        let mut prefix = sender.as_bytes().to_vec();
+        prefix.push(0xff);
+
+        self.senderkey_pusher
+            .scan_prefix(dbg!(prefix))
+            .values()
+            .map(|push| {
+                println!("DDDDDDDDDDDDDDDDDDDDDDDDDD");
+                let push =
+                    dbg!(push).map_err(|_| Error::bad_database("Invalid push bytes in db."))?;
+                Ok(serde_json::from_slice(&*push)
+                    .map_err(|_| Error::bad_database("Invalid Pusher in db."))?)
+            })
+            .collect()
+    }
+}
+
+pub async fn send_request<T: OutgoingRequest>(
+    globals: &crate::database::globals::Globals,
+    destination: &str,
+    request: T,
+) -> Result<T::IncomingResponse>
+where
+    T: Debug,
+{
+    let destination = destination.replace("/_matrix/push/v1/notify", "");
+
+    let http_request = request
+        .try_into_http_request(&destination, Some(""))
+        .map_err(|e| {
+            warn!("Failed to find destination {}: {}", destination, e);
+            Error::BadServerResponse("Invalid destination")
+        })?;
+
+    let reqwest_request = reqwest::Request::try_from(http_request)
+        .expect("all http requests are valid reqwest requests");
+
+    // TODO: we could keep this very short and let expo backoff do it's thing...
+    //*reqwest_request.timeout_mut() = Some(Duration::from_secs(5));
+
+    let url = reqwest_request.url().clone();
+    let reqwest_response = globals
+        .reqwest_client()
+        .execute(dbg!(reqwest_request))
+        .await;
+
+    // Because reqwest::Response -> http::Response is complicated:
+    match reqwest_response {
+        Ok(mut reqwest_response) => {
+            let status = reqwest_response.status();
+            let mut http_response = http::Response::builder().status(status);
+            let headers = http_response.headers_mut().unwrap();
+
+            for (k, v) in reqwest_response.headers_mut().drain() {
+                if let Some(key) = k {
+                    headers.insert(key, v);
+                }
+            }
+
+            let status = reqwest_response.status();
+
+            let body = reqwest_response
+                .bytes()
+                .await
+                .unwrap_or_else(|e| {
+                    warn!("server error {}", e);
+                    Vec::new().into()
+                }) // TODO: handle timeout
+                .into_iter()
+                .collect::<Vec<_>>();
+
+            if status != 200 {
+                info!(
+                    "Push gateway returned bad response {} {}\n{}\n{:?}",
+                    destination,
+                    status,
+                    url,
+                    crate::utils::string_from_bytes(&body)
+                );
+            }
+
+            let response = T::IncomingResponse::try_from(
+                http_response
+                    .body(body)
+                    .expect("reqwest body is valid http body"),
+            );
+            response.map_err(|_| {
+                info!(
+                    "Push gateway returned invalid response bytes {}\n{}",
+                    destination, url
+                );
+                Error::BadServerResponse("Push gateway returned bad response.")
+            })
+        }
+        Err(e) => Err(e.into()),
+    }
+}
+
+pub async fn send_push_notice(
+    user: &UserId,
+    unread: UInt,
+    pushers: &[Pusher],
+    ruleset: Ruleset,
+    pdu: &PduEvent,
+    db: &Database,
+) -> Result<()> {
+    if let Some(msgtype) = pdu.content.get("msgtype").and_then(|b| b.as_str()) {
+        if msgtype == "m.notice" {
+            return Ok(());
+        }
+    }
+
+    for rule in ruleset.into_iter() {
+        // TODO: can actions contain contradictory Actions
+        if rule
+            .actions
+            .iter()
+            .any(|act| matches!(act, ruma::push::Action::DontNotify))
+            || !rule.enabled
+        {
+            continue;
+        }
+
+        match dbg!(rule.rule_id.as_str()) {
+            ".m.rule.master" => {}
+            ".m.rule.suppress_notices" => {
+                if pdu.kind == EventType::RoomMessage
+                    && pdu
+                        .content
+                        .get("msgtype")
+                        .map_or(false, |ty| ty == "m.notice")
+                {
+                    let tweaks = rule
+                        .actions
+                        .iter()
+                        .filter_map(|a| match a {
+                            Action::SetTweak(tweak) => Some(tweak.clone()),
+                            _ => None,
+                        })
+                        .collect::<Vec<_>>();
+                    send_notice(unread, pushers, tweaks, pdu, db, rule.rule_id.as_str()).await?;
+                    break;
+                }
+            }
+            ".m.rule.invite_for_me" => {
+                if let EventType::RoomMember = &pdu.kind {
+                    if pdu.state_key.as_deref() == Some(user.as_str())
+                        && serde_json::from_value::<MemberEventContent>(pdu.content.clone())
+                            .map_err(|_| Error::bad_database("PDU contained bad message content"))?
+                            .membership
+                            == MembershipState::Invite
+                    {
+                        let tweaks = rule
+                            .actions
+                            .iter()
+                            .filter_map(|a| match a {
+                                Action::SetTweak(tweak) => Some(tweak.clone()),
+                                _ => None,
+                            })
+                            .collect::<Vec<_>>();
+                        send_notice(unread, pushers, tweaks, pdu, db, rule.rule_id.as_str())
+                            .await?;
+                        break;
+                    }
+                }
+            }
+            ".m.rule.member_event" => {
+                if let EventType::RoomMember = &pdu.kind {
+                    // TODO use this?
+                    let _member = serde_json::from_value::<MemberEventContent>(pdu.content.clone())
+                        .map_err(|_| Error::bad_database("PDU contained bad message content"))?;
+                    if let Some(conditions) = rule.conditions {
+                        if conditions.iter().any(|cond| match cond {
+                            PushCondition::EventMatch { key, pattern } => {
+                                let mut json =
+                                    serde_json::to_value(pdu).expect("PDU is valid JSON");
+                                for key in key.split('.') {
+                                    json = json[key].clone();
+                                }
+                                // TODO: this is baddddd
+                                json.to_string().contains(pattern)
+                            }
+                            _ => false,
+                        }) {
+                            let tweaks = rule
+                                .actions
+                                .iter()
+                                .filter_map(|a| match a {
+                                    Action::SetTweak(tweak) => Some(tweak.clone()),
+                                    _ => None,
+                                })
+                                .collect::<Vec<_>>();
+                            send_notice(unread, pushers, tweaks, pdu, db, rule.rule_id.as_str())
+                                .await?;
+                            break;
+                        }
+                    }
+                }
+            }
+            ".m.rule.contains_display_name" => {
+                if let EventType::RoomMessage = &pdu.kind {
+                    let msg_content =
+                        serde_json::from_value::<MessageEventContent>(pdu.content.clone())
+                            .map_err(|_| {
+                                Error::bad_database("PDU contained bad message content")
+                            })?;
+                    if let MessageType::Text(TextMessageEventContent { body, .. }) =
+                        &msg_content.msgtype
+                    {
+                        if body.contains(user.localpart()) {
+                            let tweaks = rule
+                                .actions
+                                .iter()
+                                .filter_map(|a| match a {
+                                    Action::SetTweak(tweak) => Some(tweak.clone()),
+                                    _ => None,
+                                })
+                                .collect::<Vec<_>>();
+                            send_notice(unread, pushers, tweaks, pdu, db, rule.rule_id.as_str())
+                                .await?;
+                            break;
+                        }
+                    }
+                }
+            }
+            ".m.rule.tombstone" => {
+                if pdu.kind == EventType::RoomTombstone && pdu.state_key.as_deref() == Some("") {
+                    let tweaks = rule
+                        .actions
+                        .iter()
+                        .filter_map(|a| match a {
+                            Action::SetTweak(tweak) => Some(tweak.clone()),
+                            _ => None,
+                        })
+                        .collect::<Vec<_>>();
+                    send_notice(unread, pushers, tweaks, pdu, db, rule.rule_id.as_str()).await?;
+                    break;
+                }
+            }
+            ".m.rule.roomnotif" => {
+                if let EventType::RoomMessage = &pdu.kind {
+                    let msg_content =
+                        serde_json::from_value::<MessageEventContent>(pdu.content.clone())
+                            .map_err(|_| {
+                                Error::bad_database("PDU contained bad message content")
+                            })?;
+                    if let MessageType::Text(TextMessageEventContent { body, .. }) =
+                        &msg_content.msgtype
+                    {
+                        let power_level_cmp = |pl: PowerLevelsEventContent| {
+                            &pl.notifications.room
+                                <= pl.users.get(&pdu.sender).unwrap_or(&ruma::int!(0))
+                        };
+                        let deserialize = |pl: PduEvent| {
+                            serde_json::from_value::<PowerLevelsEventContent>(pl.content).ok()
+                        };
+                        if body.contains("@room")
+                            && db
+                                .rooms
+                                .room_state_get(&pdu.room_id, &EventType::RoomPowerLevels, "")?
+                                .map(|(_, pl)| pl)
+                                .map(deserialize)
+                                .flatten()
+                                .map_or(false, power_level_cmp)
+                        {
+                            let tweaks = rule
+                                .actions
+                                .iter()
+                                .filter_map(|a| match a {
+                                    Action::SetTweak(tweak) => Some(tweak.clone()),
+                                    _ => None,
+                                })
+                                .collect::<Vec<_>>();
+                            send_notice(unread, pushers, tweaks, pdu, db, rule.rule_id.as_str())
+                                .await?;
+                            break;
+                        }
+                    }
+                }
+            }
+            ".m.rule.contains_user_name" => {
+                if let EventType::RoomMessage = &pdu.kind {
+                    let msg_content =
+                        serde_json::from_value::<MessageEventContent>(pdu.content.clone())
+                            .map_err(|_| {
+                                Error::bad_database("PDU contained bad message content")
+                            })?;
+                    if let MessageType::Text(TextMessageEventContent { body, .. }) =
+                        &msg_content.msgtype
+                    {
+                        if body.contains(user.localpart()) {
+                            let tweaks = rule
+                                .actions
+                                .iter()
+                                .filter_map(|a| match a {
+                                    Action::SetTweak(tweak) => Some(tweak.clone()),
+                                    _ => None,
+                                })
+                                .collect::<Vec<_>>();
+                            send_notice(unread, pushers, tweaks, pdu, db, rule.rule_id.as_str())
+                                .await?;
+                            break;
+                        }
+                    }
+                }
+            }
+            ".m.rule.call" => {
+                if pdu.kind == EventType::CallInvite {
+                    let tweaks = rule
+                        .actions
+                        .iter()
+                        .filter_map(|a| match a {
+                            Action::SetTweak(tweak) => Some(tweak.clone()),
+                            _ => None,
+                        })
+                        .collect::<Vec<_>>();
+                    send_notice(unread, pushers, tweaks, pdu, db, rule.rule_id.as_str()).await?;
+                    break;
+                }
+            }
+            ".m.rule.encrypted_room_one_to_one" => {
+                if db.rooms.room_members(&pdu.room_id).count() == 2
+                    && pdu.kind == EventType::RoomEncrypted
+                {
+                    let tweaks = rule
+                        .actions
+                        .iter()
+                        .filter_map(|a| match a {
+                            Action::SetTweak(tweak) => Some(tweak.clone()),
+                            _ => None,
+                        })
+                        .collect::<Vec<_>>();
+                    send_notice(unread, pushers, tweaks, pdu, db, rule.rule_id.as_str()).await?;
+                    break;
+                }
+            }
+            ".m.rule.room_one_to_one" => {
+                if db.rooms.room_members(&pdu.room_id).count() == 2
+                    && pdu.kind == EventType::RoomMessage
+                {
+                    let tweaks = rule
+                        .actions
+                        .iter()
+                        .filter_map(|a| match a {
+                            Action::SetTweak(tweak) => Some(tweak.clone()),
+                            _ => None,
+                        })
+                        .collect::<Vec<_>>();
+                    send_notice(unread, pushers, tweaks, pdu, db, rule.rule_id.as_str()).await?;
+                    break;
+                }
+            }
+            ".m.rule.message" => {
+                if pdu.kind == EventType::RoomMessage {
+                    let tweaks = rule
+                        .actions
+                        .iter()
+                        .filter_map(|a| match a {
+                            Action::SetTweak(tweak) => Some(tweak.clone()),
+                            _ => None,
+                        })
+                        .collect::<Vec<_>>();
+                    send_notice(unread, pushers, tweaks, pdu, db, rule.rule_id.as_str()).await?;
+                    break;
+                }
+            }
+            ".m.rule.encrypted" => {
+                if pdu.kind == EventType::RoomEncrypted {
+                    let tweaks = rule
+                        .actions
+                        .iter()
+                        .filter_map(|a| match a {
+                            Action::SetTweak(tweak) => Some(tweak.clone()),
+                            _ => None,
+                        })
+                        .collect::<Vec<_>>();
+                    send_notice(unread, pushers, tweaks, pdu, db, rule.rule_id.as_str()).await?;
+                    break;
+                }
+            }
+            _ => {}
+        }
+    }
+
+    Ok(())
+}
+
+async fn send_notice(
+    unread: UInt,
+    pushers: &[Pusher],
+    tweaks: Vec<Tweak>,
+    event: &PduEvent,
+    db: &Database,
+    name: &str,
+) -> Result<()> {
+    println!("BBBBBBBBBBBBBBBr");
+    let (http, _emails): (Vec<&Pusher>, _) = dbg!(pushers)
+        .iter()
+        .partition(|pusher| pusher.kind == Some(PusherKind::Http));
+
+    // TODO:
+    // Two problems with this
+    // 1. if "event_id_only" is the only format kind it seems we should never add more info
+    // 2. can pusher/devices have conflicting formats
+    for pusher in dbg!(http) {
+        let event_id_only = pusher.data.format == Some(PushFormat::EventIdOnly);
+        let url = if let Some(url) = pusher.data.url.as_ref() {
+            url
+        } else {
+            error!("Http Pusher must have URL specified.");
+            continue;
+        };
+
+        let mut device = Device::new(pusher.app_id.clone(), pusher.pushkey.clone());
+        let mut data_minus_url = pusher.data.clone();
+        // The url must be stripped off according to spec
+        data_minus_url.url = None;
+        device.data = Some(data_minus_url);
+
+        // Tweaks are only added if the format is NOT event_id_only
+        if !event_id_only {
+            device.tweaks = tweaks.clone();
+        }
+
+        let d = &[device];
+        let mut notifi = Notification::new(d);
+
+        notifi.prio = NotificationPriority::Low;
+        notifi.event_id = Some(&event.event_id);
+        notifi.room_id = Some(&event.room_id);
+        // TODO: missed calls
+        notifi.counts = NotificationCounts::new(unread, uint!(0));
+
+        if event.kind == EventType::RoomEncrypted
+            || tweaks
+                .iter()
+                .any(|t| matches!(t, Tweak::Highlight(true) | Tweak::Sound(_)))
+        {
+            notifi.prio = NotificationPriority::High
+        }
+
+        if event_id_only {
+            error!("SEND PUSH NOTICE `{}`", name);
+            send_request(
+                &db.globals,
+                &url,
+                send_event_notification::v1::Request::new(notifi),
+            )
+            .await?;
+        } else {
+            notifi.sender = Some(&event.sender);
+            notifi.event_type = Some(&event.kind);
+            notifi.content = serde_json::value::to_raw_value(&event.content).ok();
+
+            if event.kind == EventType::RoomMember {
+                notifi.user_is_target = event.state_key.as_deref() == Some(event.sender.as_str());
+            }
+
+            let user_name = db.users.displayname(&event.sender)?;
+            notifi.sender_display_name = user_name.as_deref();
+            let room_name = db
+                .rooms
+                .room_state_get(&event.room_id, &EventType::RoomName, "")?
+                .map(|(_, pdu)| match pdu.content.get("name") {
+                    Some(serde_json::Value::String(s)) => Some(s.to_string()),
+                    _ => None,
+                })
+                .flatten();
+            notifi.room_name = room_name.as_deref();
+
+            error!("SEND PUSH NOTICE Full `{}`", name);
+            send_request(
+                &db.globals,
+                &url,
+                send_event_notification::v1::Request::new(notifi),
+            )
+            .await?;
+        }
+    }
+
+    // TODO: email
+    // for email in emails {}
+
+    Ok(())
+}

src/database/rooms.rs

@@ -86,8 +86,7 @@ impl Rooms {
         room_id: &RoomId,
         state_hash: &StateHashId,
     ) -> Result<BTreeMap<(EventType, String), PduEvent>> {
-        let r = self
-            .stateid_pduid
+        self.stateid_pduid
             .scan_prefix(&state_hash)
             .values()
             .map(|short_id| {
@@ -123,9 +122,7 @@ impl Rooms {
                     pdu,
                 ))
             })
-            .collect();
-
-        r
+            .collect()
     }
 
     /// Returns a single PDU from `room_id` with key (`event_type`, `state_key`).
@@ -586,6 +583,9 @@ impl Rooms {
         self.eventid_pduid
             .insert(pdu.event_id.as_bytes(), &*pdu_id)?;
 
+        // See if the event matches any known pushers
+        db.sending.send_push_pdu(&*pdu_id)?;
+
         match pdu.kind {
             EventType::RoomRedaction => {
                 if let Some(redact_id) = &pdu.redacts {
@@ -1415,6 +1415,7 @@ impl Rooms {
             self.alias_roomid
                 .insert(alias.alias(), room_id.as_bytes())?;
             let mut aliasid = room_id.as_bytes().to_vec();
+            aliasid.push(0xff);
             aliasid.extend_from_slice(&globals.next_count()?.to_be_bytes());
             self.aliasid_alias.insert(aliasid, &*alias.alias())?;
         } else {
@@ -1427,7 +1428,10 @@ impl Rooms {
                     "Alias does not exist.",
                 ))?;
 
-            for key in self.aliasid_alias.scan_prefix(room_id).keys() {
+            let mut prefix = room_id.to_vec();
+            prefix.push(0xff);
+
+            for key in self.aliasid_alias.scan_prefix(prefix).keys() {
                 self.aliasid_alias.remove(key?)?;
             }
         }

src/database/sending.rs

@@ -6,50 +6,57 @@ use std::{
     time::{Duration, Instant, SystemTime},
 };
 
-use crate::{appservice_server, server_server, utils, Error, PduEvent, Result};
+use crate::{
+    appservice_server, database::pusher, server_server, utils, Database, Error, PduEvent, Result,
+};
 use federation::transactions::send_transaction_message;
 use log::{info, warn};
 use ring::digest;
 use rocket::futures::stream::{FuturesUnordered, StreamExt};
 use ruma::{
     api::{appservice, federation, OutgoingRequest},
-    ServerName,
+    events::{push_rules, EventType},
+    uint, ServerName, UInt,
 };
 use sled::IVec;
 use tokio::{select, sync::Semaphore};
 
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub enum OutgoingKind {
+    Appservice(Box<ServerName>),
+    Push(Vec<u8>),
+    Normal(Box<ServerName>),
+}
+
 #[derive(Clone)]
 pub struct Sending {
     /// The state for a given state hash.
-    pub(super) servernamepduids: sled::Tree, // ServernamePduId = (+)ServerName + PduId
-    pub(super) servercurrentpdus: sled::Tree, // ServerCurrentPdus = (+)ServerName + PduId (pduid can be empty for reservation)
+    pub(super) servernamepduids: sled::Tree, // ServernamePduId = (+ / $)ServerName / UserId + PduId
+    pub(super) servercurrentpdus: sled::Tree, // ServerCurrentPdus = (+ / $)ServerName / UserId + PduId (pduid can be empty for reservation)
     pub(super) maximum_requests: Arc<Semaphore>,
 }
 
 impl Sending {
-    pub fn start_handler(
-        &self,
-        globals: &super::globals::Globals,
-        rooms: &super::rooms::Rooms,
-        appservice: &super::appservice::Appservice,
-    ) {
+    pub fn start_handler(&self, db: &Database) {
         let servernamepduids = self.servernamepduids.clone();
         let servercurrentpdus = self.servercurrentpdus.clone();
-        let maximum_requests = self.maximum_requests.clone();
-        let rooms = rooms.clone();
-        let globals = globals.clone();
-        let appservice = appservice.clone();
+
+        let db = db.clone();
 
         tokio::spawn(async move {
             let mut futures = FuturesUnordered::new();
 
             // Retry requests we could not finish yet
-            let mut current_transactions = HashMap::<(Box<ServerName>, bool), Vec<IVec>>::new();
+            let mut current_transactions = HashMap::<OutgoingKind, Vec<IVec>>::new();
 
-            for (key, server, pdu, is_appservice) in servercurrentpdus
+            for (key, outgoing_kind, pdu) in servercurrentpdus
                 .iter()
                 .filter_map(|r| r.ok())
-                .filter_map(|(key, _)| Self::parse_servercurrentpdus(key).ok())
+                .filter_map(|(key, _)| {
+                    Self::parse_servercurrentpdus(&key)
+                        .ok()
+                        .map(|(k, p)| (key, k, p))
+                })
             {
                 if pdu.is_empty() {
                     // Remove old reservation key
@@ -58,7 +65,7 @@ impl Sending {
                 }
 
                 let entry = current_transactions
-                    .entry((server, is_appservice))
+                    .entry(outgoing_kind)
                     .or_insert_with(Vec::new);
 
                 if entry.len() > 30 {
@@ -70,42 +77,56 @@ impl Sending {
                 entry.push(pdu);
             }
 
-            for ((server, is_appservice), pdus) in current_transactions {
+            for (outgoing_kind, pdus) in current_transactions {
                 // Create new reservation
-                let mut prefix = if is_appservice {
-                    b"+".to_vec()
-                } else {
-                    Vec::new()
+                let mut prefix = match &outgoing_kind {
+                    OutgoingKind::Appservice(server) => {
+                        let mut p = b"+".to_vec();
+                        p.extend_from_slice(server.as_bytes());
+                        p
+                    }
+                    OutgoingKind::Push(id) => {
+                        let mut p = b"$".to_vec();
+                        p.extend_from_slice(&id);
+                        p
+                    }
+                    OutgoingKind::Normal(server) => {
+                        let mut p = Vec::new();
+                        p.extend_from_slice(server.as_bytes());
+                        p
+                    }
                 };
-                prefix.extend_from_slice(server.as_bytes());
                 prefix.push(0xff);
                 servercurrentpdus.insert(prefix, &[]).unwrap();
 
-                futures.push(Self::handle_event(
-                    server,
-                    is_appservice,
-                    pdus,
-                    &globals,
-                    &rooms,
-                    &appservice,
-                    &maximum_requests,
-                ));
+                futures.push(Self::handle_event(outgoing_kind.clone(), pdus, &db));
             }
 
-            let mut last_failed_try: HashMap<Box<ServerName>, (u32, Instant)> = HashMap::new();
+            let mut last_failed_try: HashMap<OutgoingKind, (u32, Instant)> = HashMap::new();
 
             let mut subscriber = servernamepduids.watch_prefix(b"");
             loop {
                 select! {
                     Some(response) = futures.next() => {
                         match response {
-                            Ok((server, is_appservice)) => {
-                                let mut prefix = if is_appservice {
-                                    b"+".to_vec()
-                                } else {
-                                    Vec::new()
+                            Ok(outgoing_kind) => {
+                                let mut prefix = match &outgoing_kind {
+                                    OutgoingKind::Appservice(server) => {
+                                        let mut p = b"+".to_vec();
+                                        p.extend_from_slice(server.as_bytes());
+                                        p
+                                    }
+                                    OutgoingKind::Push(id) => {
+                                        let mut p = b"$".to_vec();
+                                        p.extend_from_slice(&id);
+                                        p
+                                    },
+                                    OutgoingKind::Normal(server) => {
+                                        let mut p = vec![];
+                                        p.extend_from_slice(server.as_bytes());
+                                        p
+                                    },
                                 };
-                                prefix.extend_from_slice(server.as_bytes());
                                 prefix.push(0xff);
 
                                 for key in servercurrentpdus
@@ -138,23 +159,41 @@ impl Sending {
                                         servernamepduids.remove(&current_key).unwrap();
                                     }
 
-                                    futures.push(Self::handle_event(server, is_appservice, new_pdus, &globals, &rooms, &appservice, &maximum_requests));
+                                    futures.push(
+                                        Self::handle_event(
+                                            outgoing_kind.clone(),
+                                            new_pdus,
+                                            &db,
+                                        )
+                                    );
                                 } else {
                                     servercurrentpdus.remove(&prefix).unwrap();
                                     // servercurrentpdus with the prefix should be empty now
                                 }
                             }
-                            Err((server, is_appservice, e)) => {
-                                info!("Couldn't send transaction to {}\n{}", server, e);
-                                let mut prefix = if is_appservice {
-                                    b"+".to_vec()
-                                } else {
-                                    Vec::new()
+                            Err((outgoing_kind, e)) => {
+                                info!("Couldn't send transaction to {:?}\n{}", outgoing_kind, e);
+                                let mut prefix = match &outgoing_kind {
+                                    OutgoingKind::Appservice(serv) => {
+                                        let mut p = b"+".to_vec();
+                                        p.extend_from_slice(serv.as_bytes());
+                                        p
+                                    },
+                                    OutgoingKind::Push(id) => {
+                                        let mut p = b"$".to_vec();
+                                        p.extend_from_slice(&id);
+                                        p
+                                    },
+                                    OutgoingKind::Normal(serv) => {
+                                        let mut p = vec![];
+                                        p.extend_from_slice(serv.as_bytes());
+                                        p
+                                    },
                                 };
-                                prefix.extend_from_slice(server.as_bytes());
+
                                 prefix.push(0xff);
 
-                                last_failed_try.insert(server.clone(), match last_failed_try.get(&server) {
+                                last_failed_try.insert(outgoing_kind.clone(), match last_failed_try.get(&outgoing_kind) {
                                     Some(last_failed) => {
                                         (last_failed.0+1, Instant::now())
                                     },
@@ -180,40 +219,56 @@ impl Sending {
 
                                 instant.elapsed() < min_elapsed_duration
                             };
-                            if let Some((server, is_appservice, pdu_id)) = utils::string_from_bytes(
+                            if let Some((outgoing_kind, pdu_id)) = utils::string_from_bytes(
                                     parts
                                         .next()
                                         .expect("splitn will always return 1 or more elements"),
                                 )
-                                .map_err(|_| Error::bad_database("ServerName in servernamepduid bytes are invalid."))
-                                .map(|server_str| {
+                                .map_err(|_| Error::bad_database("[Utf8] ServerName in servernamepduid bytes are invalid."))
+                                .and_then(|ident_str| {
                                     // Appservices start with a plus
-                                    if server_str.starts_with('+') {
-                                        (server_str[1..].to_owned(), true)
+                                    Ok(if ident_str.starts_with('+') {
+                                        OutgoingKind::Appservice(
+                                            Box::<ServerName>::try_from(&ident_str[1..])
+                                                .map_err(|_| Error::bad_database("ServerName in servernamepduid is invalid."))?
+                                        )
+                                    } else if ident_str.starts_with('$') {
+                                        OutgoingKind::Push(ident_str[1..].as_bytes().to_vec())
                                     } else {
-                                        (server_str, false)
-                                    }
+                                        OutgoingKind::Normal(
+                                            Box::<ServerName>::try_from(ident_str)
+                                                .map_err(|_| Error::bad_database("ServerName in servernamepduid is invalid."))?
+                                        )
                                     })
-                                .and_then(|(server_str, is_appservice)| Box::<ServerName>::try_from(server_str)
-                                    .map_err(|_| Error::bad_database("ServerName in servernamepduid is invalid.")).map(|s| (s, is_appservice)))
-                                .ok()
-                                .and_then(|(server, is_appservice)| parts
+                                })
+                                .and_then(|outgoing_kind| parts
                                     .next()
                                     .ok_or_else(|| Error::bad_database("Invalid servernamepduid in db."))
-                                    .ok()
-                                    .map(|pdu_id| (server, is_appservice, pdu_id))
+                                    .map(|pdu_id| (outgoing_kind, pdu_id))
                                 )
-                                .filter(|(server, is_appservice, _)| {
-                                    if last_failed_try.get(server).map_or(false, exponential_backoff) {
+                                .ok()
+                                .filter(|(outgoing_kind, _)| {
+                                    if last_failed_try.get(outgoing_kind).map_or(false, exponential_backoff) {
                                         return false;
                                     }
 
-                                    let mut prefix = if *is_appservice {
-                                        b"+".to_vec()
-                                    } else {
-                                        Vec::new()
+                                    let mut prefix = match outgoing_kind {
+                                        OutgoingKind::Appservice(serv) => {
+                                            let mut p = b"+".to_vec();
+                                            p.extend_from_slice(serv.as_bytes());
+                                            p
+                                    },
+                                        OutgoingKind::Push(id) => {
+                                            let mut p = b"$".to_vec();
+                                            p.extend_from_slice(&id);
+                                            p
+                                        },
+                                        OutgoingKind::Normal(serv) => {
+                                            let mut p = vec![];
+                                            p.extend_from_slice(serv.as_bytes());
+                                            p
+                                        },
                                     };
-                                    prefix.extend_from_slice(server.as_bytes());
                                     prefix.push(0xff);
 
                                     servercurrentpdus
@@ -224,7 +279,13 @@ impl Sending {
                                 servercurrentpdus.insert(&key, &[]).unwrap();
                                 servernamepduids.remove(&key).unwrap();
 
-                                futures.push(Self::handle_event(server, is_appservice, vec![pdu_id.into()], &globals, &rooms, &appservice, &maximum_requests));
+                                futures.push(
+                                    Self::handle_event(
+                                        outgoing_kind,
+                                        vec![pdu_id.into()],
+                                        &db,
+                                    )
+                                );
                             }
                         }
                     }
@@ -233,6 +294,25 @@ impl Sending {
         });
     }
 
+    #[tracing::instrument(skip(self))]
+    pub fn send_push_pdu(&self, pdu_id: &[u8]) -> Result<()> {
+        // Make sure we don't cause utf8 errors when parsing to a String...
+        let pduid = String::from_utf8_lossy(pdu_id).as_bytes().to_vec();
+
+        // these are valid ServerName chars
+        // (byte.is_ascii_alphanumeric() || byte == b'-' || byte == b'.')
+        let mut key = b"$".to_vec();
+        // keep each pdu push unique
+        key.extend_from_slice(pduid.as_slice());
+        key.push(0xff);
+        key.extend_from_slice(pdu_id);
+        self.servernamepduids.insert(key, b"")?;
+
+        println!("AAAA");
+
+        Ok(())
+    }
+
     #[tracing::instrument(skip(self))]
     pub fn send_pdu(&self, server: &ServerName, pdu_id: &[u8]) -> Result<()> {
         let mut key = server.as_bytes().to_vec();
@@ -262,29 +342,26 @@ impl Sending {
         hash.as_ref().to_owned()
     }
 
-    #[tracing::instrument(skip(globals, rooms, appservice))]
+    #[tracing::instrument(skip(db))]
     async fn handle_event(
-        server: Box<ServerName>,
-        is_appservice: bool,
+        kind: OutgoingKind,
         pdu_ids: Vec<IVec>,
-        globals: &super::globals::Globals,
-        rooms: &super::rooms::Rooms,
-        appservice: &super::appservice::Appservice,
-        maximum_requests: &Semaphore,
-    ) -> std::result::Result<(Box<ServerName>, bool), (Box<ServerName>, bool, Error)> {
-        if is_appservice {
+        db: &Database,
+    ) -> std::result::Result<OutgoingKind, (OutgoingKind, Error)> {
+        match dbg!(&kind) {
+            OutgoingKind::Appservice(server) => {
                 let pdu_jsons = pdu_ids
                     .iter()
                     .map(|pdu_id| {
                         Ok::<_, (Box<ServerName>, Error)>(
-                        rooms
+                            db.rooms
                                 .get_pdu_from_id(pdu_id)
                                 .map_err(|e| (server.clone(), e))?
                                 .ok_or_else(|| {
                                     (
                                         server.clone(),
                                         Error::bad_database(
-                                        "Event in servernamepduids not found in db.",
+                                            "[Appservice] Event in servernamepduids not found in ",
                                         ),
                                     )
                                 })?
@@ -293,11 +370,10 @@ impl Sending {
                     })
                     .filter_map(|r| r.ok())
                     .collect::<Vec<_>>();
-
-            let permit = maximum_requests.acquire().await;
+                let permit = db.sending.maximum_requests.acquire().await;
                 let response = appservice_server::send_request(
-                &globals,
-                appservice
+                    &db.globals,
+                    db.appservice
                         .get_registration(server.as_str())
                         .unwrap()
                         .unwrap(), // TODO: handle error
@@ -310,28 +386,115 @@ impl Sending {
                     },
                 )
                 .await
-            .map(|_response| (server.clone(), is_appservice))
-            .map_err(|e| (server, is_appservice, e));
+                .map(|_response| kind.clone())
+                .map_err(|e| (kind, e));
 
                 drop(permit);
 
                 response
+            }
+            OutgoingKind::Push(id) => {
+                let pdus = pdu_ids
+                    .iter()
+                    .map(|pdu_id| {
+                        Ok::<_, (Vec<u8>, Error)>(
+                            db.rooms
+                                .get_pdu_from_id(pdu_id)
+                                .map_err(|e| (id.clone(), e))?
+                                .ok_or_else(|| {
+                                    (
+                                        id.clone(),
+                                        Error::bad_database(
+                                            "[Push] Event in servernamepduids not found in db.",
+                                        ),
+                                    )
+                                })?,
+                        )
+                    })
+                    .filter_map(|r| r.ok())
+                    .collect::<Vec<_>>();
+
+                for pdu in dbg!(&pdus) {
+                    // Redacted events are not notification targets (we don't send push for them)
+                    if pdu.unsigned.get("redacted_because").is_some() {
+                        continue;
+                    }
+
+                    for user in db.rooms.room_members(&pdu.room_id) {
+                        let user = user.map_err(|e| (OutgoingKind::Push(id.clone()), e))?;
+
+                        // Don't notify the user of their own events
+                        if user == pdu.sender {
+                            continue;
+                        }
+
+                        let pushers = dbg!(db.pusher.get_pusher(&user))
+                            .map_err(|e| (OutgoingKind::Push(id.clone()), e))?;
+
+                        let rules_for_user = db
+                            .account_data
+                            .get::<push_rules::PushRulesEvent>(None, &user, EventType::PushRules)
+                            .map_err(|e| (OutgoingKind::Push(id.clone()), e))?
+                            .map(|ev| ev.content.global)
+                            .unwrap_or_else(|| crate::push_rules::default_pushrules(&user));
+
+                        let unread: UInt = if let Some(last_read) = db
+                            .rooms
+                            .edus
+                            .private_read_get(&pdu.room_id, &user)
+                            .map_err(|e| (OutgoingKind::Push(id.clone()), e))?
+                        {
+                            (db.rooms
+                                .pdus_since(&user, &pdu.room_id, last_read)
+                                .map_err(|e| (OutgoingKind::Push(id.clone()), e))?
+                                .filter_map(|pdu| pdu.ok()) // Filter out buggy events
+                                .filter(|(_, pdu)| {
+                                    matches!(
+                                        pdu.kind.clone(),
+                                        EventType::RoomMessage | EventType::RoomEncrypted
+                                    )
+                                })
+                                .count() as u32)
+                                .into()
                         } else {
+                            // Just return zero unread messages
+                            uint!(0)
+                        };
+
+                        let permit = db.sending.maximum_requests.acquire().await;
+                        let _response = pusher::send_push_notice(
+                            &user,
+                            unread,
+                            &pushers,
+                            rules_for_user,
+                            pdu,
+                            db,
+                        )
+                        .await
+                        .map(|_response| kind.clone())
+                        .map_err(|e| (kind.clone(), e));
+
+                        drop(permit);
+                    }
+                }
+                Ok(OutgoingKind::Push(id.clone()))
+            }
+            OutgoingKind::Normal(server) => {
                 let pdu_jsons = pdu_ids
                     .iter()
                     .map(|pdu_id| {
-                    Ok::<_, (Box<ServerName>, Error)>(
+                        Ok::<_, (OutgoingKind, Error)>(
                             // TODO: check room version and remove event_id if needed
                             serde_json::from_str(
                                 PduEvent::convert_to_outgoing_federation_event(
-                                rooms
+                                    db.rooms
                                         .get_pdu_json_from_id(pdu_id)
-                                    .map_err(|e| (server.clone(), e))?
+                                        .map_err(|e| (OutgoingKind::Normal(server.clone()), e))?
                                         .ok_or_else(|| {
                                             (
-                                            server.clone(),
+                                                OutgoingKind::Normal(server.clone()),
                                                 Error::bad_database(
-                                                "Event in servernamepduids not found in db.",
+                                                    "[Normal] Event in servernamepduids not found in db.",
                                                 ),
                                             )
                                         })?,
@@ -345,14 +508,14 @@ impl Sending {
                     .filter_map(|r| r.ok())
                     .collect::<Vec<_>>();
 
-            let permit = maximum_requests.acquire().await;
+                let permit = db.sending.maximum_requests.acquire().await;
 
                 info!("sending pdus to {}: {:#?}", server, pdu_jsons);
                 let response = server_server::send_request(
-                &globals,
+                    &db.globals,
                     &*server,
                     send_transaction_message::v1::Request {
-                    origin: globals.server_name(),
+                        origin: db.globals.server_name(),
                         pdus: &pdu_jsons,
                         edus: &[],
                         origin_server_ts: SystemTime::now(),
@@ -365,19 +528,19 @@ impl Sending {
                 .await
                 .map(|response| {
                     info!("server response: {:?}", response);
-                (server.clone(), is_appservice)
+                    kind.clone()
                 })
-            .map_err(|e| (server, is_appservice, e));
+                .map_err(|e| (kind, e));
 
                 drop(permit);
 
                 response
             }
         }
+    }
 
-    fn parse_servercurrentpdus(key: IVec) -> Result<(IVec, Box<ServerName>, IVec, bool)> {
-        let key2 = key.clone();
-        let mut parts = key2.splitn(2, |&b| b == 0xff);
+    fn parse_servercurrentpdus(key: &IVec) -> Result<(OutgoingKind, IVec)> {
+        let mut parts = key.splitn(2, |&b| b == 0xff);
         let server = parts.next().expect("splitn always returns one element");
         let pdu = parts
             .next()
@@ -388,20 +551,26 @@ impl Sending {
         })?;
 
         // Appservices start with a plus
-        let (server, is_appservice) = if server.starts_with('+') {
-            (&server[1..], true)
-        } else {
-            (&*server, false)
-        };
-
-        Ok::<_, Error>((
-            key,
-            Box::<ServerName>::try_from(server).map_err(|_| {
+        Ok::<_, Error>(if server.starts_with('+') {
+            (
+                OutgoingKind::Appservice(Box::<ServerName>::try_from(server).map_err(|_| {
                     Error::bad_database("Invalid server string in server_currenttransaction")
-            })?,
+                })?),
                 IVec::from(pdu),
-            is_appservice,
-        ))
+            )
+        } else if server.starts_with('$') {
+            (
+                OutgoingKind::Push(server.as_bytes().to_vec()),
+                IVec::from(pdu),
+            )
+        } else {
+            (
+                OutgoingKind::Normal(Box::<ServerName>::try_from(server).map_err(|_| {
+                    Error::bad_database("Invalid server string in server_currenttransaction")
+                })?),
+                IVec::from(pdu),
+            )
+        })
     }
 
     #[tracing::instrument(skip(self, globals))]

src/main.rs

@@ -19,6 +19,7 @@ pub use rocket::State;
 use ruma::api::client::error::ErrorKind;
 pub use ruma_wrapper::{ConduitResult, Ruma, RumaResponse};
 
+use log::LevelFilter;
 use rocket::{
     catch, catchers,
     fairing::AdHoc,
@@ -155,6 +156,7 @@ fn setup_rocket() -> (rocket::Rocket, Config) {
                 client_server::get_key_changes_route,
                 client_server::get_pushers_route,
                 client_server::set_pushers_route,
+                // client_server::third_party_route,
                 client_server::upgrade_room_route,
                 server_server::get_server_version_route,
                 server_server::get_server_keys_route,
@@ -178,8 +180,7 @@ fn setup_rocket() -> (rocket::Rocket, Config) {
                 .await
                 .expect("config is valid");
 
-            data.sending
-                .start_handler(&data.globals, &data.rooms, &data.appservice);
+            data.sending.start_handler(&data);
 
             Ok(rocket.manage(data))
         }));