tuwunel_api/client/admin/rooms/
purge_history.rs

1use std::pin::pin;
2
3use axum::extract::State;
4use futures::StreamExt;
5use ruma::{EventId, MilliSecondsSinceUnixEpoch, OwnedRoomId, RoomId, api::Direction};
6use synapse_admin_api::purge_history::{
7	purge::{
8		by_event::{Request as PurgeByEventRequest, Response as PurgeByEventResponse},
9		v1::{Request as PurgeRequest, Response as PurgeResponse},
10	},
11	status::v1::{PurgeStatus, Request as StatusRequest, Response as StatusResponse},
12};
13use tuwunel_core::{Err, Result, err, matrix::pdu::PduCount};
14use tuwunel_service::tasks::Status;
15
16use crate::{Ruma, client::admin::require_admin};
17
18/// # `POST /_synapse/admin/v1/purge_history/{room_id}`
19///
20/// Schedules a background purge of the room's history strictly before the given
21/// event or timestamp, returning the purge task id.
22pub(crate) async fn admin_purge_history_route(
23	State(services): State<crate::State>,
24	body: Ruma<PurgeRequest>,
25) -> Result<PurgeResponse> {
26	require_admin(&services, body.sender_user()).await?;
27
28	let boundary = resolve_boundary(
29		&services,
30		&body.room_id,
31		body.purge_up_to_event_id.as_deref(),
32		body.purge_up_to_ts,
33	)
34	.await?;
35
36	let purge_id =
37		schedule_purge(services, body.room_id.clone(), boundary, body.delete_local_events);
38
39	Ok(PurgeResponse { purge_id })
40}
41
42/// # `POST /_synapse/admin/v1/purge_history/{room_id}/{event_id}`
43///
44/// Purge variant that takes the boundary event in the path.
45pub(crate) async fn admin_purge_history_by_event_route(
46	State(services): State<crate::State>,
47	body: Ruma<PurgeByEventRequest>,
48) -> Result<PurgeByEventResponse> {
49	require_admin(&services, body.sender_user()).await?;
50
51	let boundary = resolve_boundary(&services, &body.room_id, Some(&body.event_id), None).await?;
52
53	let purge_id =
54		schedule_purge(services, body.room_id.clone(), boundary, body.delete_local_events);
55
56	Ok(PurgeByEventResponse { purge_id })
57}
58
59/// # `GET /_synapse/admin/v1/purge_history_status/{purge_id}`
60///
61/// Reports the stage of a history purge. A still-scheduled purge is reported as
62/// active, matching Synapse.
63pub(crate) async fn admin_purge_history_status_route(
64	State(services): State<crate::State>,
65	body: Ruma<StatusRequest>,
66) -> Result<StatusResponse> {
67	require_admin(&services, body.sender_user()).await?;
68
69	let task = services
70		.tasks
71		.get(&body.purge_id)
72		.filter(|task| task.action == super::PURGE_HISTORY_ACTION)
73		.ok_or_else(|| err!(Request(NotFound("Unknown purge task"))))?;
74
75	Ok(StatusResponse {
76		status: purge_status(task.status),
77		error: task.error,
78	})
79}
80
81/// Resolves the exclusive purge boundary from the boundary event or timestamp,
82/// erroring when neither is given, the event is unknown or belongs to another
83/// room, or no event precedes the timestamp.
84async fn resolve_boundary(
85	services: &crate::State,
86	room_id: &RoomId,
87	event_id: Option<&EventId>,
88	ts: Option<MilliSecondsSinceUnixEpoch>,
89) -> Result<PduCount> {
90	if let Some(event_id) = event_id {
91		let pdu = services
92			.timeline
93			.get_pdu(event_id)
94			.await
95			.map_err(|_| err!(Request(NotFound("Event not found"))))?;
96
97		if pdu.room_id != *room_id {
98			return Err!(Request(BadJson("Event is for wrong room")));
99		}
100
101		return services
102			.timeline
103			.get_pdu_count(event_id)
104			.await
105			.map_err(|_| err!(Request(NotFound("Event not found"))));
106	}
107
108	let Some(ts) = ts else {
109		return Err!(Request(BadJson(
110			"One of purge_up_to_event_id or purge_up_to_ts must be provided"
111		)));
112	};
113
114	let events = services
115		.timeline
116		.pdus_near_ts(None, room_id, ts, Direction::Backward);
117
118	let mut events = pin!(events);
119
120	events
121		.next()
122		.await
123		.transpose()?
124		.map(|(count, _)| count)
125		.ok_or_else(|| err!(Request(NotFound("No event found before the given timestamp"))))
126}
127
128/// Spawns the purge on the tasks service, returning its id. Takes `services` by
129/// value (it is `Copy`) so the detached task owns a `'static` handle.
130fn schedule_purge(
131	services: crate::State,
132	room_id: OwnedRoomId,
133	boundary: PduCount,
134	delete_local_events: bool,
135) -> String {
136	let resource_id = room_id.to_string();
137
138	let work = async move {
139		let purged = services
140			.timeline
141			.purge_history(&room_id, boundary, delete_local_events)
142			.await?;
143
144		Ok(serde_json::json!({ "purged": purged }))
145	};
146
147	services
148		.tasks
149		.spawn(super::PURGE_HISTORY_ACTION, resource_id, work)
150		.to_string()
151}
152
153fn purge_status(status: Status) -> PurgeStatus {
154	match status {
155		| Status::Scheduled | Status::Active => PurgeStatus::Active,
156		| Status::Complete => PurgeStatus::Complete,
157		| Status::Failed => PurgeStatus::Failed,
158	}
159}
160
161#[cfg(test)]
162mod tests {
163	use serde_json::json;
164	use tuwunel_service::tasks::Status;
165
166	use super::purge_status;
167
168	#[test]
169	fn collapses_scheduled_into_active() {
170		let status = |status| serde_json::to_value(purge_status(status)).unwrap();
171
172		assert_eq!(status(Status::Scheduled), json!("active"));
173		assert_eq!(status(Status::Active), json!("active"));
174		assert_eq!(status(Status::Complete), json!("complete"));
175		assert_eq!(status(Status::Failed), json!("failed"));
176	}
177}
tuwunel_api/client/admin/rooms/purge_history.rs

tuwunel_api/client/admin/rooms/
purge_history.rs
