tuwunel_api_client/delayed_events/
delay_query.rs

1use std::time::Duration;
2
3use axum::{
4	extract::{FromRequestParts, State},
5	response::{IntoResponse, Response},
6};
7use futures::FutureExt;
8use http::request::Parts;
9use ruma::api::client::{
10	delayed_events::{
11		delayed_message_event::unstable::Response as DelayedMessageResponse,
12		delayed_state_event::unstable::Response as DelayedStateResponse,
13	},
14	message::send_message_event,
15	state::send_state_event,
16};
17use tuwunel_core::{Error, Result, err};
18use url::form_urlencoded::parse as parse_query;
19
20use super::{Kind, Schedule, schedule};
21use crate::{
22	Ruma, RumaResponse, send::send_message_event_route, state::send_state_event_for_key_route,
23};
24
25pub(super) const DELAY_PARAM: &str = "org.matrix.msc4140.delay";
26
27/// The `org.matrix.msc4140.delay` query parameter, in milliseconds.
28///
29/// Absent when the request does not carry it. Present with no value or a value
30/// that is not a whole number, it refuses the request rather than sending the
31/// event at once.
32pub(crate) struct DelayQuery(Option<u64>);
33
34/// # `PUT /_matrix/client/v3/rooms/{roomId}/send/{eventType}/{txnId}`
35///
36/// Sends a message event, or schedules it to be sent later (MSC4140).
37///
38/// A request carrying `org.matrix.msc4140.delay` schedules the event and
39/// answers with its `delay_id`; any other goes to the ordinary send route.
40#[tracing::instrument(level = "debug", skip_all)]
41pub(crate) async fn send_message_event_or_delayed_route(
42	State(services): State<crate::State>,
43	DelayQuery(delay): DelayQuery,
44	body: Ruma<send_message_event::v3::Request>,
45) -> Result<Response> {
46	let Some(delay) = delay else {
47		return send_message_event_route(State(services), body)
48			.boxed() // size firewall
49			.await
50			.map(|response| RumaResponse(response).into_response());
51	};
52
53	let kind = Kind::Message {
54		event_type: &body.event_type,
55		content: &body.body.body,
56	};
57
58	let request = Schedule {
59		sender: body.sender_user(),
60		device: body.sender_device.as_deref(),
61		room_id: &body.room_id,
62		kind,
63		delay: Duration::from_millis(delay),
64		txn_id: Some(&body.txn_id),
65		timestamp: body.appservice_info.as_ref().and(body.timestamp),
66		stable: false,
67	};
68
69	let delay_id = schedule(&services, request)
70		.boxed() // size firewall
71		.await?;
72
73	let response = DelayedMessageResponse::new(delay_id.into_string());
74
75	Ok(RumaResponse(response).into_response())
76}
77
78/// # `PUT /_matrix/client/v3/rooms/{roomId}/state/{eventType}/{stateKey}`
79///
80/// Sends a state event, or schedules it to be sent later (MSC4140).
81///
82/// A request carrying `org.matrix.msc4140.delay` schedules the event and
83/// answers with its `delay_id`; any other goes to the ordinary state route.
84/// It also serves the empty-key paths.
85#[tracing::instrument(level = "debug", skip_all)]
86pub(crate) async fn send_state_event_or_delayed_route(
87	State(services): State<crate::State>,
88	DelayQuery(delay): DelayQuery,
89	body: Ruma<send_state_event::v3::Request>,
90) -> Result<Response> {
91	let Some(delay) = delay else {
92		return send_state_event_for_key_route(State(services), body)
93			.boxed() // size firewall
94			.await
95			.map(|response| RumaResponse(response).into_response());
96	};
97
98	let kind = Kind::State {
99		event_type: &body.event_type,
100		state_key: &body.state_key,
101		content: &body.body.body,
102	};
103
104	let request = Schedule {
105		sender: body.sender_user(),
106		device: body.sender_device.as_deref(),
107		room_id: &body.room_id,
108		kind,
109		delay: Duration::from_millis(delay),
110		txn_id: None,
111		timestamp: body.appservice_info.as_ref().and(body.timestamp),
112		stable: false,
113	};
114
115	let delay_id = schedule(&services, request)
116		.boxed() // size firewall
117		.await?;
118
119	let response = DelayedStateResponse::new(delay_id.into_string());
120
121	Ok(RumaResponse(response).into_response())
122}
123
124impl FromRequestParts<crate::State> for DelayQuery {
125	type Rejection = Error;
126
127	async fn from_request_parts(
128		parts: &mut Parts,
129		_services: &crate::State,
130	) -> Result<Self, Self::Rejection> {
131		parse_delay(parts.uri.query().unwrap_or_default()).map(Self)
132	}
133}
134
135/// Reads the delay parameter from a raw query string.
136///
137/// When the parameter repeats, the first occurrence wins.
138pub(super) fn parse_delay(query: &str) -> Result<Option<u64>> {
139	parse_query(query.as_bytes())
140		.find(|(key, _)| key == DELAY_PARAM)
141		.map(|(_, value)| {
142			value
143				.parse()
144				.map_err(|e| err!(Request(InvalidParam("Invalid {DELAY_PARAM}: {e}"))))
145		})
146		.transpose()
147}
tuwunel_api_client/delayed_events/delay_query.rs

tuwunel_api_client/delayed_events/
delay_query.rs
