feat: implement MSC4140 delayed events

Cargo.toml

@@ -71,6 +71,7 @@ features = [
 	"http2",
 	"json",
 	"matched-path",
+	"original-uri",
 	"tokio",
 	"tracing",
 ]
@@ -397,6 +398,7 @@ features = [
     "unstable-msc4075",
     "unstable-msc4121",
     "unstable-msc4125",
+    "unstable-msc4140",
     "unstable-msc4143",
     "unstable-msc4186",
     "unstable-msc4195",

docs/calls/matrix_rtc.md

@@ -8,6 +8,14 @@ supported here, paired with the
 [lk-jwt-service](https://github.com/element-hq/lk-jwt-service) which issues
 the access tokens clients use to join Livekit rooms.
 
+Tuwunel enables MSC4140 delayed events by default. MatrixRTC uses them as a
+heartbeat: if a client loses connectivity before it can leave a call cleanly,
+the homeserver sends its previously scheduled leave event. The defaults allow
+100 scheduled events per user and delays up to 24 hours. Operators can adjust
+these limits with `max_delayed_events_per_user` and
+`max_event_delay_duration`, or set either option to `0` to disable delayed
+events.
+
 This guide shows you how to deploy MatrixRTC/Element Call using Docker and
 Docker Compose, as Livekit only provides prebuilt Docker images. It is
 possible to run Livekit using their installation script, however this method

docs/development/compliance/msc.md

@@ -19,16 +19,16 @@
 ## Counts
 
 - ✅ `yes`: 258
-- 🟨 `partial`: 34
-- ❌ `no`: 446
+- 🟨 `partial`: 35
+- ❌ `no`: 445
 - ⬛ `n/a`: 286
 
 ### Status by inventory bucket
 
 | Inv | yes | partial | no | n/a | total |
 |---|---|---|---|---|---|
 | merged | 186 | 7 | 7 | 59 | 259 |
-| open | 64 | 26 | 399 | 175 | 664 |
+| open | 64 | 27 | 398 | 175 | 664 |
 | closed | 8 | 1 | 40 | 52 | 101 |
 
 ## Merged
@@ -443,7 +443,7 @@ in the [Out of scope](#out-of-scope) section.
 | MSC4145 | ❌ ● | 0/0 | Simple verified accounts | m.verified profile field and endpoint not implemented |
 | MSC4143 | ✅ ◐ | 80/80 | MatrixRTC | GET rtc/transports routed; only HS-side requirement of the MSC |
 | MSC4141 | ❌ ● | 0/0 | Time based notification filtering | time_and_day push rule condition not supported |
-| MSC4140 | ❌ ● | 0/0 | Cancellable delayed events | delayed events endpoints not implemented despite Ruma types |
+| MSC4140 | 🟨 ● | 75/85 | Cancellable delayed events | persistent scheduling, management endpoints, capability, and sender-private IDs; finalised retention and latest authenticated action-path draft pending |
 | MSC4136 | ❌ ● | 0/0 | Shared retry hints between servers | retry_hints in /send_join response not implemented |
 | MSC4128 | ✅ ● | 100/100 | Error on invalid auth where it is optional | invalid token returns error even on optional auth endpoints |
 | MSC4127 | ❌ ● | 0/0 | Removal of query string auth | removal of query string auth not implemented; still accepted |
@@ -1112,4 +1112,3 @@ place of the (uniformly empty) `Correct/Impl` cell.
 | MSC688 | ⬛ ● | closed | Room Summaries (was: Calculate room names server-side) | stub Google doc; room summary work moved to heroes/MSC688 in spec |
 | MSC455 | ⬛ ● | closed | Do we want to specify a matrix:// URI scheme for rooms? (SPEC-5) | [→ MSC2312] stub Google doc; matrix:// URI scheme superseded by matrix: URI (... |
 | MSC441 | ⬛ ● | closed | Support for Reactions / Aggregations | [→ MSC2675/MSC2676] stub-only Google doc; superseded by MSC2675/MSC2676 react... |
-

src/api/client/capabilities.rs

@@ -94,6 +94,19 @@ pub(crate) async fn get_capabilities_route(
 		capabilities.set("im.nheko.msc3664.related_event_match", json!({"enabled": true}))?;
 	}
 
+	// MSC4140: delayed events.
+	if services.config.max_event_delay_duration > 0
+		&& services.config.max_delayed_events_per_user > 0
+	{
+		capabilities.set(
+			"org.matrix.msc4140.delayed_events",
+			json!({
+				"max_delay_ms": services.config.max_event_delay_duration.saturating_mul(1000),
+				"max_scheduled": services.config.max_delayed_events_per_user,
+			}),
+		)?;
+	}
+
 	// MSC4323: advertise admin moderation only to admins; absence implies
 	// neither suspend nor lock is available to the caller.
 	if services

src/api/client/delayed_events.rs

@@ -0,0 +1,215 @@
+//! Client-server endpoints for MSC4140 delayed events.
+
+use axum::{
+	extract::{OriginalUri, State},
+	response::{IntoResponse, Response},
+};
+use ruma::{
+	CanonicalJsonObject,
+	api::client::{
+		delayed_events::{
+			delayed_message_event, delayed_state_event, get_all_delayed_events,
+			get_delayed_event, send_delayed_event, update_delayed_event,
+		},
+		message::send_message_event,
+		state::send_state_event,
+	},
+};
+use tuwunel_core::{Result, err};
+use tuwunel_service::delayed_events::ScheduleParams;
+
+use crate::{ClientIp, Ruma, RumaResponse};
+
+fn parse_content(json: &str) -> Result<CanonicalJsonObject> {
+	serde_json::from_str(json)
+		.map_err(|error| err!(Request(BadJson("Invalid delayed event content: {error}"))))
+}
+
+fn delay_from_query(uri: &http::Uri) -> Result<Option<std::time::Duration>> {
+	let Some(query) = uri.query() else {
+		return Ok(None);
+	};
+
+	query
+		.split('&')
+		.filter_map(|part| part.split_once('='))
+		.find_map(|(key, value)| (key == "org.matrix.msc4140.delay").then_some(value))
+		.map(|value| {
+			value
+				.parse::<u64>()
+				.map(std::time::Duration::from_millis)
+				.map_err(|_| err!(Request(InvalidParam("Invalid org.matrix.msc4140.delay."))))
+		})
+		.transpose()
+}
+
+/// Dispatches the ordinary message-send path and the deprecated MSC4140
+/// query-parameter form, which intentionally share the same URL path.
+pub(crate) async fn send_message_event_or_delayed_route(
+	State(services): State<crate::State>,
+	OriginalUri(uri): OriginalUri,
+	body: Ruma<send_message_event::v3::Request>,
+) -> Result<Response> {
+	if let Some(delay) = delay_from_query(&uri)? {
+		let delay_id = services
+			.delayed_events
+			.schedule(ScheduleParams {
+				user_id: body.sender_user(),
+				device_id: body.sender_device.as_deref(),
+				room_id: body.room_id.clone(),
+				event_type: body.event_type.clone().into(),
+				state_key: None,
+				content: parse_content(body.body.body.json().get())?,
+				txn_id: Some(body.txn_id.clone()),
+				delay,
+			})
+			.await?;
+
+		return Ok(RumaResponse(delayed_message_event::unstable::Response::new(delay_id))
+			.into_response());
+	}
+
+	Ok(RumaResponse(super::send_message_event_route(State(services), body).await?)
+		.into_response())
+}
+
+/// Dispatches the ordinary state-send path and the deprecated MSC4140
+/// query-parameter form.
+pub(crate) async fn send_state_event_or_delayed_route(
+	State(services): State<crate::State>,
+	OriginalUri(uri): OriginalUri,
+	body: Ruma<send_state_event::v3::Request>,
+) -> Result<Response> {
+	if let Some(delay) = delay_from_query(&uri)? {
+		let delay_id = services
+			.delayed_events
+			.schedule(ScheduleParams {
+				user_id: body.sender_user(),
+				device_id: body.sender_device.as_deref(),
+				room_id: body.room_id.clone(),
+				event_type: body.event_type.clone().into(),
+				state_key: Some(body.state_key.clone()),
+				content: parse_content(body.body.body.json().get())?,
+				txn_id: None,
+				delay,
+			})
+			.await?;
+
+		return Ok(
+			RumaResponse(delayed_state_event::unstable::Response::new(delay_id)).into_response()
+		);
+	}
+
+	Ok(
+		RumaResponse(super::send_state_event_for_key_route(State(services), body).await?)
+			.into_response(),
+	)
+}
+
+/// `PUT /_matrix/client/unstable/org.matrix.msc4140/rooms/{room_id}/
+/// delayed_event/{event_type}/{txn_id}`
+pub(crate) async fn send_delayed_event_route(
+	State(services): State<crate::State>,
+	body: Ruma<send_delayed_event::unstable::Request>,
+) -> Result<send_delayed_event::unstable::Response> {
+	let delay_id = services
+		.delayed_events
+		.schedule(ScheduleParams {
+			user_id: body.sender_user(),
+			device_id: body.sender_device.as_deref(),
+			room_id: body.room_id.clone(),
+			event_type: body.event_type.clone(),
+			state_key: body.state_key.clone(),
+			content: parse_content(body.content.json().get())?,
+			txn_id: Some(body.txn_id.clone()),
+			delay: body.delay,
+		})
+		.await?;
+
+	Ok(send_delayed_event::unstable::Response::new(delay_id))
+}
+
+/// `POST /_matrix/client/unstable/org.matrix.msc4140/delayed_events/{delay_id}`
+pub(crate) async fn update_delayed_event_v1_route(
+	State(services): State<crate::State>,
+	ClientIp(client): ClientIp,
+	body: Ruma<update_delayed_event::unstable_v1::Request>,
+) -> Result<update_delayed_event::unstable_v1::Response> {
+	services
+		.delayed_events
+		.update(&body.delay_id, body.action.as_ref(), Some(body.sender_user()), client)
+		.await?;
+
+	Ok(update_delayed_event::unstable_v1::Response::new())
+}
+
+/// `POST /_matrix/client/unstable/org.matrix.msc4140/delayed_events/{delay_id}/
+/// {action}`
+///
+/// This is the endpoint used by delegated LiveKit JWT services. MSC4140 makes
+/// it intentionally unauthenticated; the service applies an IP rate limit.
+pub(crate) async fn update_delayed_event_v2_route(
+	State(services): State<crate::State>,
+	ClientIp(client): ClientIp,
+	body: Ruma<update_delayed_event::unstable_v2::Request>,
+) -> Result<update_delayed_event::unstable_v2::Response> {
+	services
+		.delayed_events
+		.update(&body.delay_id, body.action.as_ref(), None, client)
+		.await?;
+
+	Ok(update_delayed_event::unstable_v2::Response::new())
+}
+
+/// `GET /_matrix/client/unstable/org.matrix.msc4140/delayed_events`
+pub(crate) async fn get_all_delayed_events_route(
+	State(services): State<crate::State>,
+	body: Ruma<get_all_delayed_events::unstable::Request>,
+) -> Result<get_all_delayed_events::unstable::Response> {
+	Ok(get_all_delayed_events::unstable::Response::new(
+		services
+			.delayed_events
+			.list(body.sender_user())
+			.await?,
+	))
+}
+
+/// `GET /_matrix/client/unstable/org.matrix.msc4140/delayed_events/{delay_id}`
+pub(crate) async fn get_delayed_event_route(
+	State(services): State<crate::State>,
+	body: Ruma<get_delayed_event::unstable::Request>,
+) -> Result<get_delayed_event::unstable::Response> {
+	Ok(get_delayed_event::unstable::Response::new(
+		services
+			.delayed_events
+			.get(&body.delay_id, body.sender_user())
+			.await?,
+	))
+}
+
+#[cfg(test)]
+mod tests {
+	use super::delay_from_query;
+
+	#[test]
+	fn parses_legacy_delay_query() {
+		let uri = "/_matrix/client/v3/rooms/!room:example.org/send/m.room.message/tx?foo=bar&\
+		           org.matrix.msc4140.delay=123";
+		assert_eq!(
+			delay_from_query(&uri.parse().unwrap())
+				.unwrap()
+				.unwrap()
+				.as_millis(),
+			123
+		);
+	}
+
+	#[test]
+	fn ignores_requests_without_a_delay_query() {
+		assert!(
+			delay_from_query(&"/path".parse().unwrap())
+				.unwrap()
+				.is_none()
+		);
+	}
+}

src/api/client/message.rs

@@ -449,6 +449,9 @@ pub(crate) async fn with_membership(
 	user_id: &UserId,
 	encrypted: bool,
 ) -> PduEvent {
+	if pdu.sender() != user_id {
+		pdu.remove_transaction_id().log_err().ok();
+	}
 	annotate_membership(services, &mut pdu, user_id, encrypted).await;
 	pdu
 }

src/api/client/mod.rs

@@ -7,6 +7,7 @@ pub(super) mod backup;
 pub(super) mod capabilities;
 pub(super) mod context;
 pub(super) mod dehydrated_device;
+pub(super) mod delayed_events;
 pub(super) mod device;
 pub(super) mod directory;
 pub(super) mod events;
@@ -57,6 +58,7 @@ pub(super) use backup::*;
 pub(super) use capabilities::*;
 pub(super) use context::*;
 pub(super) use dehydrated_device::*;
+pub(super) use delayed_events::*;
 pub(super) use device::*;
 pub(super) use directory::*;
 pub(super) use events::*;

src/api/client/push/notifications.rs

@@ -63,12 +63,15 @@ pub(crate) async fn get_notifications_route(
 				count: count.into(),
 			};
 
-			let event = services
+			let mut event = services
 				.timeline
 				.get_pdu_from_id(&pdu_id.into())
 				.await
 				.ok()
 				.filter(|event| !event.is_redacted())?;
+			if event.sender() != sender_user {
+				event.remove_transaction_id().ok();
+			}
 
 			let read = services
 				.pusher

src/api/client/room/event.rs

@@ -85,6 +85,9 @@ pub(crate) async fn get_room_event_route(
 			"sender": event.sender().as_str(),
 		}));
 	}
