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			services.server.progress.advance();
31
32			let (room_id, _, user_id) = key;
33			let last_room = cur_room.replace(
34				room_id
35					.as_str()
36					.try_into()
37					.expect("invalid room_id in database"),
38			);
39
40			let last_user = cur_user.replace(
41				user_id
42					.as_str()
43					.try_into()
44					.expect("invalid user_id in database"),
45			);
46
47			let is_dup = cur_room == last_room && cur_user == last_user;
48			if is_dup {
49				readreceiptid_readreceipt.del(key);
50			}
51
52			fixed = fixed.saturating_add(is_dup.into());
53			total = total.saturating_add(1);
54		})
55		.await;
56
57	drop(cork);
58	info!(?total, ?fixed, "Fixed undeleted entries in readreceiptid_readreceipt.");
59
60	db["global"].insert(b"fix_readreceiptid_readreceipt_duplicates", []);
61	readreceiptid_readreceipt.sort()
62}

tuwunel_service/migrations/fix_readreceiptid_readreceipt_duplicates.rs

tuwunel_service/migrations/
fix_readreceiptid_readreceipt_duplicates.rs