asyncify/de-wrap pusher service getters; merge Data into Service

Signed-off-by: Jason Volk <[email protected]>

src/api/client/push.rs

@@ -379,7 +379,7 @@ pub(crate) async fn get_pushers_route(
 	let sender_user = body.sender_user.as_ref().expect("user is authenticated");
 
 	Ok(get_pushers::v3::Response {
-		pushers: services.pusher.get_pushers(sender_user)?,
+		pushers: services.pusher.get_pushers(sender_user).await,
 	})
 }
 
@@ -393,7 +393,7 @@ pub(crate) async fn set_pushers_route(
 ) -> Result<set_pusher::v3::Response> {
 	let sender_user = body.sender_user.as_ref().expect("user is authenticated");
 
-	services.pusher.set_pusher(sender_user, &body.action)?;
+	services.pusher.set_pusher(sender_user, &body.action);
 
 	Ok(set_pusher::v3::Response::default())
 }

src/service/pusher/mod.rs

@@ -1,9 +1,9 @@
-mod data;
-
 use std::{fmt::Debug, mem, sync::Arc};
 
 use bytes::BytesMut;
 use conduit::{debug_error, err, trace, utils::string_from_bytes, warn, Err, PduEvent, Result};
+use database::{Deserialized, Ignore, Interfix, Map};
+use futures::{Stream, StreamExt};
 use ipaddress::IPAddress;
 use ruma::{
 	api::{
@@ -22,12 +22,11 @@ use ruma::{
 	uint, RoomId, UInt, UserId,
 };
 
-use self::data::Data;
 use crate::{client, globals, rooms, users, Dep};
 
 pub struct Service {
-	services: Services,
 	db: Data,
+	services: Services,
 }
 
 struct Services {
@@ -38,37 +37,74 @@ struct Services {
 	users: Dep<users::Service>,
 }
 
+struct Data {
+	senderkey_pusher: Arc<Map>,
+}
+
 impl crate::Service for Service {
 	fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 		Ok(Arc::new(Self {
+			db: Data {
+				senderkey_pusher: args.db["senderkey_pusher"].clone(),
+			},
 			services: Services {
 				globals: args.depend::<globals::Service>("globals"),
 				client: args.depend::<client::Service>("client"),
 				state_accessor: args.depend::<rooms::state_accessor::Service>("rooms::state_accessor"),
 				state_cache: args.depend::<rooms::state_cache::Service>("rooms::state_cache"),
 				users: args.depend::<users::Service>("users"),
 			},
-			db: Data::new(args.db),
 		}))
 	}
 
 	fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
 }
 
 impl Service {
-	pub fn set_pusher(&self, sender: &UserId, pusher: &set_pusher::v3::PusherAction) -> Result<()> {
-		self.db.set_pusher(sender, pusher)
+	pub fn set_pusher(&self, sender: &UserId, pusher: &set_pusher::v3::PusherAction) {
+		match pusher {
+			set_pusher::v3::PusherAction::Post(data) => {
+				let mut key = sender.as_bytes().to_vec();
+				key.push(0xFF);
+				key.extend_from_slice(data.pusher.ids.pushkey.as_bytes());
+				self.db
+					.senderkey_pusher
+					.insert(&key, &serde_json::to_vec(pusher).expect("Pusher is valid JSON value"));
+			},
+			set_pusher::v3::PusherAction::Delete(ids) => {
+				let mut key = sender.as_bytes().to_vec();
+				key.push(0xFF);
+				key.extend_from_slice(ids.pushkey.as_bytes());
+				self.db.senderkey_pusher.remove(&key);
+			},
+		}
 	}
 
-	pub fn get_pusher(&self, sender: &UserId, pushkey: &str) -> Result<Option<Pusher>> {
-		self.db.get_pusher(sender, pushkey)
+	pub async fn get_pusher(&self, sender: &UserId, pushkey: &str) -> Option<Pusher> {
+		let senderkey = (sender, pushkey);
+		self.db
+			.senderkey_pusher
+			.qry(&senderkey)
+			.await
+			.deserialized_json()
 	}
 
-	pub fn get_pushers(&self, sender: &UserId) -> Result<Vec<Pusher>> { self.db.get_pushers(sender) }
+	pub async fn get_pushers(&self, sender: &UserId) -> Vec<Pusher> {
+		let prefix = (sender, Interfix);
+		self.db
+			.senderkey_pusher
+			.stream_prefix(&prefix)
+			.map(|(_, val): (Ignore, &[u8])| serde_json::from_slice(val).expect("Invalid Pusher in db."))
+			.collect()
+			.await
+	}
 
-	#[must_use]
-	pub fn get_pushkeys(&self, sender: &UserId) -> Box<dyn Iterator<Item = Result<String>> + '_> {
-		self.db.get_pushkeys(sender)
+	pub fn get_pushkeys<'a>(&'a self, sender: &'a UserId) -> impl Stream<Item = &str> + 'a {
+		let prefix = (sender, Interfix);
+		self.db
+			.senderkey_pusher
+			.keys_prefix(&prefix)
+			.map(|(_, pushkey): (Ignore, &str)| pushkey)
 	}
 
 	#[tracing::instrument(skip(self, dest, request))]

src/service/rooms/timeline/mod.rs

@@ -13,6 +13,7 @@ use conduit::{
 	utils::{MutexMap, MutexMapGuard},
 	validated, warn, Error, Result, Server,
 };
+use futures::{future, StreamExt};
 use itertools::Itertools;
 use ruma::{
 	api::{client::error::ErrorKind, federation},
@@ -422,11 +423,18 @@ impl Service {
 				highlights.push(user.clone());
 			}
 
-			for push_key in self.services.pusher.get_pushkeys(user) {
-				self.services
-					.sending
-					.send_pdu_push(&pdu_id, user, push_key?)?;
-			}
+			self.services
+				.pusher
+				.get_pushkeys(user)
+				.for_each(|push_key| {
+					self.services
+						.sending
+						.send_pdu_push(&pdu_id, user, push_key.to_owned())
+						.expect("TODO: replace with future");
+
+					future::ready(())
+				})
+				.await;
 		}
 
 		self.db

src/service/sending/sender.rs

@@ -527,12 +527,7 @@ impl Service {
 				}
 			}
 
-			let Some(pusher) = self
-				.services
-				.pusher
-				.get_pusher(userid, pushkey)
-				.map_err(|e| (dest.clone(), e))?
-			else {
+			let Some(pusher) = self.services.pusher.get_pusher(userid, pushkey).await else {
 				continue;
 			};