diff --git a/crates/core/src/client.rs b/crates/core/src/client.rs index 9fe9241fd..ecd99279f 100644 --- a/crates/core/src/client.rs +++ b/crates/core/src/client.rs @@ -3,6 +3,8 @@ pub mod admin; pub mod appservice; pub mod backup; pub mod dehydrated_device; +#[cfg(feature = "unstable-msc4140")] +pub mod delayed_events; pub mod device; pub mod directory; pub mod discovery; diff --git a/crates/core/src/client/delayed_events.rs b/crates/core/src/client/delayed_events.rs new file mode 100644 index 000000000..da86ff537 --- /dev/null +++ b/crates/core/src/client/delayed_events.rs @@ -0,0 +1,501 @@ +//! Endpoints for sending and interacting with delayed events. +//! +//! Delayed events are an unstable feature added by [MSC4140]. +//! +//! [MSC4140]: https://github.com/matrix-org/matrix-spec-proposals/pull/4140 + +use std::collections::BTreeMap; +use std::time::Duration; + +use salvo::prelude::*; +use serde::{Deserialize, Serialize}; + +use crate::events::TimelineEventType; +use crate::serde::{JsonValue, StringEnum}; +use crate::{OwnedEventId, OwnedRoomId, OwnedTransactionId, PrivOwnedStr, UnixMillis}; + +/// The standard error response body stored for a delayed event that failed to +/// be sent. +#[derive(ToSchema, Clone, Debug, Serialize, Deserialize)] +pub struct DelayedEventError { + /// The Matrix error code of the failure, e.g. `M_FORBIDDEN`. + pub errcode: String, + + /// A human-readable description of the failure. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error: Option, + + /// Additional fields carried by the Matrix standard error response. + #[serde(flatten)] + #[salvo(schema(value_type = Object, additional_properties = true))] + pub extra: BTreeMap, +} + +/// The structure of the data for returning a delayed event from a GET endpoint. +#[derive(ToSchema, Clone, Debug, Serialize, Deserialize)] +pub struct DelayedEventData { + /// The ID of the delayed event. + pub delay_id: String, + + /// The ID of the room that the delayed event was scheduled to be sent in. + pub room_id: OwnedRoomId, + + /// The event type of the delayed event. + #[serde(rename = "type")] + pub event_type: TimelineEventType, + + /// The state key if the event is a state event, nothing otherwise. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub state_key: Option, + + /// The event content to send. + /// + /// This is the content that was submitted to the send endpoint, not the + /// content of the final event. + #[salvo(schema(value_type = Object, additional_properties = true))] + pub content: JsonValue, + + /// The duration that the server should wait before sending this event. + #[serde(rename = "delay_ms", with = "crate::serde::duration::ms")] + #[salvo(schema(value_type = u64))] + pub delay: Duration, + + /// The timestamp when the delayed event was scheduled or last restarted. + #[serde(rename = "scheduled_at")] + pub running_since: UnixMillis, + + /// The error that prevented the delayed event from being sent. + /// + /// Present only for finalized events that were cancelled due to an error. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error: Option, + + /// The `event_id` this event got when it was sent. + /// + /// Present only for events that were sent successfully. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub event_id: Option, + + /// The timestamp when the event was finalized. + /// + /// Present only for events that were finalized (sent, failed to send, or + /// cancelled). + #[serde( + rename = "finalised_ts", + default, + skip_serializing_if = "Option::is_none" + )] + pub finalized_ts: Option, +} + +impl DelayedEventData { + /// Returns the status indicated by this delayed event data. + pub fn status(&self) -> DelayedEventStatus { + if self.finalized_ts.is_none() { + DelayedEventStatus::Scheduled + } else if self.event_id.is_some() { + DelayedEventStatus::Send + } else if self.error.is_some() { + DelayedEventStatus::Error + } else { + DelayedEventStatus::Cancel + } + } +} + +/// The status that a delayed event stored on the server can have. +#[doc = include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/src/doc/string_enum.md"))] +#[derive(ToSchema, Clone, StringEnum)] +#[palpo_enum(rename_all = "snake_case")] +#[non_exhaustive] +pub enum DelayedEventStatus { + /// The event is currently scheduled to be submitted at a later date. + /// It may be restarted, sent or cancelled via the management endpoint. + Scheduled, + + /// The event has been sent successfully. + Send, + + /// The event has been cancelled. + Cancel, + + /// The event has encountered an error when trying to send. + Error, + + #[doc(hidden)] + _Custom(PrivOwnedStr), +} + +/// The possible update actions for updating a delayed event. +#[doc = include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/src/doc/string_enum.md"))] +#[derive(ToSchema, Clone, StringEnum)] +#[palpo_enum(rename_all = "lowercase")] +#[non_exhaustive] +pub enum UpdateAction { + /// Restart the delayed event timeout. (heartbeat ping) + Restart, + + /// Send the delayed event immediately independent of the timeout state. + /// (deletes all timers) + Send, + + /// Delete the delayed event and never send it. (deletes all timers) + Cancel, + + #[doc(hidden)] + _Custom(PrivOwnedStr), +} + +// /// `PUT /_matrix/client/unstable/org.matrix.msc4140/rooms/{room_id}/delayed_event/{event_type}/ +// {txn_id}` /// +// /// Send a delayed event (a scheduled message) to a room. +// const METADATA: Metadata = metadata! { +// method: PUT, +// rate_limited: true, +// authentication: AccessToken, +// history: { +// unstable("org.matrix.msc4140") => +// "/_matrix/client/unstable/org.matrix.msc4140/rooms/{room_id}/delayed_event/{event_type}/{txn_id}" +// , } +// }; + +/// Request args for the `send_delayed_event` endpoint. +#[derive(ToParameters, Deserialize, Debug)] +pub struct SendDelayedEventReqArgs { + /// The room to send the event to. + #[salvo(parameter(parameter_in = Path))] + pub room_id: OwnedRoomId, + + /// The type of event to send. + #[salvo(parameter(parameter_in = Path))] + pub event_type: TimelineEventType, + + /// The transaction ID for this event. + /// + /// Clients should generate a unique ID across requests within the + /// same session. It will be used by the server to ensure idempotency of + /// requests. + #[salvo(parameter(parameter_in = Path))] + pub txn_id: OwnedTransactionId, + + /// Timestamp to use for the `origin_server_ts` of the event when it is + /// sent. + /// + /// This is called [timestamp massaging] and can only be used by + /// Appservices. + /// + /// [timestamp massaging]: https://spec.matrix.org/latest/application-service-api/#timestamp-massaging + #[salvo(parameter(parameter_in = Query))] + #[serde(default, skip_serializing_if = "Option::is_none", rename = "ts")] + pub timestamp: Option, +} + +/// Request body for the `send_delayed_event` endpoint. +#[derive(ToSchema, Serialize, Deserialize, Debug)] +pub struct SendDelayedEventReqBody { + /// The duration that the server should wait before sending this event. + #[serde(rename = "delay_ms", with = "crate::serde::duration::ms")] + #[salvo(schema(value_type = u64))] + pub delay: Duration, + + /// The state key if the event is a state event, nothing otherwise. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub state_key: Option, + + /// The event content to send. + #[salvo(schema(value_type = Object, additional_properties = true))] + pub content: JsonValue, +} + +/// Response type for the `send_delayed_event` endpoint. +#[derive(ToSchema, Serialize, Deserialize, Debug)] +pub struct SendDelayedEventResBody { + /// The `delay_id` generated for this delayed event. Used to interact with + /// delayed events. + pub delay_id: String, +} + +/// Response from a regular room send endpoint, which returns either an event +/// id for an immediate send or a delay id when MSC4140 scheduling was +/// requested. +#[derive(ToSchema, Serialize, Debug)] +pub struct SendEventResBody { + /// The event created by an immediate send. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub event_id: Option, + + /// The delayed event created by a scheduled send. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub delay_id: Option, +} + +impl SendEventResBody { + pub fn sent(event_id: OwnedEventId) -> Self { + Self { + event_id: Some(event_id), + delay_id: None, + } + } + + pub fn delayed(delay_id: String) -> Self { + Self { + event_id: None, + delay_id: Some(delay_id), + } + } +} + +impl SendDelayedEventResBody { + /// Creates a new `SendDelayedEventResBody` with the given delay id. + pub fn new(delay_id: String) -> Self { + Self { delay_id } + } +} + +// /// `GET /_matrix/client/unstable/org.matrix.msc4140/delayed_events` +// /// +// /// Get all of the user's scheduled delayed events. +// const METADATA: Metadata = metadata! { +// method: GET, +// rate_limited: true, +// authentication: AccessToken, +// history: { +// unstable("org.matrix.msc4140") => +// "/_matrix/client/unstable/org.matrix.msc4140/delayed_events", } +// }; + +/// Response type for the `get_all_delayed_events` endpoint. +#[derive(ToSchema, Serialize, Deserialize, Debug)] +pub struct DelayedEventsResBody { + /// An array of objects describing scheduled delayed events owned by the + /// requesting user. + pub delayed_events: Vec, +} + +impl DelayedEventsResBody { + /// Creates a new `DelayedEventsResBody` with the given delayed events. + pub fn new(delayed_events: Vec) -> Self { + Self { delayed_events } + } +} + +// /// `GET /_matrix/client/unstable/org.matrix.msc4140/delayed_events/{delay_id}` +// /// +// /// Get the information about a delayed event. The response body is a single +// /// `DelayedEventData` object. +// const METADATA: Metadata = metadata! { +// method: GET, +// rate_limited: true, +// authentication: AccessToken, +// history: { +// unstable("org.matrix.msc4140") => +// "/_matrix/client/unstable/org.matrix.msc4140/delayed_events/{delay_id}", } +// }; + +// /// `POST /_matrix/client/unstable/org.matrix.msc4140/delayed_events/{delay_id}/{action}` +// /// +// /// Update a delayed event: restart its timeout, send it immediately, or +// /// cancel it. +// const METADATA: Metadata = metadata! { +// method: POST, +// rate_limited: true, +// authentication: AccessToken, +// history: { +// unstable("org.matrix.msc4140") => +// "/_matrix/client/unstable/org.matrix.msc4140/delayed_events/{delay_id}/{action}", } +// }; + +/// Request args for the `update_delayed_event` endpoint. +#[derive(ToParameters, Deserialize, Debug)] +pub struct UpdateDelayedEventReqArgs { + /// The delay id that we want to update. + #[salvo(parameter(parameter_in = Path))] + pub delay_id: String, + + /// Which kind of update we want to request for the delayed event. + #[salvo(parameter(parameter_in = Path))] + pub action: UpdateAction, +} + +/// Request body for the deprecated body-action variant of the +/// `update_delayed_event` endpoint +/// (`POST /_matrix/client/unstable/org.matrix.msc4140/delayed_events/{delay_id}`). +#[derive(ToSchema, Serialize, Deserialize, Debug)] +pub struct UpdateDelayedEventReqBody { + /// Which kind of update we want to request for the delayed event. + pub action: UpdateAction, +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use serde_json::{Value as JsonValue, json}; + + use super::*; + + #[test] + fn deserialize_send_delayed_event_req_body() { + let body: SendDelayedEventReqBody = serde_json::from_value(json!({ + "delay_ms": 103, + "content": {"msgtype": "m.text", "body": "test"}, + })) + .unwrap(); + + assert_eq!(body.delay, Duration::from_millis(103)); + assert_eq!(body.state_key, None); + assert_eq!(body.content, json!({"msgtype": "m.text", "body": "test"})); + } + + #[test] + fn deserialize_send_delayed_state_event_req_body() { + let body: SendDelayedEventReqBody = serde_json::from_value(json!({ + "delay_ms": 9000, + "state_key": "a_state_key", + "content": {"topic": "test topic"}, + })) + .unwrap(); + + assert_eq!(body.delay, Duration::from_millis(9000)); + assert_eq!(body.state_key.as_deref(), Some("a_state_key")); + } + + #[test] + fn serialize_delayed_event_data() { + let event = DelayedEventData { + delay_id: "a_delay_id".to_owned(), + room_id: "!roomid:example.org".try_into().unwrap(), + event_type: "m.room.topic".into(), + state_key: Some("a_state_key".to_owned()), + content: json!({"topic": "test topic"}), + delay: Duration::from_millis(103), + running_since: crate::UnixMillis(70000), + error: None, + event_id: Some("$event:example.org".try_into().unwrap()), + finalized_ts: Some(crate::UnixMillis(70103)), + }; + assert_eq!(event.status(), DelayedEventStatus::Send); + + assert_eq!( + serde_json::to_value(&event).unwrap(), + json!({ + "content": {"topic": "test topic"}, + "delay_ms": 103, + "delay_id": "a_delay_id", + "event_id": "$event:example.org", + "finalised_ts": 70103, + "room_id": "!roomid:example.org", + "scheduled_at": 70000, + "state_key": "a_state_key", + "type": "m.room.topic", + }) + ); + } + + #[test] + fn scheduled_delayed_event_data_status() { + let event = DelayedEventData { + delay_id: "a_delay_id".to_owned(), + room_id: "!roomid:example.org".try_into().unwrap(), + event_type: "m.room.message".into(), + state_key: None, + content: json!({"msgtype": "m.text", "body": "test"}), + delay: Duration::from_millis(103), + running_since: crate::UnixMillis(70000), + error: None, + event_id: None, + finalized_ts: None, + }; + assert_eq!(event.status(), DelayedEventStatus::Scheduled); + + let serialized = serde_json::to_value(&event).unwrap(); + assert!(serialized.get("event_id").is_none()); + assert!(serialized.get("finalised_ts").is_none()); + assert!(serialized.get("error").is_none()); + assert!(serialized.get("state_key").is_none()); + } + + #[test] + fn error_delayed_event_data_status() { + let event = DelayedEventData { + delay_id: "a_delay_id".to_owned(), + room_id: "!roomid:example.org".try_into().unwrap(), + event_type: "m.room.message".into(), + state_key: None, + content: json!({"msgtype": "m.text", "body": "test"}), + delay: Duration::from_millis(103), + running_since: crate::UnixMillis(70000), + error: Some(DelayedEventError { + errcode: "M_FORBIDDEN".to_owned(), + error: Some("you shall not pass".to_owned()), + extra: BTreeMap::new(), + }), + event_id: None, + finalized_ts: Some(crate::UnixMillis(70103)), + }; + assert_eq!(event.status(), DelayedEventStatus::Error); + } + + #[test] + fn delayed_event_error_preserves_standard_error_extensions() { + let error: DelayedEventError = serde_json::from_value(json!({ + "errcode": "M_LIMIT_EXCEEDED", + "error": "slow down", + "retry_after_ms": 1500, + })) + .unwrap(); + + assert_eq!(error.extra.get("retry_after_ms"), Some(&json!(1500))); + assert_eq!( + serde_json::to_value(error).unwrap(), + json!({ + "errcode": "M_LIMIT_EXCEEDED", + "error": "slow down", + "retry_after_ms": 1500, + }) + ); + } + + #[test] + fn update_action_from_str() { + assert_eq!(UpdateAction::from("restart"), UpdateAction::Restart); + assert_eq!(UpdateAction::from("send"), UpdateAction::Send); + assert_eq!(UpdateAction::from("cancel"), UpdateAction::Cancel); + + let body: UpdateDelayedEventReqBody = + serde_json::from_value(json!({"action": "cancel"})).unwrap(); + assert_eq!(body.action, UpdateAction::Cancel); + } + + #[test] + fn cancelled_delayed_event_data_status() { + let event = DelayedEventData { + delay_id: "a_delay_id".to_owned(), + room_id: "!roomid:example.org".try_into().unwrap(), + event_type: "m.room.message".into(), + state_key: None, + content: JsonValue::Null, + delay: Duration::from_millis(103), + running_since: crate::UnixMillis(70000), + error: None, + event_id: None, + finalized_ts: Some(crate::UnixMillis(70103)), + }; + assert_eq!(event.status(), DelayedEventStatus::Cancel); + } + + #[test] + fn send_event_response_contains_exactly_one_identifier() { + let sent = SendEventResBody::sent("$event:example.org".try_into().unwrap()); + assert_eq!( + serde_json::to_value(sent).unwrap(), + json!({"event_id": "$event:example.org"}) + ); + + let delayed = SendEventResBody::delayed("a_delay_id".to_owned()); + assert_eq!( + serde_json::to_value(delayed).unwrap(), + json!({"delay_id": "a_delay_id"}) + ); + } +} diff --git a/crates/core/src/client/message.rs b/crates/core/src/client/message.rs index eb07c55c2..39ff387dc 100644 --- a/crates/core/src/client/message.rs +++ b/crates/core/src/client/message.rs @@ -161,6 +161,16 @@ pub struct CreateMessageWithTxnReqArgs { #[serde(skip_serializing_if = "Option::is_none", rename = "ts")] pub timestamp: Option, + /// Delay this event by the given number of milliseconds (MSC4140). + #[cfg(feature = "unstable-msc4140")] + #[salvo(parameter(parameter_in = Query))] + #[serde( + default, + skip_serializing_if = "Option::is_none", + rename = "org.matrix.msc4140.delay" + )] + pub delay: Option, + /// The duration for which the event should receive sticky delivery guarantees. #[cfg(feature = "unstable-msc4354")] #[salvo(parameter(parameter_in = Query))] @@ -200,6 +210,16 @@ pub struct CreateMessageReqArgs { #[serde(skip_serializing_if = "Option::is_none", rename = "ts")] pub timestamp: Option, + /// Delay this event by the given number of milliseconds (MSC4140). + #[cfg(feature = "unstable-msc4140")] + #[salvo(parameter(parameter_in = Query))] + #[serde( + default, + skip_serializing_if = "Option::is_none", + rename = "org.matrix.msc4140.delay" + )] + pub delay: Option, + /// The duration for which the event should receive sticky delivery guarantees. #[cfg(feature = "unstable-msc4354")] #[salvo(parameter(parameter_in = Query))] @@ -225,6 +245,33 @@ impl SendMessageResBody { } } +#[cfg(all(test, feature = "unstable-msc4140"))] +mod delayed_event_tests { + use super::{CreateMessageReqArgs, CreateMessageWithTxnReqArgs}; + + #[test] + fn delay_uses_msc4140_query_name() { + let args: CreateMessageWithTxnReqArgs = serde_html_form::from_str( + "room_id=%21room%3Aexample.org&event_type=m.room.message&txn_id=0000&\ + org.matrix.msc4140.delay=123456", + ) + .unwrap(); + + assert_eq!(args.delay, Some(123_456)); + } + + #[test] + fn delay_is_accepted_without_a_transaction_id() { + let args: CreateMessageReqArgs = serde_html_form::from_str( + "room_id=%21room%3Aexample.org&event_type=m.room.message&\ + org.matrix.msc4140.delay=900", + ) + .unwrap(); + + assert_eq!(args.delay, Some(900)); + } +} + #[cfg(all(test, feature = "unstable-msc4354"))] mod sticky_tests { use super::CreateMessageWithTxnReqArgs; diff --git a/crates/core/src/client/state.rs b/crates/core/src/client/state.rs index 1d1e8b7e5..64fc80589 100644 --- a/crates/core/src/client/state.rs +++ b/crates/core/src/client/state.rs @@ -223,6 +223,16 @@ pub struct SendStateEventReqArgs { #[serde(default, rename = "ts", skip_serializing_if = "Option::is_none")] pub timestamp: Option, + /// Delay this event by the given number of milliseconds (MSC4140). + #[cfg(feature = "unstable-msc4140")] + #[salvo(parameter(parameter_in = Query))] + #[serde( + default, + skip_serializing_if = "Option::is_none", + rename = "org.matrix.msc4140.delay" + )] + pub delay: Option, + /// The duration for which the event should receive sticky delivery guarantees. #[cfg(feature = "unstable-msc4354")] #[salvo(parameter(parameter_in = Query))] @@ -249,6 +259,22 @@ impl SendStateEventResBody { } } +#[cfg(all(test, feature = "unstable-msc4140"))] +mod delayed_event_tests { + use super::SendStateEventReqArgs; + + #[test] + fn delay_uses_msc4140_query_name() { + let args: SendStateEventReqArgs = serde_html_form::from_str( + "room_id=%21room%3Aexample.org&event_type=m.room.topic&state_key=&\ + org.matrix.msc4140.delay=123456", + ) + .unwrap(); + + assert_eq!(args.delay, Some(123_456)); + } +} + #[cfg(all(test, feature = "unstable-msc4354"))] mod sticky_tests { use super::SendStateEventReqArgs; diff --git a/crates/core/src/error.rs b/crates/core/src/error.rs index 4a8070010..85ec3088d 100644 --- a/crates/core/src/error.rs +++ b/crates/core/src/error.rs @@ -17,7 +17,8 @@ mod kind; /// Separate module because it's a lot of code. mod kind_serde; pub use kind::*; -use kind_serde::{ErrorCode, RetryAfter}; +use kind_serde::ErrorCode; +pub use kind_serde::RetryAfter; use crate::{MatrixVersion, OwnedUserId, RoomVersionId}; @@ -92,6 +93,7 @@ impl MatrixError { captcha_needed, CaptchaNeeded; connection_failed, ConnectionFailed; connection_timeout, ConnectionTimeout; + delay_too_large, DelayTooLarge; duplicate_annotation, DuplicateAnnotation; exclusive, Exclusive; guest_access_forbidden, GuestAccessForbidden; @@ -225,6 +227,24 @@ impl Scribe for MatrixError { }; let Self { kind, mut body, .. } = self; + if let ErrorKind::LimitExceeded { + retry_after: Some(RetryAfter::Delay(duration)), + } = &kind + { + res.add_header( + header::RETRY_AFTER, + duration + .as_secs() + .saturating_add(u64::from(duration.subsec_nanos() != 0)) + .max(1) + .to_string(), + true, + ) + .ok(); + if let Ok(ms) = u64::try_from(duration.as_millis()) { + body.0.insert("retry_after_ms".to_owned(), ms.into()); + } + } body.0 .insert("errcode".to_owned(), kind.code().to_string().into()); diff --git a/crates/core/src/error/kind.rs b/crates/core/src/error/kind.rs index 1eaf3feaf..87e38c9e4 100644 --- a/crates/core/src/error/kind.rs +++ b/crates/core/src/error/kind.rs @@ -101,6 +101,11 @@ pub enum ErrorKind { /// The connection to the application service timed out. ConnectionTimeout, + /// `M_DELAY_TOO_LARGE` + /// + /// A delayed event requested a delay larger than the homeserver permits. + DelayTooLarge, + /// `M_DUPLICATE_ANNOTATION` /// /// The request is an attempt to send a [duplicate annotation]. @@ -465,6 +470,7 @@ impl ErrorKind { ErrorKind::ConflictingUnsubscription => ErrorCode::ConflictingUnsubscription, ErrorKind::ConnectionFailed => ErrorCode::ConnectionFailed, ErrorKind::ConnectionTimeout => ErrorCode::ConnectionTimeout, + ErrorKind::DelayTooLarge => ErrorCode::DelayTooLarge, ErrorKind::DuplicateAnnotation => ErrorCode::DuplicateAnnotation, ErrorKind::Exclusive => ErrorCode::Exclusive, ErrorKind::Forbidden => ErrorCode::Forbidden, diff --git a/crates/core/src/error/kind_serde.rs b/crates/core/src/error/kind_serde.rs index 91b45a895..75b944d6b 100644 --- a/crates/core/src/error/kind_serde.rs +++ b/crates/core/src/error/kind_serde.rs @@ -203,6 +203,7 @@ impl<'de> Visitor<'de> for ErrorKindVisitor { ErrorCode::ConflictingUnsubscription => ErrorKind::ConflictingUnsubscription, ErrorCode::ConnectionFailed => ErrorKind::ConnectionFailed, ErrorCode::ConnectionTimeout => ErrorKind::ConnectionTimeout, + ErrorCode::DelayTooLarge => ErrorKind::DelayTooLarge, ErrorCode::DuplicateAnnotation => ErrorKind::DuplicateAnnotation, ErrorCode::Exclusive => ErrorKind::Exclusive, ErrorCode::Forbidden => ErrorKind::Forbidden, @@ -388,6 +389,11 @@ pub enum ErrorCode { /// The connection to the application service timed out. ConnectionTimeout, + /// `M_DELAY_TOO_LARGE` + /// + /// A delayed event requested a delay larger than the homeserver permits. + DelayTooLarge, + /// `M_DUPLICATE_ANNOTATION` /// /// The request is an attempt to send a [duplicate annotation]. @@ -822,6 +828,17 @@ mod tests { // assert_eq!(deserialized, ErrorKind::Forbidden); // } + #[test] + fn delay_too_large_serde() { + let value = json!({ "errcode": "M_DELAY_TOO_LARGE" }); + + assert_eq!( + from_json_value::(value.clone()).unwrap(), + ErrorKind::DelayTooLarge + ); + assert_eq!(to_json_value(ErrorKind::DelayTooLarge).unwrap(), value); + } + #[test] fn deserialize_incompatible_room_version() { let deserialized: ErrorKind = from_json_value(json!({ diff --git a/crates/data/migrations/2026-07-23-000001_delayed_events/down.sql b/crates/data/migrations/2026-07-23-000001_delayed_events/down.sql new file mode 100644 index 000000000..ab4b159b6 --- /dev/null +++ b/crates/data/migrations/2026-07-23-000001_delayed_events/down.sql @@ -0,0 +1,12 @@ +DROP TRIGGER IF EXISTS delayed_events_remove_output ON delayed_events; +DROP FUNCTION IF EXISTS palpo_remove_delayed_event_output; +DROP TRIGGER IF EXISTS events_record_delayed_event_output ON events; +DROP FUNCTION IF EXISTS palpo_confirm_delayed_event_output; +DROP TRIGGER IF EXISTS event_datas_track_delayed_event_output ON event_datas; +DROP FUNCTION IF EXISTS palpo_track_delayed_event_output; +DROP TABLE IF EXISTS delayed_event_outputs; +DROP INDEX IF EXISTS idx_delayed_events_finalized; +DROP INDEX IF EXISTS idx_delayed_events_due; +DROP INDEX IF EXISTS idx_delayed_events_user; +DROP INDEX IF EXISTS idx_delayed_events_txn; +DROP TABLE IF EXISTS delayed_events; diff --git a/crates/data/migrations/2026-07-23-000001_delayed_events/up.sql b/crates/data/migrations/2026-07-23-000001_delayed_events/up.sql new file mode 100644 index 000000000..321d14fbb --- /dev/null +++ b/crates/data/migrations/2026-07-23-000001_delayed_events/up.sql @@ -0,0 +1,127 @@ +-- MSC4140 delayed events. +-- +-- Scheduled events that the homeserver sends into a room on the user's behalf +-- once their delay elapses. Rows survive restarts so pending events are +-- recovered and sent after the server comes back up. Finalized rows (sent, +-- cancelled, or errored) are retained for lookup and pruned periodically. +CREATE TABLE delayed_events ( + id BIGSERIAL NOT NULL PRIMARY KEY, + delay_id TEXT NOT NULL, + user_id TEXT NOT NULL, + device_id TEXT, + room_id TEXT NOT NULL, + event_type TEXT NOT NULL, + state_key TEXT, + content JSONB NOT NULL, + delay_ms BIGINT NOT NULL, + txn_id TEXT NOT NULL, + origin_server_ts BIGINT, + running_since BIGINT NOT NULL, + send_at BIGINT NOT NULL, + event_id TEXT, + error JSONB, + finalized_at BIGINT, + created_at BIGINT NOT NULL, + UNIQUE (delay_id) +); +-- Idempotency: one delayed event per (user, device session, transaction id). +CREATE UNIQUE INDEX idx_delayed_events_txn ON delayed_events(user_id, COALESCE(device_id, ''), txn_id); +CREATE INDEX idx_delayed_events_user ON delayed_events(user_id); +CREATE INDEX idx_delayed_events_due ON delayed_events(send_at) WHERE finalized_at IS NULL; +CREATE INDEX idx_delayed_events_finalized ON delayed_events(finalized_at) WHERE finalized_at IS NOT NULL; + +-- The timeline append and the delayed-event outcome are written through +-- different connections. Record the output in the same transaction that +-- promotes the event from an outlier into the timeline, so a process crash +-- between append and outcome finalization cannot cause a second room event on +-- recovery. The primary key also rejects any accidental second append for the +-- same delay id. +CREATE TABLE delayed_event_outputs ( + delay_id TEXT NOT NULL PRIMARY KEY, + event_id TEXT NOT NULL UNIQUE +); + +-- Remember the candidate as soon as its outlier is persisted. If an attempt +-- fails before promotion, a later attempt may replace that still-outlier +-- candidate. Once the mapped event is promoted, this mapping is immutable. +CREATE FUNCTION palpo_track_delayed_event_output() RETURNS TRIGGER AS $$ +DECLARE + v_delayed_id TEXT; +BEGIN + v_delayed_id := NEW.json_data -> 'unsigned' ->> 'org.matrix.msc4140.delay_id'; + IF v_delayed_id IS NOT NULL AND EXISTS ( + SELECT 1 FROM delayed_events + WHERE delay_id = v_delayed_id + AND user_id = NEW.json_data ->> 'sender' + AND room_id = NEW.room_id + ) THEN + INSERT INTO delayed_event_outputs (delay_id, event_id) + VALUES (v_delayed_id, NEW.event_id) + ON CONFLICT (delay_id) DO UPDATE + SET event_id = EXCLUDED.event_id + WHERE EXISTS ( + SELECT 1 FROM events + WHERE id = delayed_event_outputs.event_id + AND is_outlier = TRUE + ); + END IF; + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +CREATE TRIGGER event_datas_track_delayed_event_output +AFTER INSERT ON event_datas +FOR EACH ROW EXECUTE FUNCTION palpo_track_delayed_event_output(); + +-- Promotion is the definitive commit point. A different event already mapped +-- to this delay id must make this promotion fail, otherwise two events could +-- become visible after recovery from a partially completed append. +CREATE FUNCTION palpo_confirm_delayed_event_output() RETURNS TRIGGER AS $$ +DECLARE + v_delayed_id TEXT; + v_mapped_event_id TEXT; +BEGIN + SELECT json_data -> 'unsigned' ->> 'org.matrix.msc4140.delay_id' + INTO v_delayed_id + FROM event_datas + WHERE event_id = NEW.id; + IF v_delayed_id IS NOT NULL AND EXISTS ( + SELECT 1 FROM delayed_events + WHERE delay_id = v_delayed_id + AND user_id = NEW.sender_id + AND room_id = NEW.room_id + ) THEN + SELECT output.event_id INTO v_mapped_event_id + FROM delayed_event_outputs AS output + WHERE output.delay_id = v_delayed_id; + + IF v_mapped_event_id IS NULL THEN + INSERT INTO delayed_event_outputs (delay_id, event_id) + VALUES (v_delayed_id, NEW.id); + ELSIF v_mapped_event_id <> NEW.id THEN + RAISE EXCEPTION 'delay id % is already mapped to event %', + v_delayed_id, v_mapped_event_id + USING ERRCODE = 'unique_violation'; + END IF; + END IF; + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +CREATE TRIGGER events_record_delayed_event_output +AFTER UPDATE OF is_outlier ON events +FOR EACH ROW +WHEN (OLD.is_outlier IS DISTINCT FROM FALSE AND NEW.is_outlier = FALSE) +EXECUTE FUNCTION palpo_confirm_delayed_event_output(); + +-- Outcome markers only need to live as long as their delayed-event row. +CREATE FUNCTION palpo_remove_delayed_event_output() RETURNS TRIGGER AS $$ +BEGIN + DELETE FROM delayed_event_outputs WHERE delay_id = OLD.delay_id; + RETURN OLD; +END; +$$ LANGUAGE plpgsql; + +CREATE TRIGGER delayed_events_remove_output +AFTER DELETE ON delayed_events +FOR EACH ROW EXECUTE FUNCTION palpo_remove_delayed_event_output(); diff --git a/crates/data/src/room.rs b/crates/data/src/room.rs index 7616cc05f..2f2ce6834 100644 --- a/crates/data/src/room.rs +++ b/crates/data/src/room.rs @@ -9,6 +9,7 @@ use crate::core::{MatrixError, Seqnum, UnixMillis}; use crate::schema::*; use crate::{DataResult, connect}; +pub mod delayed_event; pub mod event; pub mod event_report; pub mod lazy_loading; diff --git a/crates/data/src/room/delayed_event.rs b/crates/data/src/room/delayed_event.rs new file mode 100644 index 000000000..d09992806 --- /dev/null +++ b/crates/data/src/room/delayed_event.rs @@ -0,0 +1,439 @@ +//! Persistence for MSC4140 delayed events. +//! +//! A delayed event is stored when scheduled and stays in the table after it is +//! finalized (sent, cancelled, or errored) so clients can look up the outcome. +//! A sender holds a PostgreSQL row lock from selection through the room append +//! and outcome write. A worker crash rolls that transaction back, leaving the +//! event scheduled for another worker without a time-based lease race. + +use diesel::prelude::*; +use diesel_async::{AsyncConnection, AsyncPgConnection, RunQueryDsl}; + +use crate::core::identifiers::*; +use crate::core::serde::JsonValue; +use crate::core::{DeviceId, TransactionId, UnixMillis, UserId}; +use crate::schema::*; +use crate::{DataResult, connect}; + +#[derive(Identifiable, Queryable, Debug, Clone)] +#[diesel(table_name = delayed_events)] +pub struct DbDelayedEvent { + pub id: i64, + pub delay_id: String, + pub user_id: OwnedUserId, + pub device_id: Option, + pub room_id: OwnedRoomId, + pub event_type: String, + pub state_key: Option, + pub content: JsonValue, + pub delay_ms: i64, + pub txn_id: OwnedTransactionId, + pub origin_server_ts: Option, + pub running_since: i64, + pub send_at: i64, + pub event_id: Option, + pub error: Option, + pub finalized_at: Option, + pub created_at: i64, +} + +#[derive(Insertable, Debug, Clone)] +#[diesel(table_name = delayed_events)] +pub struct NewDbDelayedEvent { + pub delay_id: String, + pub user_id: OwnedUserId, + pub device_id: Option, + pub room_id: OwnedRoomId, + pub event_type: String, + pub state_key: Option, + pub content: JsonValue, + pub delay_ms: i64, + pub txn_id: OwnedTransactionId, + pub origin_server_ts: Option, + pub running_since: i64, + pub send_at: i64, + pub created_at: i64, +} + +/// Outcome of trying to schedule a delayed event. +pub enum Scheduled { + /// The event was stored. + Created(DbDelayedEvent), + /// A concurrent retry of the same transaction won the race; this is its + /// row. + AlreadyScheduled(DbDelayedEvent), + /// The user is already at `max_scheduled`. + LimitReached, +} + +/// Store a newly scheduled delayed event, enforcing the per-user limit. +/// +/// The count and the insert run under a transaction-scoped advisory lock keyed +/// on the user. Checking the count in the caller and inserting here separately +/// let concurrent requests with distinct transaction ids all observe a count +/// below the limit and all succeed, so a limit of 100 could be pushed well +/// past it; the unique transaction index does not serialize distinct +/// transactions. +/// +/// Two concurrent retries of the *same* transaction are a different race: both +/// pass the caller's [`get_by_txn_id`] lookup before either commits. The +/// transaction id is therefore re-resolved under the lock, so the loser is +/// answered with the winner's row and scheduling stays idempotent. +pub async fn create(new: NewDbDelayedEvent, max_scheduled: i64) -> DataResult { + let user_id = new.user_id.clone(); + let device_id = new.device_id.clone(); + let txn_id = new.txn_id.clone(); + + let mut conn = connect().await?; + conn.transaction::<_, diesel::result::Error, _>(async |conn| { + // Serialize scheduling per user for the rest of this transaction; + // released automatically on commit or rollback. + diesel::sql_query("SELECT pg_advisory_xact_lock(hashtext($1))") + .bind::(user_id.as_str()) + .execute(&mut *conn) + .await?; + + // Re-resolve the transaction now that this request holds the lock. The + // caller's lookup ran before it, so a concurrent retry of the same + // request may have committed in between. Doing this first means the + // insert can no longer hit the unique index -- which matters because a + // unique violation aborts the surrounding transaction in Postgres, and + // every statement after it would fail with 25P02 -- and it means an + // idempotent retry is answered with its delay id rather than being + // rejected by the limit check below when it took the last slot. + let mut existing = delayed_events::table + .filter(delayed_events::user_id.eq(&user_id)) + .filter(delayed_events::txn_id.eq(&txn_id)) + .into_boxed(); + existing = match device_id.as_deref() { + Some(device_id) => existing.filter(delayed_events::device_id.eq(device_id)), + None => existing.filter(delayed_events::device_id.is_null()), + }; + if let Some(row) = existing + .first::(&mut *conn) + .await + .optional()? + { + return Ok(Scheduled::AlreadyScheduled(row)); + } + + let scheduled: i64 = delayed_events::table + .filter(delayed_events::user_id.eq(&user_id)) + .filter(delayed_events::finalized_at.is_null()) + .count() + .get_result(&mut *conn) + .await?; + if scheduled >= max_scheduled { + return Ok(Scheduled::LimitReached); + } + + diesel::insert_into(delayed_events::table) + .values(&new) + .get_result::(&mut *conn) + .await + .map(Scheduled::Created) + }) + .await + .map_err(Into::into) +} + +/// Look up a delayed event previously scheduled with the same transaction id +/// on the same session, for idempotent retries of the scheduling request. +pub async fn get_by_txn_id( + user_id: &UserId, + device_id: Option<&DeviceId>, + txn_id: &TransactionId, +) -> DataResult> { + let mut query = delayed_events::table + .filter(delayed_events::user_id.eq(user_id)) + .filter(delayed_events::txn_id.eq(txn_id)) + .into_boxed(); + if let Some(device_id) = device_id { + query = query.filter(delayed_events::device_id.eq(device_id)); + } else { + query = query.filter(delayed_events::device_id.is_null()); + } + query + .first::(&mut connect().await?) + .await + .optional() + .map_err(Into::into) +} + +/// Fetch one delayed event owned by the user, whether scheduled or finalized. +pub async fn get_by_delay_id( + user_id: &UserId, + delay_id: &str, +) -> DataResult> { + delayed_events::table + .filter(delayed_events::user_id.eq(user_id)) + .filter(delayed_events::delay_id.eq(delay_id)) + .first::(&mut connect().await?) + .await + .optional() + .map_err(Into::into) +} + +/// List the user's scheduled (not yet finalized) delayed events in +/// chronological order of their intended send time. +pub async fn list_scheduled(user_id: &UserId) -> DataResult> { + delayed_events::table + .filter(delayed_events::user_id.eq(user_id)) + .filter(delayed_events::finalized_at.is_null()) + .order(delayed_events::send_at.asc()) + .load::(&mut connect().await?) + .await + .map_err(Into::into) +} + +/// Count the user's scheduled (not yet finalized) delayed events. +pub async fn count_scheduled(user_id: &UserId) -> DataResult { + delayed_events::table + .filter(delayed_events::user_id.eq(user_id)) + .filter(delayed_events::finalized_at.is_null()) + .count() + .get_result::(&mut connect().await?) + .await + .map_err(Into::into) +} + +/// The soonest scheduled send time among the user's delayed events, used for +/// the `Retry-After` header when the per-user limit is hit. +pub async fn next_send_at_of_user(user_id: &UserId) -> DataResult> { + delayed_events::table + .filter(delayed_events::user_id.eq(user_id)) + .filter(delayed_events::finalized_at.is_null()) + .select(diesel::dsl::min(delayed_events::send_at)) + .get_result::>(&mut connect().await?) + .await + .map_err(Into::into) +} + +/// The soonest scheduled send time across all users, used by the scheduler to +/// compute how long to sleep. +pub async fn next_send_at() -> DataResult> { + delayed_events::table + .filter(delayed_events::finalized_at.is_null()) + .select(diesel::dsl::min(delayed_events::send_at)) + .get_result::>(&mut connect().await?) + .await + .map_err(Into::into) +} + +/// Lock the next due event without waiting for another worker's row. +/// +/// The caller owns the surrounding transaction and must keep it open through +/// the append and outcome write. `SKIP LOCKED` lets multiple server processes +/// work on different rows while guaranteeing that only one can append a given +/// delayed event. +pub async fn lock_next_due( + conn: &mut AsyncPgConnection, + now: i64, +) -> DataResult> { + delayed_events::table + .filter(delayed_events::finalized_at.is_null()) + .filter(delayed_events::send_at.le(now)) + .order((delayed_events::send_at.asc(), delayed_events::id.asc())) + .for_update() + .skip_locked() + .first::(conn) + .await + .optional() + .map_err(Into::into) +} + +/// Lock one user's delayed event for a manual send. +/// +/// Unlike the scheduler this deliberately waits for an existing row holder: +/// once the lock is acquired the caller can return the definitive sent/error +/// result rather than guessing from an expiring lease. +pub async fn lock_for_send( + conn: &mut AsyncPgConnection, + user_id: &UserId, + delay_id: &str, +) -> DataResult> { + delayed_events::table + .filter(delayed_events::user_id.eq(user_id)) + .filter(delayed_events::delay_id.eq(delay_id)) + .for_update() + .first::(conn) + .await + .optional() + .map_err(Into::into) +} + +#[derive(QueryableByName)] +struct DelayedEventOutput { + #[diesel(sql_type = diesel::sql_types::Text)] + event_id: String, +} + +/// Find the event atomically recorded when an outlier enters the timeline. +/// +/// This is the durable recovery fence for a process that dies after appending +/// the room event but before finalizing the delayed-event row or recording its +/// regular transaction-id mapping. +pub async fn get_output(delay_id: &str) -> DataResult> { + let row = diesel::sql_query( + "SELECT output.event_id \ + FROM delayed_event_outputs AS output \ + INNER JOIN events ON events.id = output.event_id \ + WHERE output.delay_id = $1 AND events.is_outlier = FALSE", + ) + .bind::(delay_id) + .get_result::(&mut connect().await?) + .await + .optional()?; + + row.map(|row| OwnedEventId::try_from(row.event_id).map_err(Into::into)) + .transpose() +} + +/// Restart a scheduled delayed event's timer. Returns the updated row, or +/// `None` if the event does not exist, is owned by another user, or was +/// already finalized. +pub async fn restart( + conn: &mut AsyncPgConnection, + user_id: &UserId, + delay_id: &str, +) -> DataResult> { + conn.transaction::<_, diesel::result::Error, _>(async |conn| { + // Take the row lock before reading the clock. If another process is + // sending this event, the restarted delay must begin after that wait, + // not when the HTTP request first arrived. + let Some(row) = delayed_events::table + .filter(delayed_events::user_id.eq(user_id)) + .filter(delayed_events::delay_id.eq(delay_id)) + .for_update() + .first::(&mut *conn) + .await + .optional()? + else { + return Ok(None); + }; + if row.finalized_at.is_some() { + return Ok(None); + } + + let now = UnixMillis::now().get() as i64; + diesel::update(delayed_events::table.find(row.id)) + .filter(delayed_events::finalized_at.is_null()) + .set(( + delayed_events::running_since.eq(now), + delayed_events::send_at.eq(delayed_events::delay_ms + now), + )) + .get_result::(&mut *conn) + .await + .optional() + }) + .await + .map_err(Into::into) +} + +/// Record the successful outcome while the caller still holds the row lock. +pub async fn set_sent_locked( + conn: &mut AsyncPgConnection, + row_id: i64, + event_id: &EventId, + now: i64, +) -> DataResult<()> { + diesel::update(delayed_events::table.find(row_id)) + .set(( + delayed_events::event_id.eq(event_id), + delayed_events::finalized_at.eq(now), + )) + .execute(conn) + .await?; + Ok(()) +} + +/// Record a scheduled-send failure while the caller still holds the row lock. +pub async fn set_error_locked( + conn: &mut AsyncPgConnection, + row_id: i64, + error: &JsonValue, + now: i64, +) -> DataResult<()> { + diesel::update(delayed_events::table.find(row_id)) + .set(( + delayed_events::error.eq(error), + delayed_events::finalized_at.eq(now), + )) + .execute(conn) + .await?; + Ok(()) +} + +/// Cancel a scheduled delayed event. Returns `true` if the event was +/// cancelled, `false` if it did not exist unfinalized (caller decides between +/// idempotent success and conflict from the row's current state). +pub async fn cancel( + conn: &mut AsyncPgConnection, + user_id: &UserId, + delay_id: &str, +) -> DataResult { + conn.transaction::<_, diesel::result::Error, _>(async |conn| { + // As with restart, wait for an in-flight sender before timestamping the + // outcome so `finalized_at` reflects when cancellation actually won. + let Some(row) = delayed_events::table + .filter(delayed_events::user_id.eq(user_id)) + .filter(delayed_events::delay_id.eq(delay_id)) + .for_update() + .first::(&mut *conn) + .await + .optional()? + else { + return Ok(false); + }; + if row.finalized_at.is_some() { + return Ok(false); + } + + let now = UnixMillis::now().get() as i64; + let count = diesel::update(delayed_events::table.find(row.id)) + .filter(delayed_events::finalized_at.is_null()) + .set(delayed_events::finalized_at.eq(now)) + .execute(&mut *conn) + .await?; + Ok(count > 0) + }) + .await + .map_err(Into::into) +} + +/// Delete finalized delayed events whose retention period has passed. +pub async fn prune_finalized(finalized_before: i64) -> DataResult { + let mut conn = connect().await?; + conn.transaction::<_, diesel::result::Error, _>(async |conn| { + // The normal send endpoint also records the transaction in + // `event_idempotents`. Remove that mapping with the delayed row, but + // only when it points at this delayed event's actual output. Leaving a + // stale mapping would let a later retry schedule a new delayed event + // while ordinary transaction lookup still points at the old output. + diesel::sql_query( + "DELETE FROM event_idempotents AS idempotent \ + USING delayed_events AS delayed \ + WHERE delayed.finalized_at IS NOT NULL \ + AND delayed.finalized_at <= $1 \ + AND delayed.event_id IS NOT NULL \ + AND idempotent.event_id = delayed.event_id \ + AND idempotent.user_id = delayed.user_id \ + AND idempotent.device_id IS NOT DISTINCT FROM delayed.device_id \ + AND idempotent.room_id = delayed.room_id \ + AND idempotent.txn_id = delayed.txn_id", + ) + .bind::(finalized_before) + .execute(&mut *conn) + .await?; + + diesel::delete( + delayed_events::table + .filter(delayed_events::finalized_at.is_not_null()) + .filter(delayed_events::finalized_at.le(finalized_before)), + ) + .execute(&mut *conn) + .await + }) + .await + .map_err(Into::into) +} diff --git a/crates/data/src/schema.rs b/crates/data/src/schema.rs index ba3a645d6..e8d8aede8 100644 --- a/crates/data/src/schema.rs +++ b/crates/data/src/schema.rs @@ -31,6 +31,31 @@ diesel::table! { } } +diesel::table! { + use diesel::sql_types::*; + use crate::full_text_search::*; + + delayed_events (id) { + id -> Int8, + delay_id -> Text, + user_id -> Text, + device_id -> Nullable, + room_id -> Text, + event_type -> Text, + state_key -> Nullable, + content -> Jsonb, + delay_ms -> Int8, + txn_id -> Text, + origin_server_ts -> Nullable, + running_since -> Int8, + send_at -> Int8, + event_id -> Nullable, + error -> Nullable, + finalized_at -> Nullable, + created_at -> Int8, + } +} + diesel::table! { use diesel::sql_types::*; use crate::full_text_search::*; @@ -1160,6 +1185,7 @@ diesel::joinable!(user_ratelimit_override -> users (user_id)); diesel::allow_tables_to_appear_in_same_query!( appservice_registrations, banned_rooms, + delayed_events, device_inboxes, device_streams, e2e_cross_signing_keys, diff --git a/crates/server/src/config.rs b/crates/server/src/config.rs index 1be898865..7185d31b8 100644 --- a/crates/server/src/config.rs +++ b/crates/server/src/config.rs @@ -27,6 +27,8 @@ mod compression; pub use compression::*; mod db; pub use db::*; +mod delayed_event; +pub use delayed_event::*; // mod dns; // pub use dns::*; mod federation; diff --git a/crates/server/src/config/delayed_event.rs b/crates/server/src/config/delayed_event.rs new file mode 100644 index 000000000..d4452054f --- /dev/null +++ b/crates/server/src/config/delayed_event.rs @@ -0,0 +1,76 @@ +use serde::Deserialize; + +use crate::macros::config_example; + +#[config_example(filename = "palpo-example.toml", section = "delayed_events")] +#[derive(Clone, Debug, Deserialize)] +pub struct DelayedEventsConfig { + /// Allow scheduling MSC4140 delayed events. + /// + /// Delayed events let clients schedule message or state events that the + /// server sends into a room after a delay, e.g. reliable MatrixRTC + /// "hang up" events. When disabled the endpoints are not registered and + /// the feature is not advertised. Disabled by default while MSC4140 is + /// unstable. + /// + /// default: false + #[serde(default)] + pub enable: bool, + + /// The maximum delay in milliseconds a client may request for a delayed + /// event. Requests above this limit are rejected with `M_DELAY_TOO_LARGE`. + /// Defaults to 24 hours. + /// + /// default: 86400_000 + #[serde(default = "default_max_delay_ms")] + pub max_delay_ms: u64, + + /// How many delayed events a user may have scheduled at once. Requests + /// above this limit are rejected with `M_LIMIT_EXCEEDED`. + /// Defaults to 100. + /// + /// default: 100 + #[serde(default = "default_max_scheduled")] + pub max_scheduled: u64, + + /// How long finalized (sent, cancelled, or errored) delayed events are + /// retained for lookup before they are pruned, in milliseconds. + /// Defaults to 7 days. + /// + /// default: 604800_000 + #[serde(default = "default_retention_ms")] + pub retention_ms: u64, +} + +impl Default for DelayedEventsConfig { + fn default() -> Self { + Self { + enable: false, + max_delay_ms: default_max_delay_ms(), + max_scheduled: default_max_scheduled(), + retention_ms: default_retention_ms(), + } + } +} + +fn default_max_delay_ms() -> u64 { + 24 * 60 * 60_000 +} + +fn default_max_scheduled() -> u64 { + 100 +} + +fn default_retention_ms() -> u64 { + 7 * 24 * 60 * 60_000 +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn delayed_events_are_disabled_by_default() { + assert!(!DelayedEventsConfig::default().enable); + } +} diff --git a/crates/server/src/config/server.rs b/crates/server/src/config/server.rs index ea826abac..573045638 100644 --- a/crates/server/src/config/server.rs +++ b/crates/server/src/config/server.rs @@ -5,10 +5,10 @@ use salvo::http::HeaderValue; use serde::Deserialize; use super::{ - AdminConfig, BlurhashConfig, CompressionConfig, DbConfig, DelegatedAuthConfig, - FederationConfig, HttpClientConfig, JwtConfig, LoggerConfig, MediaConfig, OidcConfig, - PresenceConfig, ProxyConfig, ReadReceiptConfig, StorageConfig, TurnConfig, TypingConfig, - UrlPreviewConfig, WellKnownConfig, + AdminConfig, BlurhashConfig, CompressionConfig, DbConfig, DelayedEventsConfig, + DelegatedAuthConfig, FederationConfig, HttpClientConfig, JwtConfig, LoggerConfig, MediaConfig, + OidcConfig, PresenceConfig, ProxyConfig, ReadReceiptConfig, StorageConfig, TurnConfig, + TypingConfig, UrlPreviewConfig, WellKnownConfig, }; use crate::core::serde::{default_false, default_true}; use crate::core::{OwnedRoomOrAliasId, OwnedServerName, RoomVersionId}; @@ -75,7 +75,8 @@ impl ListenerConfig { ### https://palpo.im/guide/configuration.html "#, ignore = "federation well_known compression typing read_receipt presence \ - admin url_preview turn media storage blurhash keypair ldap proxy jwt oidc logger db appservice" + admin url_preview turn media storage blurhash keypair ldap proxy jwt oidc logger db appservice \ + delayed_events" )] #[derive(Clone, Debug, Deserialize)] pub struct ServerConfig { @@ -725,6 +726,10 @@ pub struct ServerConfig { #[serde(default)] pub admin: AdminConfig, + // external structure; separate section + #[serde(default)] + pub delayed_events: DelayedEventsConfig, + // external structure; separate section #[serde(default)] pub presence: PresenceConfig, diff --git a/crates/server/src/delayed_event.rs b/crates/server/src/delayed_event.rs new file mode 100644 index 000000000..97731523f --- /dev/null +++ b/crates/server/src/delayed_event.rs @@ -0,0 +1,646 @@ +//! MSC4140 delayed events. +//! +//! Scheduling, management actions (`restart`/`send`/`cancel`), and the +//! background scheduler that sends events into their room once the delay +//! elapses. Scheduled events are persisted, so pending delayed events survive +//! restarts: on startup the scheduler picks up overdue events and sends them +//! in chronological order of their scheduled send times. +//! +//! Power levels and other auth rules are deliberately evaluated only at the +//! point of sending, as the MSC requires. + +use std::collections::BTreeMap; +use std::sync::OnceLock; +use std::time::Duration; + +use diesel_async::{AsyncConnection, AsyncPgConnection, RunQueryDsl}; +use salvo::http::StatusCode; +use serde_json::value::to_raw_value; +use tokio::sync::{Notify, Semaphore}; + +use crate::core::client::delayed_events::{DelayedEventData, DelayedEventError, UpdateAction}; +use crate::core::error::{ErrorKind, RetryAfter}; +use crate::core::events::{StateEventType, TimelineEventType}; +use crate::core::identifiers::*; +use crate::core::serde::{JsonValue, to_canonical_value}; +use crate::core::{MatrixError, UnixMillis}; +use crate::data::room::delayed_event::{self, DbDelayedEvent, NewDbDelayedEvent}; +use crate::room::timeline; +use crate::{AppError, AppResult, PduBuilder, config, room, utils}; + +/// Retention sweep cadence for finalized delayed events. +const PRUNE_INTERVAL: Duration = Duration::from_secs(60 * 60); +/// Upper bound on the scheduler's sleep so newly due work is never missed for +/// long even if a wakeup signal is lost. +const MAX_IDLE: Duration = Duration::from_secs(60); +/// Avoid a hot loop when another server process holds the earliest due row. +const LOCK_RETRY: Duration = Duration::from_secs(1); + +static WAKEUP: OnceLock = OnceLock::new(); +static OPERATION_GATE: OnceLock = OnceLock::new(); + +fn wakeup() -> &'static Notify { + WAKEUP.get_or_init(Notify::new) +} + +/// Bound both delayed appends and row-locking management operations. Appends +/// perform nested queries through the shared pool, while the same limit keeps +/// management requests from opening an unbounded number of dedicated database +/// connections when they wait for an in-flight sender's row lock. +fn operation_gate() -> &'static Semaphore { + OPERATION_GATE.get_or_init(|| { + let permits = config::get().db.pool_size.saturating_sub(1).max(1) as usize; + Semaphore::new(permits) + }) +} + +/// Open a connection outside the shared pool for a transaction that holds a +/// delayed-event row lock while the timeline append uses pooled connections. +/// Management actions use the same path, so waiting on that row can never +/// consume the pool capacity needed by the sender that owns it. +async fn dedicated_delayed_event_connection() -> AppResult { + let db_config = config::get().db.clone().into_data_db_config(); + let url = crate::data::connection_url(&db_config, &db_config.url); + let mut conn = AsyncPgConnection::establish(&url) + .await + .map_err(|error| AppError::internal(format!("failed to connect to database: {error}")))?; + let statement_timeout = db_config.statement_timeout.min(3_600_000); + diesel::sql_query(format!("SET statement_timeout = {statement_timeout}")) + .execute(&mut conn) + .await?; + Ok(conn) +} + +/// Start the background scheduler that sends due delayed events. +pub fn start() { + tokio::spawn(async move { + let mut prune = tokio::time::interval(PRUNE_INTERVAL); + loop { + if let Err(error) = process_due_events().await { + tracing::warn!(?error, "failed to process due delayed events"); + } + + let now = UnixMillis::now().get() as i64; + let next_wake = match delayed_event::next_send_at().await { + Ok(send_at) => send_at, + Err(error) => { + tracing::warn!(?error, "failed to load next delayed event wake-up time"); + None + } + }; + let sleep = next_wake + .map(|at| { + let until = Duration::from_millis(at.saturating_sub(now) as u64); + if until.is_zero() { + LOCK_RETRY + } else { + until.min(MAX_IDLE) + } + }) + .unwrap_or(MAX_IDLE); + tokio::select! { + _ = wakeup().notified() => {}, + _ = tokio::time::sleep(sleep) => {}, + _ = prune.tick() => { + let conf = config::get(); + let retention_ms = i64::try_from(conf.delayed_events.retention_ms) + .unwrap_or(i64::MAX); + let cutoff = (UnixMillis::now().get() as i64).saturating_sub(retention_ms); + if let Err(error) = delayed_event::prune_finalized(cutoff).await { + tracing::warn!(?error, "failed to prune finalized delayed events"); + } + }, + } + } + }); +} + +/// Send every delayed event that is due, in chronological order of scheduled +/// send times. Failures are recorded on the event instead of being retried, +/// per the MSC. +async fn process_due_events() -> AppResult<()> { + while process_one_due_event().await? {} + Ok(()) +} + +/// Lock, send, and finalize one due row in a single database transaction. +async fn process_one_due_event() -> AppResult { + let _permit = operation_gate() + .acquire() + .await + .expect("the delayed-event operation semaphore is never closed"); + let mut conn = dedicated_delayed_event_connection().await?; + conn.transaction::<_, AppError, _>(async |conn| { + let now = UnixMillis::now().get() as i64; + let Some(event) = delayed_event::lock_next_due(conn, now).await? else { + return Ok(false); + }; + + match send_delayed_pdu(&event).await { + Ok(event_id) => { + delayed_event::set_sent_locked( + conn, + event.id, + &event_id, + UnixMillis::now().get() as i64, + ) + .await?; + } + Err(error) => { + // `build_and_append_pdu` can report a later delivery or + // bookkeeping failure after the event has already entered the + // timeline. The promotion trigger is the authoritative commit + // point, so do not misclassify that case as a failed send. + if let Some(event_id) = delayed_event::get_output(&event.delay_id).await? { + tracing::warn!( + delay_id = %event.delay_id, + room_id = %event.room_id, + %event_id, + ?error, + "delayed event entered the timeline before a later send step failed" + ); + delayed_event::set_sent_locked( + conn, + event.id, + &event_id, + UnixMillis::now().get() as i64, + ) + .await?; + } else { + tracing::debug!( + delay_id = %event.delay_id, + room_id = %event.room_id, + ?error, + "delayed event failed to send at its scheduled time" + ); + // The MSC says a scheduled send is not retried. The error + // is committed under the same row lock that fenced the + // append. + delayed_event::set_error_locked( + conn, + event.id, + &error_body(error), + UnixMillis::now().get() as i64, + ) + .await?; + } + } + } + Ok(true) + }) + .await +} + +/// Build and append the PDU for a locked delayed event through the normal +/// event authorization and federation paths. +/// +/// A send interrupted after the append but before the outcome is recorded +/// leaves the delayed row scheduled. Database triggers track tentative +/// outliers and atomically confirm the output when it enters the timeline, so +/// recovery can replace an abandoned outlier but can never promote a second +/// event for the same delay id. +async fn send_delayed_pdu(event: &DbDelayedEvent) -> AppResult { + let event_type: TimelineEventType = event.event_type.clone().into(); + let state_lock = room::lock_state(&event.room_id).await; + + // This runs before the state check below: recovering a state event that + // already reached the room must not re-run authorization, because the + // event it just sent may itself have changed the state that check reads, + // which would turn a completed send into a permanent failure. + // Use the delay-specific mapping: a reused transaction id can point at + // an older ordinary send, while this marker identifies the exact delayed + // event that actually entered the timeline. + if let Some(event_id) = delayed_event::get_output(&event.delay_id).await? { + // Repair the conventional transaction-id lookup when possible. The + // trigger-backed output is already a sufficient idempotency fence, so + // failure to write this secondary mapping must not turn a completed + // room append into a failed delayed event. + if let Err(error) = crate::transaction_id::add_txn_id( + &event.txn_id, + &event.user_id, + event.device_id.as_deref(), + Some(&event.room_id), + Some(&event_id), + ) + .await + { + tracing::warn!( + delay_id = %event.delay_id, + %event_id, + ?error, + "failed to repair delayed-event transaction-id mapping" + ); + } + return Ok(event_id); + } + // A delayed send does not pass through the access-token hoop again. Apply + // the same current account-usability policy explicitly so deactivated, + // locked, or suspended users cannot emit previously queued events. + let user = crate::data::user::get_user(&event.user_id).await?; + crate::user::ensure_account_usable(&user)?; + + // Re-evaluate server-side send policy as well as room authorization. An + // event scheduled while encryption was enabled must not bypass a later + // administrator decision to disable encrypted messages. This belongs + // after output recovery so an event that already entered the timeline is + // still finalized correctly. + if event_type == TimelineEventType::RoomEncrypted && !config::get().allow_encryption { + return Err(MatrixError::forbidden("Encryption has been disabled", None).into()); + } + if let Some(state_key) = &event.state_key { + let state_event_type: StateEventType = event.event_type.clone().into(); + crate::state::allowed_to_send_state_event( + &event.room_id, + &state_event_type, + state_key, + &serde_json::from_value(event.content.clone())?, + ) + .await?; + } + + let mut unsigned = BTreeMap::new(); + unsigned.insert( + "org.matrix.msc4140.delay_id".to_owned(), + to_raw_value(&event.delay_id)?, + ); + unsigned.insert("transaction_id".to_owned(), to_raw_value(&event.txn_id)?); + + let event_id = timeline::build_and_append_pdu_force( + PduBuilder { + event_type, + content: to_raw_value(&event.content)?, + unsigned, + state_key: event.state_key.clone(), + redacts: None, + timestamp: event.origin_server_ts.map(|ts| UnixMillis(ts as u64)), + }, + &event.user_id, + &event.room_id, + &crate::room::get_version(&event.room_id).await?, + &state_lock, + ) + .await? + .pdu + .event_id; + + // The database trigger has already recorded the authoritative output in + // the same transaction that promoted the event into the timeline. Keep the + // standard transaction-id mapping for normal idempotency lookups, but do + // not misreport a completed room append if this secondary write fails. + if let Err(error) = crate::transaction_id::add_txn_id( + &event.txn_id, + &event.user_id, + event.device_id.as_deref(), + Some(&event.room_id), + Some(&event_id), + ) + .await + { + tracing::warn!( + delay_id = %event.delay_id, + event_id = %event_id, + ?error, + "failed to record delayed-event transaction-id mapping" + ); + } + drop(state_lock); + + Ok((*event_id).to_owned()) +} + +/// Schedule a new delayed event, enforcing the configured limits. Returns the +/// `delay_id`, reusing the one from a previous identical transaction for +/// idempotency. +pub async fn schedule( + user_id: &UserId, + device_id: Option<&DeviceId>, + is_appservice: bool, + room_id: &RoomId, + event_type: &TimelineEventType, + txn_id: &TransactionId, + timestamp: Option, + delay: Duration, + state_key: Option, + content: JsonValue, +) -> AppResult { + let conf = config::get(); + let now = UnixMillis::now().get() as i64; + + // An already accepted transaction stays idempotent even if server limits, + // room state, or feature-related configuration changed since the original + // request. `create` repeats this lookup under its advisory lock to close + // the concurrent-first-request race. + if let Some(existing) = delayed_event::get_by_txn_id(user_id, device_id, txn_id).await? { + return Ok(existing.delay_id); + } + + let requested_delay_ms = delay.as_millis(); + if requested_delay_ms == 0 { + return Err( + MatrixError::invalid_param("delay must be a positive number of milliseconds").into(), + ); + } + let max_delay_ms = + u128::from(conf.delayed_events.max_delay_ms).min(i64::MAX.saturating_sub(now) as u128); + if requested_delay_ms > max_delay_ms { + return Err(MatrixError::delay_too_large(format!( + "the requested delay exceeds the maximum allowed delay of {} ms", + max_delay_ms + )) + .into()); + } + let delay_ms = i64::try_from(requested_delay_ms) + .map_err(|_| MatrixError::invalid_param("delay is too large"))?; + + if !content.is_object() { + return Err(MatrixError::bad_json("event content is not an object").into()); + } + to_canonical_value(&content).map_err(|e| { + MatrixError::bad_json(format!("event content is not valid canonical JSON: {e}")) + })?; + + // Forbid m.room.encrypted if encryption is disabled, matching /send. + if event_type == &TimelineEventType::RoomEncrypted && !conf.allow_encryption { + return Err(MatrixError::forbidden("Encryption has been disabled", None).into()); + } + + // The room must be known; auth rules themselves are evaluated at send time. + crate::room::get_version(room_id).await?; + + let origin_server_ts = if is_appservice { + timestamp + .map(|ts| { + i64::try_from(ts.get()) + .map_err(|_| MatrixError::invalid_param("timestamp is too large")) + }) + .transpose()? + } else { + None + }; + let new = NewDbDelayedEvent { + delay_id: utils::random_string(18), + user_id: user_id.to_owned(), + device_id: device_id.map(|d| d.to_owned()), + room_id: room_id.to_owned(), + event_type: event_type.to_string(), + state_key, + content, + delay_ms, + txn_id: txn_id.to_owned(), + origin_server_ts, + running_since: now, + send_at: now + delay_ms, + created_at: now, + }; + // The limit is enforced inside the same transaction as the insert, so + // concurrent requests cannot each observe a count below it and all succeed. + let max_scheduled = i64::try_from(conf.delayed_events.max_scheduled).unwrap_or(i64::MAX); + match delayed_event::create(new, max_scheduled).await? { + delayed_event::Scheduled::Created(row) => { + wakeup().notify_one(); + Ok(row.delay_id) + } + // A concurrent retry of this same transaction already scheduled it. + delayed_event::Scheduled::AlreadyScheduled(row) => Ok(row.delay_id), + delayed_event::Scheduled::LimitReached => { + let retry_after = delayed_event::next_send_at_of_user(user_id) + .await? + .map(|send_at| { + RetryAfter::Delay(Duration::from_millis(send_at.saturating_sub(now) as u64)) + }); + Err(MatrixError::limit_exceeded( + "The maximum number of delayed events has been reached.", + retry_after, + ) + .into()) + } + } +} + +/// Apply a management action (`restart`/`send`/`cancel`) to a delayed event. +pub async fn update(user_id: &UserId, delay_id: &str, action: &UpdateAction) -> AppResult<()> { + let Some(event) = delayed_event::get_by_delay_id(user_id, delay_id).await? else { + return Err(MatrixError::not_found("no delayed event with that delay_id was found").into()); + }; + match action { + UpdateAction::Restart => { + let _permit = operation_gate() + .acquire() + .await + .expect("the delayed-event operation semaphore is never closed"); + let mut conn = dedicated_delayed_event_connection().await?; + if delayed_event::restart(&mut conn, user_id, delay_id) + .await? + .is_some() + { + wakeup().notify_one(); + Ok(()) + } else { + let refreshed = delayed_event::get_by_delay_id(user_id, delay_id) + .await? + .unwrap_or(event); + Err(finalized_conflict(&refreshed, "restart")) + } + } + UpdateAction::Send => send_now(user_id, delay_id).await, + UpdateAction::Cancel => { + let _permit = operation_gate() + .acquire() + .await + .expect("the delayed-event operation semaphore is never closed"); + let mut conn = dedicated_delayed_event_connection().await?; + if delayed_event::cancel(&mut conn, user_id, delay_id).await? { + Ok(()) + } else { + let refreshed = delayed_event::get_by_delay_id(user_id, delay_id) + .await? + .unwrap_or(event); + if refreshed.event_id.is_some() { + Err(finalized_conflict(&refreshed, "cancel")) + } else { + // The MSC treats cancelling an event that was already + // cancelled -- "either due to user action or an error" -- + // as an idempotent success. + Ok(()) + } + } + } + _ => Err(MatrixError::invalid_param("unknown delayed event action").into()), + } +} + +/// Manually send one row while holding its database lock through the append. +async fn send_now(user_id: &UserId, delay_id: &str) -> AppResult<()> { + let _permit = operation_gate() + .acquire() + .await + .expect("the delayed-event operation semaphore is never closed"); + let mut conn = dedicated_delayed_event_connection().await?; + conn.transaction::<_, AppError, _>(async |conn| { + let Some(event) = delayed_event::lock_for_send(conn, user_id, delay_id).await? else { + return Err( + MatrixError::not_found("no delayed event with that delay_id was found").into(), + ); + }; + + if event.finalized_at.is_some() { + return if event.event_id.is_some() { + Ok(()) + } else { + Err(finalized_conflict(&event, "send")) + }; + } + + match send_delayed_pdu(&event).await { + Ok(event_id) => { + delayed_event::set_sent_locked( + conn, + event.id, + &event_id, + UnixMillis::now().get() as i64, + ) + .await?; + Ok(()) + } + Err(error) => { + // The room append may already have committed before a later + // delivery or bookkeeping step failed. The trigger-backed + // output is authoritative, just as it is for scheduled sends. + if let Some(event_id) = delayed_event::get_output(&event.delay_id).await? { + tracing::warn!( + delay_id = %event.delay_id, + room_id = %event.room_id, + %event_id, + ?error, + "manually sent delayed event entered the timeline before a later step failed" + ); + delayed_event::set_sent_locked( + conn, + event.id, + &event_id, + UnixMillis::now().get() as i64, + ) + .await?; + Ok(()) + } else { + Err(error) + } + } + } + }) + .await +} + +/// List the user's scheduled delayed events in chronological send order. +pub async fn list(user_id: &UserId) -> AppResult> { + Ok(delayed_event::list_scheduled(user_id) + .await? + .into_iter() + .map(to_event_data) + .collect()) +} + +/// Fetch one delayed event owned by the user, whether scheduled or finalized. +pub async fn get(user_id: &UserId, delay_id: &str) -> AppResult { + delayed_event::get_by_delay_id(user_id, delay_id) + .await? + .map(to_event_data) + .ok_or_else(|| { + MatrixError::not_found("no delayed event with that delay_id was found").into() + }) +} + +fn to_event_data(event: DbDelayedEvent) -> DelayedEventData { + DelayedEventData { + delay_id: event.delay_id, + room_id: event.room_id, + event_type: event.event_type.into(), + state_key: event.state_key, + content: event.content, + delay: Duration::from_millis(event.delay_ms as u64), + running_since: UnixMillis(event.running_since as u64), + error: event + .error + .and_then(|error| serde_json::from_value::(error).ok()), + event_id: event.event_id, + finalized_ts: event.finalized_at.map(|ts| UnixMillis(ts as u64)), + } +} + +/// HTTP 409 for a management action that conflicts with the outcome the +/// delayed event was already finalized with. +fn finalized_conflict(event: &DbDelayedEvent, action: &str) -> AppError { + let outcome = if event.event_id.is_some() { + "already been sent" + } else if event.error.is_some() { + "already failed to send" + } else { + "already been cancelled" + }; + let mut error = MatrixError::unknown(format!( + "cannot {action} a delayed event that has {outcome}" + )); + error.status_code = Some(StatusCode::CONFLICT); + error.into() +} + +/// The standard error body stored for a delayed event that failed to send. +fn error_body(error: AppError) -> JsonValue { + match error { + AppError::Matrix(e) => { + let retry_after = match &e.kind { + ErrorKind::LimitExceeded { + retry_after: Some(RetryAfter::Delay(duration)), + } => Some(*duration), + _ => None, + }; + let mut body = serde_json::to_value(&e).unwrap_or_default(); + if let Some(map) = body.as_object_mut() { + map.insert("errcode".to_owned(), e.kind.code().to_string().into()); + if let Some(duration) = retry_after + && let Ok(ms) = u64::try_from(duration.as_millis()) + { + map.insert("retry_after_ms".to_owned(), ms.into()); + } + } + body + } + _ => serde_json::json!({ + "errcode": "M_UNKNOWN", + "error": "internal server error", + }), + } +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use super::error_body; + use crate::AppError; + use crate::core::MatrixError; + use crate::core::error::RetryAfter; + + #[test] + fn delayed_event_internal_errors_do_not_expose_details() { + let body = error_body(AppError::internal("database secret path")); + + assert_eq!(body["errcode"], "M_UNKNOWN"); + assert_eq!(body["error"], "internal server error"); + assert!(!body.to_string().contains("database secret path")); + } + + #[test] + fn delayed_event_rate_limit_errors_keep_retry_delay() { + let body = error_body( + MatrixError::limit_exceeded( + "slow down", + Some(RetryAfter::Delay(Duration::from_millis(1500))), + ) + .into(), + ); + + assert_eq!(body["errcode"], "M_LIMIT_EXCEEDED"); + assert_eq!(body["retry_after_ms"], 1500); + } +} diff --git a/crates/server/src/event/pdu.rs b/crates/server/src/event/pdu.rs index 6a78d7fec..3808dd648 100644 --- a/crates/server/src/event/pdu.rs +++ b/crates/server/src/event/pdu.rs @@ -1,3 +1,4 @@ +use std::borrow::Cow; use std::cmp::Ordering; use std::collections::BTreeMap; use std::ops::{Deref, DerefMut}; @@ -445,11 +446,35 @@ impl PduEvent { } } - pub fn remove_transaction_id(&mut self) -> AppResult<()> { + /// Strips the `unsigned` fields that only the event's own sender may see. + /// + /// `transaction_id` is sender-only per the spec, and MSC4140 requires the same of a delayed + /// event's `delay_id`: it is included "if, and only if, the client being given the event is + /// authenticated as the event's sender". + pub fn remove_sender_only_unsigned(&mut self) -> AppResult<()> { self.unsigned.remove("transaction_id"); + self.unsigned.remove("org.matrix.msc4140.delay_id"); Ok(()) } + fn unsigned_for_recipient( + &self, + recipient: &UserId, + ) -> Cow<'_, BTreeMap>> { + if self.sender == recipient { + Cow::Borrowed(&self.unsigned) + } else { + Cow::Owned(self.unsigned_without_sender_only()) + } + } + + fn unsigned_without_sender_only(&self) -> BTreeMap> { + let mut unsigned = self.unsigned.clone(); + unsigned.remove("transaction_id"); + unsigned.remove("org.matrix.msc4140.delay_id"); + unsigned + } + pub fn add_age(&mut self) -> AppResult<()> { let now: i128 = UnixMillis::now().get().into(); let then: i128 = self.origin_server_ts.get().into(); @@ -463,6 +488,13 @@ impl PduEvent { #[tracing::instrument] pub fn to_sync_room_event(&self) -> RawJson { + self.to_sync_room_event_with_unsigned(&self.unsigned) + } + + fn to_sync_room_event_with_unsigned( + &self, + unsigned: &BTreeMap>, + ) -> RawJson { let mut json = json!({ "content": self.content, "type": self.event_ty, @@ -471,8 +503,8 @@ impl PduEvent { "origin_server_ts": self.origin_server_ts, }); - if !self.unsigned.is_empty() { - json["unsigned"] = json!(self.unsigned); + if !unsigned.is_empty() { + json["unsigned"] = json!(unsigned); } if let Some(state_key) = &self.state_key { json["state_key"] = json!(state_key); @@ -484,8 +516,19 @@ impl PduEvent { serde_json::from_value(json).expect("RawJson::from_value always works") } + pub fn to_sync_room_event_for(&self, recipient: &UserId) -> RawJson { + self.to_sync_room_event_with_unsigned(self.unsigned_for_recipient(recipient).as_ref()) + } + #[tracing::instrument] pub fn to_room_event(&self) -> RawJson { + self.to_room_event_with_unsigned(&self.unsigned) + } + + fn to_room_event_with_unsigned( + &self, + unsigned: &BTreeMap>, + ) -> RawJson { let age = UnixMillis::now() .get() .saturating_sub(self.origin_server_ts.get()); @@ -498,12 +541,12 @@ impl PduEvent { "room_id": self.room_id, }); - if self.unsigned.is_empty() { + if unsigned.is_empty() { data["unsigned"] = json!({ "age": age }); } else { - let mut unsigned = json!(self.unsigned); - unsigned["age"] = json!(age); - data["unsigned"] = unsigned; + let mut unsigned_json = json!(unsigned); + unsigned_json["age"] = json!(age); + data["unsigned"] = unsigned_json; } if let Some(state_key) = &self.state_key { data["state_key"] = json!(state_key); @@ -515,8 +558,23 @@ impl PduEvent { serde_json::from_value(data).expect("RawJson::from_value always works") } + pub fn to_room_event_for(&self, recipient: &UserId) -> RawJson { + self.to_room_event_with_unsigned(self.unsigned_for_recipient(recipient).as_ref()) + } + + pub fn to_room_event_without_sender_only_unsigned(&self) -> RawJson { + self.to_room_event_with_unsigned(&self.unsigned_without_sender_only()) + } + #[tracing::instrument] pub fn to_message_like_event(&self) -> RawJson { + self.to_message_like_event_with_unsigned(&self.unsigned) + } + + fn to_message_like_event_with_unsigned( + &self, + unsigned: &BTreeMap>, + ) -> RawJson { let mut data = json!({ "content": self.content, "type": self.event_ty, @@ -526,8 +584,8 @@ impl PduEvent { "room_id": self.room_id, }); - if !self.unsigned.is_empty() { - data["unsigned"] = json!(self.unsigned); + if !unsigned.is_empty() { + data["unsigned"] = json!(unsigned); } if let Some(state_key) = &self.state_key { data["state_key"] = json!(state_key); @@ -539,13 +597,38 @@ impl PduEvent { serde_json::from_value(data).expect("RawJson::from_value always works") } + pub fn to_message_like_event_for(&self, recipient: &UserId) -> RawJson { + self.to_message_like_event_with_unsigned(self.unsigned_for_recipient(recipient).as_ref()) + } + + pub fn to_message_like_event_without_sender_only_unsigned( + &self, + ) -> RawJson { + self.to_message_like_event_with_unsigned(&self.unsigned_without_sender_only()) + } + #[tracing::instrument] pub fn to_state_event(&self) -> RawJson { - serde_json::from_value(self.to_state_event_value()) + serde_json::from_value(self.to_state_event_value_with_unsigned(&self.unsigned)) .expect("RawJson::from_value always works") } + pub fn to_state_event_for(&self, recipient: &UserId) -> RawJson { + serde_json::from_value( + self.to_state_event_value_with_unsigned( + self.unsigned_for_recipient(recipient).as_ref(), + ), + ) + .expect("RawJson::from_value always works") + } #[tracing::instrument] pub fn to_state_event_value(&self) -> JsonValue { + self.to_state_event_value_with_unsigned(&self.unsigned) + } + + fn to_state_event_value_with_unsigned( + &self, + unsigned: &BTreeMap>, + ) -> JsonValue { let JsonValue::Object(mut data) = json!({ "content": self.content, "type": self.event_ty, @@ -558,8 +641,8 @@ impl PduEvent { panic!("Invalid JSON value, never happened!"); }; - if !self.unsigned.is_empty() { - data.insert("unsigned".into(), json!(self.unsigned)); + if !unsigned.is_empty() { + data.insert("unsigned".into(), json!(unsigned)); } for (key, value) in &self.extra_data { @@ -571,8 +654,19 @@ impl PduEvent { JsonValue::Object(data) } + pub fn to_state_event_value_for(&self, recipient: &UserId) -> JsonValue { + self.to_state_event_value_with_unsigned(self.unsigned_for_recipient(recipient).as_ref()) + } + #[tracing::instrument] pub fn to_sync_state_event(&self) -> RawJson { + self.to_sync_state_event_with_unsigned(&self.unsigned) + } + + fn to_sync_state_event_with_unsigned( + &self, + unsigned: &BTreeMap>, + ) -> RawJson { let mut data = json!({ "content": self.content, "type": self.event_ty, @@ -582,13 +676,17 @@ impl PduEvent { "state_key": self.state_key, }); - if !self.unsigned.is_empty() { - data["unsigned"] = json!(self.unsigned); + if !unsigned.is_empty() { + data["unsigned"] = json!(unsigned); } serde_json::from_value(data).expect("RawJson::from_value always works") } + pub fn to_sync_state_event_for(&self, recipient: &UserId) -> RawJson { + self.to_sync_state_event_with_unsigned(self.unsigned_for_recipient(recipient).as_ref()) + } + #[tracing::instrument] pub async fn to_stripped_state_event(&self) -> RawJson { if self.event_ty == TimelineEventType::RoomCreate { @@ -1107,3 +1205,79 @@ impl Default for PduBuilder { } } } + +#[cfg(test)] +mod sender_only_unsigned_tests { + use serde_json::value::to_raw_value; + + use super::*; + + fn event_with_sender_only_unsigned() -> PduEvent { + let mut unsigned = BTreeMap::new(); + unsigned.insert("transaction_id".to_owned(), to_raw_value("txn").unwrap()); + unsigned.insert( + "org.matrix.msc4140.delay_id".to_owned(), + to_raw_value("delay").unwrap(), + ); + unsigned.insert("age".to_owned(), to_raw_value(&10_u64).unwrap()); + + PduEvent { + event_id: "$event:example.org".try_into().unwrap(), + sender: "@alice:example.org".try_into().unwrap(), + origin_server_ts: UnixMillis(1), + event_ty: TimelineEventType::RoomMessage, + content: to_raw_value(&json!({"body": "hi", "msgtype": "m.text"})).unwrap(), + state_key: None, + room_id: "!room:example.org".try_into().unwrap(), + prev_events: Vec::new(), + depth: 1, + auth_events: Vec::new(), + redacts: None, + hashes: EventHash { + sha256: String::new(), + }, + signatures: None, + unsigned, + extra_data: Default::default(), + rejection_reason: None, + } + } + + #[test] + fn recipient_conversion_preserves_sender_only_fields_for_sender() { + let event = event_with_sender_only_unsigned(); + let sender: OwnedUserId = "@alice:example.org".try_into().unwrap(); + + let converted = event.unsigned_for_recipient(&sender); + + assert!(converted.contains_key("transaction_id")); + assert!(converted.contains_key("org.matrix.msc4140.delay_id")); + } + + #[test] + fn recipient_conversion_strips_sender_only_fields_for_other_users() { + let event = event_with_sender_only_unsigned(); + let recipient: OwnedUserId = "@bob:example.org".try_into().unwrap(); + + let converted = event.unsigned_for_recipient(&recipient); + + assert!(!converted.contains_key("transaction_id")); + assert!(!converted.contains_key("org.matrix.msc4140.delay_id")); + assert!(converted.contains_key("age")); + } + + #[test] + fn nested_message_conversion_always_strips_sender_only_fields() { + let event = event_with_sender_only_unsigned(); + + let converted = event.to_message_like_event_without_sender_only_unsigned(); + let json: JsonValue = serde_json::from_str(converted.as_str()).unwrap(); + + assert!(json.pointer("/unsigned/transaction_id").is_none()); + assert!( + json.pointer("/unsigned/org.matrix.msc4140.delay_id") + .is_none() + ); + assert_eq!(json.pointer("/unsigned/age"), Some(&json!(10))); + } +} diff --git a/crates/server/src/event/search.rs b/crates/server/src/event/search.rs index 09a5661bc..d5132eaff 100644 --- a/crates/server/src/event/search.rs +++ b/crates/server/src/event/search.rs @@ -129,7 +129,7 @@ pub async fn search_pdus( .await .unwrap_or_default(), rank: Some(rank as f64), - result: Some(pdu.to_room_event()), + result: Some(pdu.to_room_event_for(user_id)), }); } @@ -200,11 +200,11 @@ async fn calc_event_context( .map(|(sn, _)| BatchToken::new_live(*sn).to_string()), events_before: before_pdus .into_iter() - .map(|(_, pdu)| pdu.to_room_event()) + .map(|(_, pdu)| pdu.to_room_event_for(user_id)) .collect(), events_after: after_pdus .into_iter() - .map(|(_, pdu)| pdu.to_room_event()) + .map(|(_, pdu)| pdu.to_room_event_for(user_id)) .collect(), profile_info: BTreeMap::new(), }; diff --git a/crates/server/src/main.rs b/crates/server/src/main.rs index d08d23538..0d59af398 100644 --- a/crates/server/src/main.rs +++ b/crates/server/src/main.rs @@ -25,6 +25,7 @@ pub mod utils; pub use auth::{AuthArgs, AuthedInfo}; pub mod admin; pub mod appservice; +pub mod delayed_event; pub mod directory; pub mod event; pub mod exts; @@ -251,6 +252,13 @@ async fn async_main() -> Result<(), Box> { } }); + // MSC4140: send scheduled delayed events once their delay elapses; on + // startup this also recovers events that became due while the server was + // offline. + if config::get().delayed_events.enable { + crate::delayed_event::start(); + } + // MSC2444: periodically renew our outbound room peeks and drop lapsed inbound // peekers. tokio::spawn(async move { diff --git a/crates/server/src/room/pdu_metadata.rs b/crates/server/src/room/pdu_metadata.rs index 994325e0e..4afce9d4a 100644 --- a/crates/server/src/room/pdu_metadata.rs +++ b/crates/server/src/room/pdu_metadata.rs @@ -99,7 +99,7 @@ pub async fn paginate_relations_with_filter( let events: Vec<_> = events .into_iter() - .map(|(_, pdu)| pdu.to_message_like_event()) + .map(|(_, pdu)| pdu.to_message_like_event_for(user_id)) .collect(); Ok(RelationEventsResBody { @@ -138,7 +138,7 @@ pub async fn get_relations( for relation in relations { if let Ok(mut pdu) = timeline::get_pdu(&relation.child_id).await { if pdu.sender != user_id { - pdu.remove_transaction_id()?; + pdu.remove_sender_only_unsigned()?; } if pdu.user_can_see(user_id).await.unwrap_or(false) { pdus.push((relation.child_sn, pdu)); diff --git a/crates/server/src/room/thread.rs b/crates/server/src/room/thread.rs index f84bed870..80f317659 100644 --- a/crates/server/src/room/thread.rs +++ b/crates/server/src/room/thread.rs @@ -43,7 +43,7 @@ pub async fn add_to_thread(thread_id: &EventId, pdu: &SnPduEvent) -> AppResult<( { // Thread already existed relations.count += 1; - relations.latest_event = pdu.to_message_like_event(); + relations.latest_event = pdu.to_message_like_event_without_sender_only_unsigned(); let content = serde_json::to_value(relations).expect("to_value always works"); @@ -56,7 +56,7 @@ pub async fn add_to_thread(thread_id: &EventId, pdu: &SnPduEvent) -> AppResult<( } else { // New thread let relations = BundledThread { - latest_event: pdu.to_message_like_event(), + latest_event: pdu.to_message_like_event_without_sender_only_unsigned(), count: 1, current_user_participated: true, }; diff --git a/crates/server/src/room/timeline.rs b/crates/server/src/room/timeline.rs index 3426d4fcf..632022416 100644 --- a/crates/server/src/room/timeline.rs +++ b/crates/server/src/room/timeline.rs @@ -800,7 +800,41 @@ pub async fn build_and_append_pdu( room_version: &RoomVersionId, state_lock: &RoomMutexGuard, ) -> AppResult { - if let Some(state_key) = &pdu_builder.state_key + build_and_append_pdu_inner( + pdu_builder, + sender, + room_id, + room_version, + state_lock, + false, + ) + .await +} + +/// Creates and appends a PDU even when an equivalent state event is already +/// current. Delayed events need this so each successful delay id is observable +/// on its own timeline event. +#[tracing::instrument(skip_all)] +pub async fn build_and_append_pdu_force( + pdu_builder: PduBuilder, + sender: &UserId, + room_id: &RoomId, + room_version: &RoomVersionId, + state_lock: &RoomMutexGuard, +) -> AppResult { + build_and_append_pdu_inner(pdu_builder, sender, room_id, room_version, state_lock, true).await +} + +async fn build_and_append_pdu_inner( + pdu_builder: PduBuilder, + sender: &UserId, + room_id: &RoomId, + room_version: &RoomVersionId, + state_lock: &RoomMutexGuard, + force: bool, +) -> AppResult { + if !force + && let Some(state_key) = &pdu_builder.state_key && let Ok(curr_state) = super::get_state( room_id, &pdu_builder.event_type.to_string().into(), diff --git a/crates/server/src/room/timeline/stream.rs b/crates/server/src/room/timeline/stream.rs index b6f627f1d..a8774d544 100644 --- a/crates/server/src/room/timeline/stream.rs +++ b/crates/server/src/room/timeline/stream.rs @@ -203,7 +203,7 @@ pub async fn load_pdus( } if let Some(user_id) = user_id { if pdu.sender != user_id { - pdu.remove_transaction_id()?; + pdu.remove_sender_only_unsigned()?; } let _ = pdu.add_unsigned_membership(user_id).await; } diff --git a/crates/server/src/room/timeline/topolo.rs b/crates/server/src/room/timeline/topolo.rs index ba7cd5c38..91f9c2e55 100644 --- a/crates/server/src/room/timeline/topolo.rs +++ b/crates/server/src/room/timeline/topolo.rs @@ -226,7 +226,7 @@ pub async fn load_pdus( continue; } if pdu.sender != user_id { - pdu.remove_transaction_id()?; + pdu.remove_sender_only_unsigned()?; } pdu.add_unsigned_membership(user_id).await?; } diff --git a/crates/server/src/routing/client.rs b/crates/server/src/routing/client.rs index 9203eb5b3..a4c963af0 100644 --- a/crates/server/src/routing/client.rs +++ b/crates/server/src/routing/client.rs @@ -2,6 +2,7 @@ mod account; mod admin; mod appservice; mod auth; +mod delayed_event; mod device; mod directory; mod key; @@ -29,6 +30,7 @@ use std::collections::BTreeMap; use salvo::oapi::extract::*; use salvo::prelude::*; +use serde_json::json; use crate::config; use crate::core::client::discovery::capabilities::{ @@ -41,6 +43,10 @@ use crate::core::client::search::{ResultCategories, SearchReqArgs, SearchReqBody use crate::routing::prelude::*; pub fn router() -> Router { + router_with_delayed_events(config::get().delayed_events.enable) +} + +fn router_with_delayed_events(delayed_events: bool) -> Router { let mut client = Router::with_path("client").oapi_tag("client"); for v in ["v3", "v1", "r0"] { client = client @@ -127,7 +133,7 @@ pub fn router() -> Router { .push(Router::with_path("callback").get(oidc::oidc_callback)) .push(Router::with_path("login").post(oidc::oidc_login)), ) - .push(unstable::router()) + .push(unstable::router(delayed_events)) } /// #POST /_matrix/client/r0/search @@ -168,21 +174,30 @@ fn get_capabilities(_aa: AuthArgs, depot: &mut Depot) -> JsonResult JsonResult JsonResult { - json_ok(supported_versions_body()) + json_ok(supported_versions_body(config::get().delayed_events.enable)) } /// Client-Server specification versions whose behavior has been reviewed for @@ -209,37 +224,46 @@ const SUPPORTED_MATRIX_VERSIONS: &[&str] = &[ "v1.10", "v1.11", "v1.12", ]; -fn supported_versions_body() -> VersionsResBody { +/// Builds the `/versions` body. +/// +/// `delayed_events` is passed in rather than read from the global config so this stays a pure +/// function that unit tests can drive both ways. +fn supported_versions_body(delayed_events: bool) -> VersionsResBody { + let mut unstable_features = BTreeMap::from_iter([ + ("org.matrix.e2e_cross_signing".to_owned(), true), + ("org.matrix.msc2285.stable".to_owned(), true), /* private read receipts (https://github.com/matrix-org/matrix-spec-proposals/pull/2285) */ + ("uk.half-shot.msc2666.query_mutual_rooms".to_owned(), true), /* query mutual rooms (https://github.com/matrix-org/matrix-spec-proposals/pull/2666) */ + ( + "uk.half-shot.msc2666.query_mutual_rooms.stable".to_owned(), + true, + ), + ("org.matrix.msc2836".to_owned(), true), /* threading/threads (https://github.com/matrix-org/matrix-spec-proposals/pull/2836) */ + ("org.matrix.msc2946".to_owned(), true), /* spaces/hierarchy summaries (https://github.com/matrix-org/matrix-spec-proposals/pull/2946) */ + ("org.matrix.msc3026.busy_presence".to_owned(), true), /* busy presence status (https://github.com/matrix-org/matrix-spec-proposals/pull/3026) */ + ("org.matrix.msc3827".to_owned(), true), /* filtering of /publicRooms by room type (https://github.com/matrix-org/matrix-spec-proposals/pull/3827) */ + ("org.matrix.msc3952_intentional_mentions".to_owned(), true), /* intentional mentions (https://github.com/matrix-org/matrix-spec-proposals/pull/3952) */ + ("org.matrix.msc3575".to_owned(), true), /* sliding sync (https://github.com/matrix-org/matrix-spec-proposals/pull/3575/files#r1588877046) */ + ("org.matrix.msc3916.stable".to_owned(), true), /* authenticated media (https://github.com/matrix-org/matrix-spec-proposals/pull/3916) */ + ("org.matrix.msc4180".to_owned(), true), /* stable flag for 3916 (https://github.com/matrix-org/matrix-spec-proposals/pull/4180) */ + ("uk.tcpip.msc4133".to_owned(), true), /* Extending User Profile API with Key:Value Pairs (https://github.com/matrix-org/matrix-spec-proposals/pull/4133) */ + ("uk.tcpip.msc4133.stable".to_owned(), true), // profile fields also use stable `/v3` routes + ("us.cloke.msc4175".to_owned(), true), /* Profile field for user time zone (https://github.com/matrix-org/matrix-spec-proposals/pull/4175) */ + ("org.matrix.simplified_msc3575".to_owned(), true), /* Simplified Sliding sync (https://github.com/matrix-org/matrix-spec-proposals/pull/4186) */ + ("uk.timedout.msc4323".to_owned(), true), // Account suspension and locking. + ("net.zemos.msc4383".to_owned(), true), /* Homeserver implementation metadata (https://github.com/matrix-org/matrix-spec-proposals/pull/4383) */ + ]); + + if delayed_events { + // delayed events (https://github.com/matrix-org/matrix-spec-proposals/pull/4140) + unstable_features.insert("org.matrix.msc4140".to_owned(), true); + } + VersionsResBody { versions: SUPPORTED_MATRIX_VERSIONS .iter() .map(|version| (*version).to_owned()) .collect(), - unstable_features: BTreeMap::from_iter([ - ("org.matrix.e2e_cross_signing".to_owned(), true), - ("org.matrix.msc2285.stable".to_owned(), true), /* private read receipts (https://github.com/matrix-org/matrix-spec-proposals/pull/2285) */ - ("uk.half-shot.msc2666.query_mutual_rooms".to_owned(), true), /* query mutual rooms (https://github.com/matrix-org/matrix-spec-proposals/pull/2666) */ - ( - "uk.half-shot.msc2666.query_mutual_rooms.stable".to_owned(), - true, - ), - ("org.matrix.msc2836".to_owned(), true), /* threading/threads (https://github.com/matrix-org/matrix-spec-proposals/pull/2836) */ - ("org.matrix.msc2946".to_owned(), true), /* spaces/hierarchy summaries (https://github.com/matrix-org/matrix-spec-proposals/pull/2946) */ - ("org.matrix.msc3026.busy_presence".to_owned(), true), /* busy presence status (https://github.com/matrix-org/matrix-spec-proposals/pull/3026) */ - ("org.matrix.msc3827".to_owned(), true), /* filtering of /publicRooms by room type (https://github.com/matrix-org/matrix-spec-proposals/pull/3827) */ - ("org.matrix.msc3952_intentional_mentions".to_owned(), true), /* intentional mentions (https://github.com/matrix-org/matrix-spec-proposals/pull/3952) */ - ("org.matrix.msc3575".to_owned(), true), /* sliding sync (https://github.com/matrix-org/matrix-spec-proposals/pull/3575/files#r1588877046) */ - ("org.matrix.msc3916.stable".to_owned(), true), /* authenticated media (https://github.com/matrix-org/matrix-spec-proposals/pull/3916) */ - ("org.matrix.msc4180".to_owned(), true), /* stable flag for 3916 (https://github.com/matrix-org/matrix-spec-proposals/pull/4180) */ - ("uk.tcpip.msc4133".to_owned(), true), /* Extending User Profile API with Key:Value Pairs (https://github.com/matrix-org/matrix-spec-proposals/pull/4133) */ - ("uk.tcpip.msc4133.stable".to_owned(), true), /* the profile-field endpoints are - * served on the stable `/v3` prefix - * too */ - ("us.cloke.msc4175".to_owned(), true), /* Profile field for user time zone (https://github.com/matrix-org/matrix-spec-proposals/pull/4175) */ - ("org.matrix.simplified_msc3575".to_owned(), true), /* Simplified Sliding sync (https://github.com/matrix-org/matrix-spec-proposals/pull/4186) */ - ("uk.timedout.msc4323".to_owned(), true), // Account suspension and locking. - ("net.zemos.msc4383".to_owned(), true), /* Homeserver implementation metadata (https://github.com/matrix-org/matrix-spec-proposals/pull/4383) */ - ]), + unstable_features, server: Some(Server::new( "Palpo".to_owned(), crate::info::version().to_owned(), @@ -255,7 +279,7 @@ mod supported_versions_tests { #[test] fn advertised_versions_are_explicitly_reviewed() { - let body = supported_versions_body(); + let body = supported_versions_body(false); assert_eq!( body.versions, @@ -269,7 +293,7 @@ mod supported_versions_tests { #[test] fn advertises_msc4133_profile_fields_on_the_stable_prefix() { - let body = supported_versions_body(); + let body = supported_versions_body(false); assert_eq!(body.unstable_features.get("uk.tcpip.msc4133"), Some(&true)); assert_eq!( @@ -280,7 +304,7 @@ mod supported_versions_tests { #[test] fn includes_msc4383_server_metadata_and_feature_flag() { - let body = supported_versions_body(); + let body = supported_versions_body(false); let server = body.server.as_ref().unwrap(); assert_eq!(body.unstable_features.get("net.zemos.msc4383"), Some(&true)); @@ -292,9 +316,26 @@ mod supported_versions_tests { ); } + /// MSC4140 is only advertised when delayed events are actually enabled. + #[test] + fn msc4140_flag_tracks_the_delayed_events_setting() { + assert_eq!( + supported_versions_body(false) + .unstable_features + .get("org.matrix.msc4140"), + None + ); + assert_eq!( + supported_versions_body(true) + .unstable_features + .get("org.matrix.msc4140"), + Some(&true) + ); + } + #[test] fn advertises_stable_mutual_rooms_endpoint() { - let body = supported_versions_body(); + let body = supported_versions_body(false); assert_eq!( body.unstable_features @@ -309,17 +350,33 @@ mod router_tests { use salvo::http::{Method, Request}; use salvo::routing::PathState; - use super::router; + use super::router_with_delayed_events; /// Resolve `path` against the client router without running any handler. - async fn is_routed(method: Method, path: &str) -> bool { - let router = router(); + async fn is_routed_with_delayed_events( + method: Method, + path: &str, + delayed_events: bool, + ) -> bool { + let router = router_with_delayed_events(delayed_events); let mut req = Request::default(); *req.method_mut() = method; let mut path_state = PathState::from_owned_path(path.to_owned()); router.detect(&mut req, &mut path_state).await.is_some() } + async fn is_routed(method: Method, path: &str) -> bool { + is_routed_with_delayed_events(method, path, false).await + } + + #[tokio::test] + async fn delayed_event_routes_follow_config() { + let path = "/client/unstable/org.matrix.msc4140/delayed_events"; + + assert!(!is_routed_with_delayed_events(Method::GET, path, false).await); + assert!(is_routed_with_delayed_events(Method::GET, path, true).await); + } + #[tokio::test] async fn delete_devices_is_a_sibling_of_devices() { for version in ["v3", "r0"] { diff --git a/crates/server/src/routing/client/delayed_event.rs b/crates/server/src/routing/client/delayed_event.rs new file mode 100644 index 000000000..14d9d9c33 --- /dev/null +++ b/crates/server/src/routing/client/delayed_event.rs @@ -0,0 +1,112 @@ +//! Client endpoints for MSC4140 delayed events. + +use salvo::oapi::extract::*; +use salvo::prelude::*; + +use crate::core::client::delayed_events::{ + DelayedEventData, DelayedEventsResBody, SendDelayedEventReqArgs, SendDelayedEventReqBody, + SendDelayedEventResBody, UpdateDelayedEventReqArgs, UpdateDelayedEventReqBody, +}; +use crate::routing::prelude::*; + +pub(super) fn authed_router() -> Router { + Router::with_path("org.matrix.msc4140") + .push( + Router::with_path("rooms/{room_id}/delayed_event/{event_type}/{txn_id}") + .put(send_delayed_event), + ) + .push( + Router::with_path("delayed_events") + .get(list_delayed_events) + .push( + Router::with_path("{delay_id}") + .get(get_delayed_event) + .post(update_delayed_event_v1) + .push(Router::with_path("{action}").post(update_delayed_event)), + ), + ) +} + +/// `PUT /_matrix/client/unstable/org.matrix.msc4140/rooms/{room_id}/delayed_event/{event_type}/ +/// {txn_id}` +/// +/// Schedule a message or state event to be sent into the room after a delay. +#[endpoint] +async fn send_delayed_event( + _aa: AuthArgs, + args: SendDelayedEventReqArgs, + body: JsonBody, + depot: &mut Depot, +) -> JsonResult { + let authed = depot.authed_info()?; + let delay_id = crate::delayed_event::schedule( + authed.user_id(), + Some(authed.device_id()), + authed.appservice().is_some(), + &args.room_id, + &args.event_type, + &args.txn_id, + args.timestamp, + body.delay, + body.state_key.clone(), + body.content.clone(), + ) + .await?; + json_ok(SendDelayedEventResBody::new(delay_id)) +} + +/// `GET /_matrix/client/unstable/org.matrix.msc4140/delayed_events` +/// +/// List the requesting user's scheduled delayed events in chronological order +/// of their intended send time. +#[endpoint] +async fn list_delayed_events(_aa: AuthArgs, depot: &mut Depot) -> JsonResult { + let authed = depot.authed_info()?; + let delayed_events = crate::delayed_event::list(authed.user_id()).await?; + json_ok(DelayedEventsResBody::new(delayed_events)) +} + +/// `GET /_matrix/client/unstable/org.matrix.msc4140/delayed_events/{delay_id}` +/// +/// Get the details of one of the requesting user's delayed events, whether +/// still scheduled or already finalized. +#[endpoint] +async fn get_delayed_event( + _aa: AuthArgs, + delay_id: PathParam, + depot: &mut Depot, +) -> JsonResult { + let authed = depot.authed_info()?; + let delayed_event = crate::delayed_event::get(authed.user_id(), &delay_id.into_inner()).await?; + json_ok(delayed_event) +} + +/// `POST /_matrix/client/unstable/org.matrix.msc4140/delayed_events/{delay_id}/{action}` +/// +/// Restart, send, or cancel a scheduled delayed event. +#[endpoint] +async fn update_delayed_event( + _aa: AuthArgs, + args: UpdateDelayedEventReqArgs, + depot: &mut Depot, +) -> EmptyResult { + let authed = depot.authed_info()?; + crate::delayed_event::update(authed.user_id(), &args.delay_id, &args.action).await?; + empty_ok() +} + +/// `POST /_matrix/client/unstable/org.matrix.msc4140/delayed_events/{delay_id}` +/// +/// Deprecated body-action variant of the update endpoint, kept for clients +/// that implement the earlier iteration of MSC4140. +#[endpoint] +async fn update_delayed_event_v1( + _aa: AuthArgs, + delay_id: PathParam, + body: JsonBody, + depot: &mut Depot, +) -> EmptyResult { + let authed = depot.authed_info()?; + crate::delayed_event::update(authed.user_id(), &delay_id.into_inner(), &body.action).await?; + empty_ok() +} diff --git a/crates/server/src/routing/client/room.rs b/crates/server/src/routing/client/room.rs index 4b284dd63..9b69fde50 100644 --- a/crates/server/src/routing/client/room.rs +++ b/crates/server/src/routing/client/room.rs @@ -176,7 +176,7 @@ async fn initial_sync( .server_name() .is_ok_and(|server| *server != config::get().server_name) { - return remote_peek_preview(room_id, args.limit.unwrap_or(20)).await; + return remote_peek_preview(sender_id, room_id, args.limit.unwrap_or(20)).await; } return Err(MatrixError::forbidden("No room preview available.", None).into()); } @@ -195,7 +195,7 @@ async fn initial_sync( .await .unwrap_or_default() .into_values() - .map(|event| event.to_state_event()) + .map(|event| event.to_state_event_for(sender_id)) .collect::>(); let messages = PaginationChunk { @@ -212,7 +212,7 @@ async fn initial_sync( .unwrap_or_default(), chunk: events .into_iter() - .map(|(_sn, event)| event.to_room_event()) + .map(|(_sn, event)| event.to_room_event_for(sender_id)) .collect(), }; @@ -241,7 +241,11 @@ async fn initial_sync( /// Any failure (room not world-readable, federation disabled, server /// unreachable, malformed response) collapses to a clean "no preview" error so /// the client falls back to showing the join prompt instead of a hard error. -async fn remote_peek_preview(room_id: &RoomId, limit: usize) -> JsonResult { +async fn remote_peek_preview( + recipient: &UserId, + room_id: &RoomId, + limit: usize, +) -> JsonResult { let no_preview = || MatrixError::forbidden("No room preview available.", None); let server = room_id.server_name().map_err(|_| no_preview())?; @@ -278,7 +282,7 @@ async fn remote_peek_preview(room_id: &RoomId, limit: usize) -> JsonResult JsonResult>(); let events_after = timeline::stream::load_pdus_forward( Some(sender_id), @@ -285,7 +288,7 @@ pub(super) async fn get_context( .unwrap_or_else(|| base_token); let events_after: Vec<_> = events_after .into_iter() - .map(|(_, pdu)| pdu.to_room_event()) + .map(|(_, pdu)| pdu.to_room_event_for(sender_id)) .collect(); let mut state = Vec::new(); @@ -304,7 +307,7 @@ pub(super) async fn get_context( continue; } }; - state.push(pdu.to_state_event()); + state.push(pdu.to_state_event_for(sender_id)); } else if !lazy_load_enabled || lazy_loaded.contains(&state_key) { let pdu = match timeline::get_pdu(&event_id).await { Ok(pdu) => pdu, @@ -313,7 +316,7 @@ pub(super) async fn get_context( continue; } }; - state.push(pdu.to_state_event()); + state.push(pdu.to_state_event_for(sender_id)); } } diff --git a/crates/server/src/routing/client/room/message.rs b/crates/server/src/routing/client/room/message.rs index 1b2a390f7..65db02c6a 100644 --- a/crates/server/src/routing/client/room/message.rs +++ b/crates/server/src/routing/client/room/message.rs @@ -1,16 +1,17 @@ use std::collections::{BTreeMap, HashSet}; +use std::time::Duration; use diesel::prelude::*; use diesel_async::RunQueryDsl; use serde_json::value::to_raw_value; -use crate::core::Direction; +use crate::core::client::delayed_events::SendEventResBody; use crate::core::client::message::{ CreateMessageReqArgs, CreateMessageWithTxnReqArgs, MessagesReqArgs, MessagesResBody, - SendMessageResBody, }; use crate::core::events::{StateEventType, TimelineEventType}; -use crate::core::serde::{RawJsonValue, to_canonical_value}; +use crate::core::serde::to_canonical_value; +use crate::core::{Direction, UnixMillis}; use crate::data::schema::*; use crate::data::{connect, diesel_exists}; use crate::event::BatchToken; @@ -18,7 +19,7 @@ use crate::room::timeline::{self, topolo}; use crate::routing::prelude::*; use crate::{PduBuilder, room}; -fn parse_event_content(payload: &[u8]) -> AppResult> { +fn parse_event_content(payload: &[u8]) -> AppResult { let content: JsonValue = serde_json::from_slice(payload).map_err(|_| MatrixError::bad_json("invalid json body"))?; if !content.is_object() { @@ -27,7 +28,18 @@ fn parse_event_content(payload: &[u8]) -> AppResult> { to_canonical_value(&content).map_err(|e| { MatrixError::bad_json(format!("event content is not valid canonical JSON: {e}")) })?; - Ok(to_raw_value(&content).expect("validated JSON content can be serialized as raw JSON")) + Ok(content) +} + +fn appservice_timestamp( + is_appservice: bool, + requested_timestamp: Option, +) -> Option { + if is_appservice { + requested_timestamp + } else { + None + } } /// #GET /_matrix/client/r0/rooms/{room_id}/messages @@ -145,7 +157,7 @@ pub(super) async fn get_messages( let events: Vec<_> = events .into_iter() - .map(|(_, pdu)| pdu.to_room_event()) + .map(|(_, pdu)| pdu.to_room_event_for(sender_id)) .collect(); resp.start = from_tk.to_string(); @@ -200,7 +212,10 @@ pub(super) async fn get_messages( next_token = events.last().map(|(_, pdu)| pdu.prev_historic_token()); resp.start = from_tk.to_string(); resp.end = next_token.map(|tk| tk.to_string()); - resp.chunk = events.values().map(|pdu| pdu.to_room_event()).collect(); + resp.chunk = events + .values() + .map(|pdu| pdu.to_room_event_for(sender_id)) + .collect(); } } @@ -214,7 +229,7 @@ pub(super) async fn get_messages( ) .await { - resp.state.push(member_event.to_state_event()); + resp.state.push(member_event.to_state_event_for(sender_id)); } } @@ -244,7 +259,7 @@ pub(super) async fn send_message( args: CreateMessageWithTxnReqArgs, req: &mut Request, depot: &mut Depot, -) -> JsonResult { +) -> JsonResult { let authed = depot.authed_info()?; let conf = config::get(); @@ -258,6 +273,27 @@ pub(super) async fn send_message( let payload = req.payload().await?; let content = parse_event_content(payload)?; + if let Some(delay_ms) = args.delay { + if !conf.delayed_events.enable { + return Err(MatrixError::unrecognized("MSC4140 delayed events are disabled").into()); + } + let event_type: TimelineEventType = args.event_type.to_string().into(); + let delay_id = crate::delayed_event::schedule( + authed.user_id(), + Some(authed.device_id()), + authed.appservice().is_some(), + &args.room_id, + &event_type, + &args.txn_id, + args.timestamp, + Duration::from_millis(delay_ms), + None, + content, + ) + .await?; + return json_ok(SendEventResBody::delayed(delay_id)); + } + let state_lock = room::lock_state(&args.room_id).await; // Check if this is a new transaction id if let Some(event_id) = crate::transaction_id::get_event_id( @@ -268,7 +304,7 @@ pub(super) async fn send_message( ) .await? { - return json_ok(SendMessageResBody::new(event_id)); + return json_ok(SendEventResBody::sent(event_id)); } let mut unsigned = BTreeMap::new(); @@ -280,13 +316,9 @@ pub(super) async fn send_message( let event_id = timeline::build_and_append_pdu( PduBuilder { event_type: args.event_type.to_string().into(), - content, + content: to_raw_value(&content)?, unsigned, - timestamp: if authed.appservice().is_some() { - args.timestamp - } else { - None - }, + timestamp: appservice_timestamp(authed.appservice().is_some(), args.timestamp), ..Default::default() }, authed.user_id(), @@ -307,7 +339,7 @@ pub(super) async fn send_message( ) .await?; - json_ok(SendMessageResBody::new((*event_id).to_owned())) + json_ok(SendEventResBody::sent((*event_id).to_owned())) } /// #POST /_matrix/client/r0/rooms/{room_id}/send/{event_type} @@ -322,11 +354,10 @@ pub(super) async fn post_message( args: CreateMessageReqArgs, req: &mut Request, depot: &mut Depot, -) -> JsonResult { +) -> JsonResult { let authed = depot.authed_info()?; let conf = config::get(); - let state_lock = room::lock_state(&args.room_id).await; // Forbid m.room.encrypted if encryption is disabled if TimelineEventType::RoomEncrypted == args.event_type.to_string().into() && !conf.allow_encryption @@ -337,11 +368,35 @@ pub(super) async fn post_message( let payload = req.payload().await?; let content = parse_event_content(payload)?; + if let Some(delay_ms) = args.delay { + if !conf.delayed_events.enable { + return Err(MatrixError::unrecognized("MSC4140 delayed events are disabled").into()); + } + let txn_id: OwnedTransactionId = crate::utils::random_string(18).into(); + let event_type: TimelineEventType = args.event_type.to_string().into(); + let delay_id = crate::delayed_event::schedule( + authed.user_id(), + Some(authed.device_id()), + authed.appservice().is_some(), + &args.room_id, + &event_type, + &txn_id, + args.timestamp, + Duration::from_millis(delay_ms), + None, + content, + ) + .await?; + return json_ok(SendEventResBody::delayed(delay_id)); + } + + let state_lock = room::lock_state(&args.room_id).await; let event_id = timeline::build_and_append_pdu( PduBuilder { event_type: args.event_type.to_string().into(), - content, + content: to_raw_value(&content)?, unsigned: BTreeMap::new(), + timestamp: appservice_timestamp(authed.appservice().is_some(), args.timestamp), ..Default::default() }, authed.user_id(), @@ -353,5 +408,19 @@ pub(super) async fn post_message( .pdu .event_id; - json_ok(SendMessageResBody::new((*event_id).to_owned())) + json_ok(SendEventResBody::sent((*event_id).to_owned())) +} + +#[cfg(test)] +mod tests { + use super::appservice_timestamp; + use crate::core::UnixMillis; + + #[test] + fn timestamp_massaging_is_limited_to_appservices() { + let timestamp = UnixMillis(123_456); + + assert_eq!(appservice_timestamp(true, Some(timestamp)), Some(timestamp)); + assert_eq!(appservice_timestamp(false, Some(timestamp)), None); + } } diff --git a/crates/server/src/routing/client/room/state.rs b/crates/server/src/routing/client/room/state.rs index 99a0c6a26..ba951a2d8 100644 --- a/crates/server/src/routing/client/room/state.rs +++ b/crates/server/src/routing/client/room/state.rs @@ -1,17 +1,20 @@ +use std::time::Duration; + use salvo::oapi::extract::*; use salvo::prelude::*; use serde_json::json; use crate::core::UnixMillis; +use crate::core::client::delayed_events::SendEventResBody; use crate::core::client::room::ReportContentReqBody; use crate::core::client::state::{ - SendStateEventReqBody, SendStateEventResBody, StateEventFormat, StateEventsForEmptyKeyReqArgs, + SendStateEventReqArgs, SendStateEventReqBody, StateEventFormat, StateEventsForEmptyKeyReqArgs, StateEventsForKeyReqArgs, StateEventsForKeyResBody, StateEventsResBody, }; use crate::core::client::typing::{CreateTypingEventReqBody, Typing}; use crate::core::events::room::message::RoomMessageEventContent; use crate::core::identifiers::*; -use crate::core::room::{RoomEventReqArgs, RoomEventTypeReqArgs, RoomTypingReqArgs}; +use crate::core::room::{RoomEventReqArgs, RoomTypingReqArgs}; use crate::room::{state, timeline}; use crate::utils::HtmlEscape; use crate::{AuthArgs, DepotExt, EmptyResult, JsonResult, MatrixError, empty_ok, json_ok, room}; @@ -50,7 +53,7 @@ pub(super) async fn get_state( .await .unwrap_or_default() .values() - .map(|pdu| pdu.to_state_event()) + .map(|pdu| pdu.to_state_event_for(sender_id)) .collect(); json_ok(StateEventsResBody::new(room_state)) } @@ -166,7 +169,7 @@ pub(super) async fn state_for_key( json_ok(StateEventsForKeyResBody { content: Some(event.get_content()?), event: if event_format { - Some(event.to_state_event_value()) + Some(event.to_state_event_value_for(sender_id)) } else { None }, @@ -207,7 +210,7 @@ pub(super) async fn state_for_empty_key( json_ok(StateEventsForKeyResBody { content: Some(event.get_content()?), event: if event_format { - Some(event.to_state_event_value()) + Some(event.to_state_event_value_for(sender_id)) } else { None }, @@ -223,12 +226,36 @@ pub(super) async fn state_for_empty_key( #[endpoint] pub(super) async fn send_state_for_key( _aa: AuthArgs, - args: StateEventsForKeyReqArgs, + args: SendStateEventReqArgs, body: JsonBody, depot: &mut Depot, -) -> JsonResult { +) -> JsonResult { let authed = depot.authed_info()?; let body = body.into_inner(); + let state_key = args.state_key.clone().unwrap_or_default(); + + if let Some(delay_ms) = args.delay { + if !crate::config::get().delayed_events.enable { + return Err(MatrixError::unrecognized("MSC4140 delayed events are disabled").into()); + } + let txn_id: OwnedTransactionId = crate::utils::random_string(18).into(); + let event_type = args.event_type.to_string().into(); + let content = serde_json::from_str(body.0.as_str())?; + let delay_id = crate::delayed_event::schedule( + authed.user_id(), + Some(authed.device_id()), + authed.appservice().is_some(), + &args.room_id, + &event_type, + &txn_id, + args.timestamp, + Duration::from_millis(delay_ms), + Some(state_key), + content, + ) + .await?; + return json_ok(SendEventResBody::delayed(delay_id)); + } let event_id = crate::state::send_state_event_for_key( authed.user_id(), @@ -236,13 +263,12 @@ pub(super) async fn send_state_for_key( &crate::room::get_version(&args.room_id).await?, &args.event_type, body.0, - args.state_key.to_owned(), + state_key, + appservice_timestamp(authed.appservice().is_some(), args.timestamp), ) .await?; - json_ok(SendStateEventResBody { - event_id: (*event_id).to_owned(), - }) + json_ok(SendEventResBody::sent((*event_id).to_owned())) } /// #PUT /_matrix/client/r0/rooms/{room_id}/state/{event_type} @@ -254,12 +280,34 @@ pub(super) async fn send_state_for_key( #[endpoint] pub(super) async fn send_state_for_empty_key( _aa: AuthArgs, - args: RoomEventTypeReqArgs, + args: SendStateEventReqArgs, body: JsonBody, depot: &mut Depot, -) -> JsonResult { +) -> JsonResult { let authed = depot.authed_info()?; let body = body.into_inner(); + if let Some(delay_ms) = args.delay { + if !crate::config::get().delayed_events.enable { + return Err(MatrixError::unrecognized("MSC4140 delayed events are disabled").into()); + } + let txn_id: OwnedTransactionId = crate::utils::random_string(18).into(); + let event_type = args.event_type.to_string().into(); + let content = serde_json::from_str(body.0.as_str())?; + let delay_id = crate::delayed_event::schedule( + authed.user_id(), + Some(authed.device_id()), + authed.appservice().is_some(), + &args.room_id, + &event_type, + &txn_id, + args.timestamp, + Duration::from_millis(delay_ms), + Some(String::new()), + content, + ) + .await?; + return json_ok(SendEventResBody::delayed(delay_id)); + } let event_id = crate::state::send_state_event_for_key( authed.user_id(), &args.room_id, @@ -267,12 +315,36 @@ pub(super) async fn send_state_for_empty_key( &args.event_type.to_string().into(), body.0, "".into(), + appservice_timestamp(authed.appservice().is_some(), args.timestamp), ) .await?; - json_ok(SendStateEventResBody { - event_id: (*event_id).to_owned(), - }) + json_ok(SendEventResBody::sent((*event_id).to_owned())) +} + +fn appservice_timestamp( + is_appservice: bool, + requested_timestamp: Option, +) -> Option { + if is_appservice { + requested_timestamp + } else { + None + } +} + +#[cfg(test)] +mod tests { + use super::appservice_timestamp; + use crate::core::UnixMillis; + + #[test] + fn timestamp_massaging_is_limited_to_appservices() { + let timestamp = UnixMillis(123_456); + + assert_eq!(appservice_timestamp(true, Some(timestamp)), Some(timestamp)); + assert_eq!(appservice_timestamp(false, Some(timestamp)), None); + } } /// #PUT /_matrix/client/r0/rooms/{room_id}/typing/{user_id} diff --git a/crates/server/src/routing/client/room/thread.rs b/crates/server/src/routing/client/room/thread.rs index 8520f5be5..f7276ac83 100644 --- a/crates/server/src/routing/client/room/thread.rs +++ b/crates/server/src/routing/client/room/thread.rs @@ -42,7 +42,7 @@ pub(super) async fn list_threads( json_ok(ThreadsResBody { chunk: threads .into_iter() - .map(|(_, pdu)| pdu.to_room_event()) + .map(|(_, pdu)| pdu.to_room_event_for(authed.user_id())) .collect(), next_batch: next_batch.map(|b| b.to_string()), }) diff --git a/crates/server/src/routing/client/unstable.rs b/crates/server/src/routing/client/unstable.rs index 7d54fff08..e46c356f7 100644 --- a/crates/server/src/routing/client/unstable.rs +++ b/crates/server/src/routing/client/unstable.rs @@ -4,7 +4,13 @@ use crate::core::MatrixError; use crate::core::client::discovery::rendezvous::DiscoverRendezvousResBody; use crate::{JsonResult, config, hoops, json_ok}; -pub(super) fn router() -> Router { +pub(super) fn router(delayed_events: bool) -> Router { + let mut authed = Router::new() + .hoop(hoops::limit_rate) + .hoop(hoops::auth_by_access_token); + if delayed_events { + authed = authed.push(super::delayed_event::authed_router()); + } Router::with_path("unstable") // Public routes (no auth required) — MSC2965 OIDC discovery .push( @@ -17,9 +23,7 @@ pub(super) fn router() -> Router { .push(super::profile::msc4133_public_router()) // Authed routes .push( - Router::new() - .hoop(hoops::limit_rate) - .hoop(hoops::auth_by_access_token) + authed .push( Router::with_path( "org.matrix.msc3391/user/{user_id}/account_data/{account_type}", diff --git a/crates/server/src/sending.rs b/crates/server/src/sending.rs index b21b9d43f..32514e46d 100644 --- a/crates/server/src/sending.rs +++ b/crates/server/src/sending.rs @@ -440,7 +440,7 @@ async fn send_events( timeline::get_pdu(event_id) .await .map_err(|e| (kind.clone(), e))? - .to_room_event(), + .to_room_event_without_sender_only_unsigned(), ); } SendingEventType::Edu(_) => { @@ -1016,9 +1016,10 @@ async fn claim_queued_requests( /// This does not return a full `Pdu` it is only to satisfy palpo's types. /// -/// Strips internal fields (`event_sn`, `transaction_id`) and conditionally removes -/// `event_id` based on the room version. Room versions V1/V2 require `event_id` in -/// the federation format; V3+ derive it from the event hash. +/// Strips internal and sender-only fields (`event_sn`, `transaction_id`, MSC4140's +/// `delay_id`) and conditionally removes `event_id` based on the room version. Room +/// versions V1/V2 require `event_id` in the federation format; V3+ derive it from the +/// event hash. #[tracing::instrument] pub async fn convert_to_outgoing_federation_event( mut pdu_json: CanonicalJsonObject, @@ -1028,6 +1029,9 @@ pub async fn convert_to_outgoing_federation_event( .and_then(|val| val.as_object_mut()) { unsigned.remove("transaction_id"); + // MSC4140: the delay_id is only ever shown to the event's own sender, so it must + // never cross a federation boundary. + unsigned.remove("org.matrix.msc4140.delay_id"); } // Determine room version to decide whether to strip event_id. diff --git a/crates/server/src/state.rs b/crates/server/src/state.rs index 01fc8204e..e74ed6cfd 100644 --- a/crates/server/src/state.rs +++ b/crates/server/src/state.rs @@ -1,3 +1,4 @@ +use crate::core::UnixMillis; use crate::core::events::room::canonical_alias::RoomCanonicalAliasEventContent; use crate::core::events::room::history_visibility::{ HistoryVisibility, RoomHistoryVisibilityEventContent, @@ -20,6 +21,7 @@ pub async fn send_state_event_for_key( event_type: &StateEventType, json: RawJson, state_key: String, + timestamp: Option, ) -> AppResult { allowed_to_send_state_event(room_id, event_type, &state_key, &json).await?; let pdu = timeline::build_and_append_pdu( @@ -27,6 +29,7 @@ pub async fn send_state_event_for_key( event_type: event_type.to_string().into(), content: serde_json::from_value(serde_json::to_value(json)?)?, state_key: Some(state_key), + timestamp, ..Default::default() }, user_id, @@ -40,7 +43,7 @@ pub async fn send_state_event_for_key( Ok(pdu.event_id) } -async fn allowed_to_send_state_event( +pub(crate) async fn allowed_to_send_state_event( room_id: &RoomId, event_type: &StateEventType, state_key: &str, diff --git a/crates/server/src/sync_v3.rs b/crates/server/src/sync_v3.rs index 3637d4aba..6b6bb9cb0 100644 --- a/crates/server/src/sync_v3.rs +++ b/crates/server/src/sync_v3.rs @@ -918,13 +918,13 @@ async fn load_joined_room( events: timeline .events .iter() - .map(|(_, pdu)| pdu.to_sync_room_event()) + .map(|(_, pdu)| pdu.to_sync_room_event_for(sender_id)) .collect(), }, state: State::Before( state_events .iter() - .map(|pdu| pdu.to_sync_state_event()) + .map(|pdu| pdu.to_sync_state_event_for(sender_id)) .collect::>() .into(), ), @@ -996,7 +996,7 @@ async fn load_left_room( prev_batch: Some(next_batch.to_string()), events: Vec::new(), }, - state: State::Before(vec![event.to_sync_state_event()].into()), + state: State::Before(vec![event.to_sync_state_event_for(sender_id)].into()), }); } @@ -1093,13 +1093,13 @@ async fn load_left_room( events: timeline .events .iter() - .map(|(_, pdu)| pdu.to_sync_room_event()) + .map(|(_, pdu)| pdu.to_sync_room_event_for(sender_id)) .collect(), }, state: State::Before( state_events .iter() - .map(|pdu| pdu.to_sync_state_event()) + .map(|pdu| pdu.to_sync_state_event_for(sender_id)) .collect::>() .into(), ), diff --git a/crates/server/src/sync_v5.rs b/crates/server/src/sync_v5.rs index e1ac085c3..8c3ba76bc 100644 --- a/crates/server/src/sync_v5.rs +++ b/crates/server/src/sync_v5.rs @@ -758,7 +758,7 @@ async fn process_rooms( .events .iter() .filter(|item| ignored_filter_with_ignored_users(*item, ignored_users)) - .map(|(_, pdu)| pdu.to_sync_room_event()) + .map(|(_, pdu)| pdu.to_sync_room_event_for(sender_id)) .collect(); for (_, pdu) in &timeline.events { @@ -811,7 +811,7 @@ async fn process_rooms( pdu.event_sn, ) .await; - required_state_events.push(pdu.to_sync_state_event()); + required_state_events.push(pdu.to_sync_state_event_for(sender_id)); } } } @@ -837,7 +837,7 @@ async fn process_rooms( pdu.event_sn, ) .await; - required_state_events.push(pdu.to_sync_state_event()); + required_state_events.push(pdu.to_sync_state_event_for(sender_id)); } } } @@ -861,7 +861,7 @@ async fn process_rooms( pdu.event_sn, ) .await; - required_state_events.push(pdu.to_sync_state_event()); + required_state_events.push(pdu.to_sync_state_event_for(sender_id)); } } // Specific state key @@ -882,7 +882,7 @@ async fn process_rooms( pdu.event_sn, ) .await; - required_state_events.push(pdu.to_sync_state_event()); + required_state_events.push(pdu.to_sync_state_event_for(sender_id)); } } } diff --git a/palpo-example.toml b/palpo-example.toml index 9829402d5..ea711604d 100644 --- a/palpo-example.toml +++ b/palpo-example.toml @@ -798,6 +798,26 @@ # # enforce_tls = +# [delayed_events] + +# Allow scheduling MSC4140 delayed events. Disabled by default while the MSC +# is unstable. When disabled, the endpoints are not registered and the feature +# is not advertised. +# +# enable = false + +# Maximum requested delay in milliseconds. Defaults to 24 hours. +# +# max_delay_ms = 86400_000 + +# Maximum number of delayed events a user may have scheduled at once. +# +# max_scheduled = 100 + +# Retention period for finalized delayed events, in milliseconds. +# +# retention_ms = 604800_000 + # [federation] # Controls whether federation is allowed or not. It is not recommended to