fix(admin): Own pushkeys before deleting all of a user's pushers

changelog.d/1726.bugfix.md

@@ -0,0 +1 @@
+Fixed `!admin query pusher delete-all-user` reporting success without deleting any pushers. Contributed by @eleboucher.

src/admin/query/pusher.rs

@@ -59,7 +59,7 @@ pub(super) async fn process(subcommand: PusherCommand, context: &Context<'_>) ->
 			pushers
 				.stream()
 				.for_each(async |pushkey| {
-					services.pusher.delete_pusher(&user_id, pushkey).await;
+					services.pusher.delete_pusher(&user_id, &pushkey).await;
 				})
 				.await;
 			write!(context, "Deleted {pusher_count} pushers for {user_id}.")
@@ -68,7 +68,6 @@ pub(super) async fn process(subcommand: PusherCommand, context: &Context<'_>) ->
 			let pushers = services
 				.pusher
 				.get_pushkeys(&user_id)
-				.map(ToOwned::to_owned)
 				.broad_filter_map(async |pushkey| {
 					services
 						.pusher

src/api/client/account/mod.rs

@@ -145,7 +145,6 @@ pub(crate) async fn change_password_route(
 		services
 			.pusher
 			.get_pushkeys(&sender_user)
-			.map(ToOwned::to_owned)
 			.broad_filter_map(async |pushkey| {
 				services
 					.pusher
@@ -339,7 +338,7 @@ pub async fn full_user_deactivate(
 		.pusher
 		.get_pushkeys(user_id)
 		.for_each(async |pushkey| {
-			services.pusher.delete_pusher(user_id, pushkey).await;
+			services.pusher.delete_pusher(user_id, &pushkey).await;
 		})
 		.await;
 

src/api/client/session.rs

@@ -298,7 +298,6 @@ pub(crate) async fn logout_route(
 	services
 		.pusher
 		.get_pushkeys(sender_user)
-		.map(ToOwned::to_owned)
 		.broad_filter_map(async |pushkey| {
 			services
 				.pusher
@@ -345,7 +344,7 @@ pub(crate) async fn logout_all_route(
 		.pusher
 		.get_pushkeys(sender_user)
 		.for_each(async |pushkey| {
-			services.pusher.delete_pusher(sender_user, pushkey).await;
+			services.pusher.delete_pusher(sender_user, &pushkey).await;
 		})
 		.await;
 

src/service/pusher/mod.rs

@@ -182,13 +182,14 @@ pub async fn get_pushers(&self, sender: &UserId) -> Vec<Pusher> {
 	pub fn get_pushkeys<'a>(
 		&'a self,
 		sender: &'a UserId,
-	) -> impl Stream<Item = &'a str> + Send + 'a {
+	) -> impl Stream<Item = String> + Send + 'a {
 		let prefix = (sender, Interfix);
 		self.db
 			.senderkey_pusher
 			.keys_prefix(&prefix)
 			.ignore_err()
 			.map(|(_, pushkey): (Ignore, &str)| pushkey)
+			.map(ToOwned::to_owned)
 	}
 
 	#[tracing::instrument(skip(self, dest, request))]

src/service/rooms/timeline/append.rs

@@ -540,7 +540,7 @@ async fn notify_local_users(&self, pdu: &PduEvent, pdu_id: &RawPduId, room_id: &
 				.ready_for_each(|push_key| {
 					self.services
 						.sending
-						.send_pdu_push(pdu_id, user, push_key.to_owned())
+						.send_pdu_push(pdu_id, user, push_key)
 						.expect("TODO: replace with future");
 				})
 				.await;