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

tuwunel_api_admin/rooms/
purge_history.rs
