magmaus3/continuwuity
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

modernize remove_to_device_events

Browse source 
Signed-off-by: Jason Volk <jason@zemos.net>
This commit is contained in:
Jason Volk 2025-02-02 08:59:14 +00:00
parent ffe3b0faf2
commit a774afe837
1 changed files with 17 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

43
src/service/users/mod.rs
View file

@ -1,12 +1,12 @@
use std::{collections::BTreeMap, mem, mem::size_of, sync::Arc}; use std::{collections::BTreeMap, mem, sync::Arc};
use conduwuit::{ use conduwuit::{
debug_warn, err, trace, debug_warn, err, trace,
utils::{self, stream::TryIgnore, string::Unquoted, ReadyExt}, utils::{self, stream::TryIgnore, string::Unquoted, ReadyExt},
Err, Error, Result, Server, Err, Error, Result, Server,
}; };
use database::{Database, Deserialized, Ignore, Interfix, Json, Map}; use database::{Deserialized, Ignore, Interfix, Json, Map};
use futures::{FutureExt, Stream, StreamExt, TryFutureExt}; use futures::{Stream, StreamExt, TryFutureExt};
use ruma::{ use ruma::{
api::client::{device::Device, error::ErrorKind, filter::FilterDefinition}, api::client::{device::Device, error::ErrorKind, filter::FilterDefinition},
encryption::{CrossSigningKey, DeviceKeys, OneTimeKey}, encryption::{CrossSigningKey, DeviceKeys, OneTimeKey},
@ -28,7 +28,6 @@ pub struct Service {
struct Services { struct Services {
server: Arc<Server>, server: Arc<Server>,
db: Arc<Database>,
account_data: Dep<account_data::Service>, account_data: Dep<account_data::Service>,
admin: Dep<admin::Service>, admin: Dep<admin::Service>,
globals: Dep<globals::Service>, globals: Dep<globals::Service>,
@ -64,7 +63,6 @@ impl crate::Service for Service {
Ok(Arc::new(Self { Ok(Arc::new(Self {
services: Services { services: Services {
server: args.server.clone(), server: args.server.clone(),
db: args.db.clone(),
account_data: args.depend::<account_data::Service>("account_data"), account_data: args.depend::<account_data::Service>("account_data"),
admin: args.depend::<admin::Service>("admin"), admin: args.depend::<admin::Service>("admin"),
globals: args.depend::<globals::Service>("globals"), globals: args.depend::<globals::Service>("globals"),
@ -801,35 +799,28 @@ impl Service {
.map(|(_, val): (Ignore, Raw<AnyToDeviceEvent>)| val) .map(|(_, val): (Ignore, Raw<AnyToDeviceEvent>)| val)
} }
pub async fn remove_to_device_events( pub async fn remove_to_device_events<Until>(
&self, &self,
user_id: &UserId, user_id: &UserId,
device_id: &DeviceId, device_id: &DeviceId,
until: u64, until: Until,
) { ) where
let mut prefix = user_id.as_bytes().to_vec(); Until: Into<Option<u64>> + Send,
prefix.push(0xFF); {
prefix.extend_from_slice(device_id.as_bytes()); type Key<'a> = (&'a UserId, &'a DeviceId, u64);
prefix.push(0xFF);
let mut last = prefix.clone(); let until = until.into().unwrap_or(u64::MAX);
last.extend_from_slice(&until.to_be_bytes()); let from = (user_id, device_id, until);
let _cork = self.services.db.cork_and_flush();
self.db self.db
.todeviceid_events .todeviceid_events
.rev_raw_keys_from(&last) // this includes last .rev_keys_from(&from)
.ignore_err() .ignore_err()
.ready_take_while(move |key| key.starts_with(&prefix)) .ready_take_while(move |(user_id_, device_id_, _): &Key<'_>| {
.map(|key| { user_id == *user_id_ && device_id == *device_id_
let len = key.len(); })
let start = len.saturating_sub(size_of::<u64>()); .ready_for_each(|key: Key<'_>| {
let count = utils::u64_from_u8(&key[start..len]); self.db.todeviceid_events.del(key);
(key, count)
}) })
.ready_take_while(move |(_, count)| *count <= until)
.ready_for_each(|(key, _)| self.db.todeviceid_events.remove(&key))
.boxed()
.await; .await;
} }