+	if event.sender() != sender_user {
+		event.remove_transaction_id().ok();
+	}
 
 	debug_assert!(
 		event.event_id() == event_id && event.room_id() == room_id,

src/api/client/state.rs

@@ -135,7 +135,7 @@ pub(crate) async fn get_state_events_for_key_route(
 		))));
 	}
 
-	let event = services
+	let mut event = services
 		.state_accessor
 		.room_state_get(&body.room_id, &body.event_type, &body.state_key)
 		.await
@@ -146,6 +146,9 @@ pub(crate) async fn get_state_events_for_key_route(
 				"Failed to get state event: {e}.",
 			))))
 		})?;
+	if event.sender() != sender_user {
+		event.remove_transaction_id().ok();
+	}
 
 	let event_or_content = match body.format {
 		| StateEventFormat::Event => json!({

src/api/client/versions.rs

@@ -43,6 +43,11 @@ pub(crate) async fn get_supported_versions_route(
 					.rendezvous_enabled
 					.then_some("org.matrix.msc4108"),
 			)
+			.chain(
+				(services.config.max_event_delay_duration > 0
+					&& services.config.max_delayed_events_per_user > 0)
+					.then_some("org.matrix.msc4140"),
+			)
 			.map(Into::into)
 			.zip(once(true).cycle())
 			.collect(),