tuwunel_service/migrations/
fix_readreceiptid_readreceipt_duplicates.rs

1use ruma::{RoomId, UserId, identifiers_validation::ID_MAX_BYTES};
2use tuwunel_core::{
3	Result,
4	arrayvec::ArrayString,
5	info,
6	utils::{ReadyExt, stream::TryExpect},
7	warn,
8};
9
10use crate::Services;
11
12type ArrayId = ArrayString<ID_MAX_BYTES>;
13type Key<'a> = (&'a RoomId, u64, &'a UserId);
14
15pub(super) async fn fix_readreceiptid_readreceipt_duplicates(services: &Services) -> Result {
16	warn!("Fixing undeleted entries in readreceiptid_readreceipt...");
17
18	let db = &services.db;
19	let cork = db.cork_and_sync();
20	let readreceiptid_readreceipt = db["readreceiptid_readreceipt"].clone();
21
22	let mut cur_room: Option<ArrayId> = None;
23	let mut cur_user: Option<ArrayId> = None;
24	let (mut total, mut fixed): (usize, usize) = (0, 0);
25
26	readreceiptid_readreceipt
27		.keys()
28		.expect_ok()
29		.ready_for_each(|key: Key<'_>| {
30			let (room_id, _, user_id) = key;
31			let last_room = cur_room.replace(
32				room_id
33					.as_str()
34					.try_into()
35					.expect("invalid room_id in database"),
36			);
37
38			let last_user = cur_user.replace(
39				user_id
40					.as_str()
41					.try_into()
42					.expect("invalid user_id in database"),
43			);
44
45			let is_dup = cur_room == last_room && cur_user == last_user;
46			if is_dup {
47				readreceiptid_readreceipt.del(key);
48			}
49
50			fixed = fixed.saturating_add(is_dup.into());
51			total = total.saturating_add(1);
52		})
53		.await;
54
55	drop(cork);
56	info!(?total, ?fixed, "Fixed undeleted entries in readreceiptid_readreceipt.");
57
58	db["global"].insert(b"fix_readreceiptid_readreceipt_duplicates", []);
59	readreceiptid_readreceipt.sort()
60}

tuwunel_service/migrations/fix_readreceiptid_readreceipt_duplicates.rs

tuwunel_service/migrations/
fix_readreceiptid_readreceipt_duplicates.rs