tuwunel_service/rooms/timeline/
purge.rs

1//! Selectively removes historical accepted events from a room timeline.
2//!
3//! Purging preserves state events and leaves forward-extremity mappings
4//! untouched while coordinating deletion from search, relation, and
5//! redaction-retention indexes. Selection follows room stream order rather
6//! than graph depth.
7
8use futures::TryStreamExt;
9use ruma::{RoomId, api::Direction, events::TimelineEventType};
10use tuwunel_core::{
11	Result, implement,
12	matrix::{
13		Event,
14		pdu::{PduCount, PduEvent},
15	},
16	trace,
17	utils::stream::TryReadyExt,
18};
19
20use super::{ExtractBody, RawPduId, bias_count};
21
22/// Purges eligible room history strictly before a timeline count.
23///
24/// State events are preserved, and locally sent events remain unless
25/// `delete_local_events` is set. Forward extremities are untouched, while
26/// search, relation, and retained-original data are removed for each deleted
27/// event; an error can stop the operation after partial progress.
28#[implement(super::Service)]
29pub async fn purge_history(
30	&self,
31	room_id: &RoomId,
32	until: PduCount,
33	delete_local_events: bool,
34) -> Result<usize> {
35	let shortroomid = self
36		.services
37		.short
38		.get_shortroomid(room_id)
39		.await?;
40
41	let start = self
42		.count_to_id(room_id, PduCount::min(), Direction::Forward)
43		.await?;
44
45	let prefix = start.shortroomid();
46
47	self.db
48		.pduid_pdu
49		.raw_stream_from(&start)
50		.ready_try_take_while(move |kv| {
51			let (key, _) = *kv;
52			Ok(key.starts_with(&prefix) && RawPduId::from(key).pdu_count() < until)
53		})
54		.try_fold(0_usize, async |purged, (key, value)| {
55			let pdu = serde_json::from_slice::<PduEvent>(value)?;
56
57			if pdu.state_key.is_some()
58				|| (!delete_local_events && self.services.globals.user_is_local(&pdu.sender))
59			{
60				return Ok(purged);
61			}
62
63			let mut txn = self.db.db.txn();
64
65			let raw_id = RawPduId::from(key);
66			let count = raw_id.pdu_count();
67			let event_id = pdu.event_id.clone();
68			let ts: u64 = pdu.origin_server_ts.into();
69
70			txn.del_raw(&self.db.pduid_pdu, key);
71			txn.del_raw(&self.db.eventid_pduid, &event_id);
72			txn.del_raw(&self.db.eventid_outlierpdu, &event_id);
73
74			let room_id_ts_id = (room_id, ts, bias_count(raw_id.count()));
75			txn.del(&self.db.roomid_tscount_pducount, room_id_ts_id);
76
77			txn.execute();
78
79			if pdu.kind == TimelineEventType::RoomMessage
80				&& let Ok(ExtractBody { body: Some(body) }) = pdu.get_content()
81			{
82				self.services
83					.search
84					.deindex_pdu(shortroomid, &raw_id, &body);
85			}
86
87			self.services
88				.pdu_metadata
89				.purge_event_relations(shortroomid, count, room_id, &event_id)
90				.await;
91
92			self.services.retention.purge_original(&event_id);
93
94			trace!(?event_id, ?room_id, "Purged");
95
96			Ok(purged.saturating_add(1))
97		})
98		.await
99}
tuwunel_service/rooms/timeline/purge.rs

tuwunel_service/rooms/timeline/
purge.rs
