events: Add CallMemberEvent

Signed-off-by: Timo K <toger5@hotmail.de>

crates/ruma-events/Cargo.toml

@@ -30,6 +30,7 @@ unstable-msc3245 = ["unstable-msc3246"]
 unstable-msc3245-v1-compat = []
 unstable-msc3246 = ["unstable-msc3927"]
 unstable-msc3381 = ["unstable-msc1767"]
+unstable-msc3401 = []
 unstable-msc3488 = ["unstable-msc1767"]
 unstable-msc3551 = ["unstable-msc3956"]
 unstable-msc3552 = ["unstable-msc3551"]

crates/ruma-events/src/call.rs

@@ -6,6 +6,8 @@ pub mod answer;
 pub mod candidates;
 pub mod hangup;
 pub mod invite;
+#[cfg(feature = "unstable-msc3401")]
+pub mod member;
 pub mod negotiate;
 pub mod reject;
 pub mod select_answer;

crates/ruma-events/src/call/member.rs

@@ -0,0 +1,498 @@
+//! Types for matrixRTC state events ([MSC3401]).
+//!
+//! This implements a newer/updated version of MSC3401.
+//!
+//! [MSC3401]: https://github.com/matrix-org/matrix-spec-proposals/pull/3401
+
+use std::time::Duration;
+
+use as_variant::as_variant;
+use ruma_common::{serde::StringEnum, MilliSecondsSinceUnixEpoch, OwnedUserId};
+use ruma_macros::EventContent;
+use serde::{Deserialize, Serialize};
+use tracing::warn;
+
+use crate::PrivOwnedStr;
+
+/// The member state event for a matrixRTC session.
+///
+/// This is the object containing all the data related to a matrix users participation in a
+/// matrixRTC session. It consists of memberships / sessions.
+#[derive(Clone, Debug, Serialize, Deserialize, EventContent, PartialEq)]
+#[cfg_attr(not(feature = "unstable-exhaustive-types"), non_exhaustive)]
+#[ruma_event(type = "org.matrix.msc3401.call.member", kind = State, state_key_type = OwnedUserId)]
+pub struct CallMemberEventContent {
+    /// A list of all the memberships that user currently has in this room.
+    ///
+    /// There can be multiple ones in cases the user participates with multiple devices or there
+    /// are multiple RTC applications running.
+    ///
+    /// e.g. a call and a spacial experience.
+    ///
+    /// Important: This includes expired memberships.
+    /// To retrieve a list including only valid memberships,
+    /// see [`active_memberships`](CallMemberEventContent::active_memberships).
+    pub memberships: Vec<Membership>,
+}
+
+impl CallMemberEventContent {
+    /// Creates a new `CallMemberEventContent`.
+    pub fn new(memberships: Vec<Membership>) -> Self {
+        Self { memberships }
+    }
+
+    /// All non expired memberships in this member event.
+    ///
+    /// In most cases you want tu use this method instead of the public memberships field.
+    /// The memberships field will also include expired events.
+    ///
+    /// # Arguments
+    ///
+    /// * `origin_server_ts` - optionally the `origin_server_ts` can be passed as a fallback in case
+    ///   the Membership does not contain `created_ts`. (`origin_server_ts` will be ignored if
+    ///   `created_ts` is `Some`)
+    pub fn active_memberships(
+        &self,
+        origin_server_ts: Option<MilliSecondsSinceUnixEpoch>,
+    ) -> Vec<&Membership> {
+        self.memberships.iter().filter(|m| !m.is_expired(origin_server_ts)).collect()
+    }
+
+    /// Set the `created_ts` of each [Membership] in this event.
+    ///
+    /// Each call member event contains the `origin_server_ts` and `content.create_ts`.
+    /// `content.create_ts` is undefined for the initial event of a session (because the
+    /// `origin_server_ts` is not known on the client).
+    /// In the rust sdk we want to copy over the `origin_server_ts` of the event into the content.
+    /// (This allows to use `MinimalStateEvents` and still be able to determine if a membership is
+    /// expired)
+    pub fn set_created_ts_if_none(&mut self, origin_server_ts: MilliSecondsSinceUnixEpoch) {
+        self.memberships.iter_mut().for_each(|m| {
+            m.created_ts.get_or_insert(origin_server_ts);
+        });
+    }
+}
+
+/// A membership describes one of the sessions this user currently partakes.
+///
+/// The application defines the type of the session.
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+#[cfg_attr(not(feature = "unstable-exhaustive-types"), non_exhaustive)]
+pub struct Membership {
+    /// The type of the matrixRTC session the membership belongs to.
+    ///
+    /// e.g. call, spacial, document...
+    #[serde(flatten)]
+    pub application: Application,
+
+    /// The device id of this membership.
+    ///
+    /// The same user can join with their phone/computer.
+    pub device_id: String,
+
+    /// The duration in milliseconds relative to the time this membership joined
+    /// during which the membership is valid.
+    ///
+    /// The time a member has joined is defined as:
+    /// `MIN(content.created_ts, event.origin_server_ts)`
+    #[serde(with = "ruma_common::serde::duration::ms")]
+    pub expires: Duration,
+
+    /// Stores a copy of the `origin_server_ts` of the initial session event.
+    ///
+    /// If the membership is updated this field will be used to track to
+    /// original `origin_server_ts`.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub created_ts: Option<MilliSecondsSinceUnixEpoch>,
+
+    /// A list of the foci in use for this membership.
+    pub foci_active: Vec<Focus>,
+
+    /// The id of the membership.
+    ///
+    /// This is required to guarantee uniqueness of the event.
+    /// Sending the same state event twice to synapse makes the HS drop the second one and return
+    /// 200.
+    #[serde(rename = "membershipID")]
+    pub membership_id: String,
+}
+
+impl Membership {
+    /// The application of the membership is "m.call" and the scope is "m.room".
+    pub fn is_room_call(&self) -> bool {
+        as_variant!(&self.application, Application::Call)
+            .is_some_and(|call| call.scope == CallScope::Room)
+    }
+
+    /// The application of the membership is "m.call".
+    pub fn is_call(&self) -> bool {
+        as_variant!(&self.application, Application::Call).is_some()
+    }
+
+    /// Checks if the event is expired.
+    ///
+    /// Defaults to using `created_ts` of the `Membership`.
+    /// If no `origin_server_ts` is provided and the event does not contain `created_ts`
+    /// the event will be considered as not expired.
+    /// In this case, a warning will be logged.
+    ///
+    /// # Arguments
+    ///
+    /// * `origin_server_ts` - a fallback if `created_ts` is not present
+    pub fn is_expired(&self, origin_server_ts: Option<MilliSecondsSinceUnixEpoch>) -> bool {
+        let ev_created_ts = self.created_ts.or(origin_server_ts);
+
+        if let Some(ev_created_ts) = ev_created_ts {
+            let now = MilliSecondsSinceUnixEpoch::now().to_system_time();
+            let expire_ts = ev_created_ts.to_system_time().map(|t| t + self.expires);
+            now > expire_ts
+        } else {
+            // This should not be reached since we only allow events that have copied over
+            // the origin server ts. `set_created_ts_if_none`
+            warn!("Encountered a Call Member state event where the origin_ts (or origin_server_ts) could not be found.\
+            It is treated as a non expired event but this might be wrong.");
+            false
+        }
+    }
+}
+
+/// Initial set of fields of [`Membership`].
+#[cfg_attr(not(feature = "unstable-exhaustive-types"), non_exhaustive)]
+pub struct MembershipInit {
+    /// The type of the matrixRTC session the membership belongs to.
+    ///
+    /// e.g. call, spacial, document...
+    pub application: Application,
+
+    /// The device id of this membership.
+    ///
+    /// The same user can join with their phone/computer.
+    pub device_id: String,
+
+    /// The duration in milliseconds relative to the time this membership joined
+    /// during which the membership is valid.
+    ///
+    /// The time a member has joined is defined as:
+    /// `MIN(content.created_ts, event.origin_server_ts)`
+    pub expires: Duration,
+
+    /// Stores a copy of the `origin_server_ts` of the initial session event.
+    ///
+    /// If the membership is updated this field will be used to track to
+    /// original `origin_server_ts`.
+    pub created_ts: Option<MilliSecondsSinceUnixEpoch>,
+
+    /// A list of the focuses (foci) in use for this membership.
+    pub foci_active: Vec<Focus>,
+
+    /// The id of the membership.
+    ///
+    /// This is required to guarantee uniqueness of the event.
+    /// Sending the same state event twice to synapse makes the HS drop the second one and return
+    /// 200.
+    pub membership_id: String,
+}
+
+impl From<MembershipInit> for Membership {
+    fn from(init: MembershipInit) -> Self {
+        let MembershipInit {
+            application,
+            device_id,
+            expires,
+            created_ts,
+            foci_active,
+            membership_id,
+        } = init;
+        Self { application, device_id, expires, created_ts, foci_active, membership_id }
+    }
+}
+
+/// Description of the SFU/Focus a membership can be connected to.
+///
+/// A focus can be any server powering the matrixRTC session (SFU,
+/// MCU). It serves as a node to redistribute RTC streams.
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+#[serde(tag = "type")]
+#[serde(rename_all = "snake_case")]
+#[cfg_attr(not(feature = "unstable-exhaustive-types"), non_exhaustive)]
+pub enum Focus {
+    /// Livekit is one possible type of SFU/Focus that can be used for a matrixRTC session.
+    Livekit(LivekitFocus),
+}
+
+/// The fields to describe livekit as an `active_foci`.
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+#[cfg_attr(not(feature = "unstable-exhaustive-types"), non_exhaustive)]
+pub struct LivekitFocus {
+    /// The alias where the livekit sessions can be reached.
+    #[serde(rename = "livekit_alias")]
+    pub alias: String,
+
+    /// The url of the jwt server for the livekit instance.
+    #[serde(rename = "livekit_service_url")]
+    pub service_url: String,
+}
+
+impl LivekitFocus {
+    /// Initialize a [`LivekitFocus`].
+    ///
+    /// # Arguments
+    ///
+    /// * `alias` - The alias where the livekit sessions can be reached.
+    /// * `service_url` - The url of the jwt server for the livekit instance.
+    pub fn new(alias: String, service_url: String) -> Self {
+        Self { alias, service_url }
+    }
+}
+
+/// The type of the matrixRTC session.
+///
+/// This is not the application/client used by the user but the
+/// type of matrixRTC session e.g. calling (`m.call`), third-room, whiteboard could be
+/// possible applications.
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+#[serde(tag = "application")]
+#[cfg_attr(not(feature = "unstable-exhaustive-types"), non_exhaustive)]
+pub enum Application {
+    #[serde(rename = "m.call")]
+    /// A VoIP call.
+    Call(CallApplicationContent),
+}
+
+/// Call specific parameters membership parameters.
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+#[cfg_attr(not(feature = "unstable-exhaustive-types"), non_exhaustive)]
+pub struct CallApplicationContent {
+    /// An identifier for calls.
+    ///
+    /// All members using the same `call_id` will end up in the same call.
+    ///
+    /// Does not need to be a uuid.
+    ///
+    /// `""` is used for room scoped calls.
+    pub call_id: String,
+
+    /// Who owns/joins/controls (can modify) the call.
+    pub scope: CallScope,
+}
+
+impl CallApplicationContent {
+    /// Initialize a [`CallApplicationContent`].
+    ///
+    /// # Arguments
+    ///
+    /// * `call_id` - An identifier for calls. All members using the same `call_id` will end up in
+    ///   the same call. Does not need to be a uuid. `""` is used for room scoped calls.
+    /// * `scope` - Who owns/joins/controls (can modify) the call.
+    pub fn new(call_id: String, scope: CallScope) -> Self {
+        Self { call_id, scope }
+    }
+}
+
+/// The call scope defines different call ownership models.
+#[doc = include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/src/doc/string_enum.md"))]
+#[derive(Clone, PartialEq, StringEnum)]
+#[cfg_attr(not(feature = "unstable-exhaustive-types"), non_exhaustive)]
+#[ruma_enum(rename_all = "m.snake_case")]
+pub enum CallScope {
+    /// A call which every user of a room can join and create.
+    ///
+    /// There is no particular name associated with it.
+    ///
+    /// There can only be one per room.
+    Room,
+    /// A user call is owned by a user.
+    ///
+    /// Each user can create one there can be multiple per room. They are started and ended by the
+    /// owning user.
+    User,
+
+    #[doc(hidden)]
+    _Custom(PrivOwnedStr),
+}
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use ruma_common::MilliSecondsSinceUnixEpoch as TS;
+    use serde_json::json;
+
+    use super::{
+        Application, CallApplicationContent, CallMemberEventContent, CallScope, Focus,
+        LivekitFocus, Membership,
+    };
+
+    fn create_call_member_event_content() -> CallMemberEventContent {
+        CallMemberEventContent::new(vec![Membership {
+            application: Application::Call(CallApplicationContent {
+                call_id: "123456".to_owned(),
+                scope: CallScope::Room,
+            }),
+            device_id: "ABCDE".to_owned(),
+            expires: Duration::from_secs(3600),
+            foci_active: vec![Focus::Livekit(LivekitFocus {
+                alias: "1".to_owned(),
+                service_url: "https://livekit.com".to_owned(),
+            })],
+            membership_id: "0".to_owned(),
+            created_ts: None,
+        }])
+    }
+
+    #[test]
+    fn serialize_call_member_event_content() {
+        let call_member_event = &json!({
+            "memberships": [
+                {
+                    "application": "m.call",
+                    "call_id": "123456",
+                    "scope": "m.room",
+                    "device_id": "ABCDE",
+                    "expires": 3_600_000,
+                    "foci_active": [
+                        {
+                            "livekit_alias": "1",
+                            "livekit_service_url": "https://livekit.com",
+                            "type": "livekit"
+                        }
+                    ],
+                    "membershipID": "0"
+                }
+            ]
+        });
+
+        assert_eq!(
+            call_member_event,
+            &serde_json::to_value(create_call_member_event_content()).unwrap()
+        );
+    }
+
+    #[test]
+    fn deserialize_call_member_event_content() {
+        let call_member_ev: CallMemberEventContent = CallMemberEventContent::new(vec![
+            Membership {
+                application: Application::Call(CallApplicationContent {
+                    call_id: "123456".to_owned(),
+                    scope: CallScope::Room,
+                }),
+                device_id: "THIS_DEVICE".to_owned(),
+                expires: Duration::from_secs(3600),
+                foci_active: vec![Focus::Livekit(LivekitFocus {
+                    alias: "room1".to_owned(),
+                    service_url: "https://livekit1.com".to_owned(),
+                })],
+                membership_id: "0".to_owned(),
+                created_ts: None,
+            },
+            Membership {
+                application: Application::Call(CallApplicationContent {
+                    call_id: "".to_owned(),
+                    scope: CallScope::Room,
+                }),
+                device_id: "OTHER_DEVICE".to_owned(),
+                expires: Duration::from_secs(3600),
+                foci_active: vec![Focus::Livekit(LivekitFocus {
+                    alias: "room2".to_owned(),
+                    service_url: "https://livekit2.com".to_owned(),
+                })],
+                membership_id: "0".to_owned(),
+                created_ts: None,
+            },
+        ]);
+
+        let call_member_ev_json = json!({
+            "memberships": [
+                {
+                    "application": "m.call",
+                    "call_id": "123456",
+                    "scope": "m.room",
+                    "device_id": "THIS_DEVICE",
+                    "expires": 3_600_000,
+                    "foci_active": [
+                        {
+                            "livekit_alias": "room1",
+                            "livekit_service_url": "https://livekit1.com",
+                            "type": "livekit"
+                        }
+                    ],
+                    "membershipID": "0",
+                },
+                {
+                    "application": "m.call",
+                    "call_id": "",
+                    "scope": "m.room",
+                    "device_id": "OTHER_DEVICE",
+                    "expires": 3_600_000,
+                    "foci_active": [
+                        {
+                            "livekit_alias": "room2",
+                            "livekit_service_url": "https://livekit2.com",
+                            "type": "livekit"
+                        }
+                    ],
+                    "membershipID": "0"
+                }
+            ]
+        });
+
+        let ev_content: CallMemberEventContent =
+            serde_json::from_value(call_member_ev_json).unwrap();
+        assert_eq!(ev_content, call_member_ev);
+    }
+
+    fn timestamps() -> (TS, TS, TS) {
+        let now = TS::now();
+        let one_second_ago =
+            now.to_system_time().unwrap().checked_sub(Duration::from_secs(1)).unwrap();
+        let two_hours_ago =
+            now.to_system_time().unwrap().checked_sub(Duration::from_secs(60 * 60 * 2)).unwrap();
+        (
+            now,
+            TS::from_system_time(one_second_ago).unwrap(),
+            TS::from_system_time(two_hours_ago).unwrap(),
+        )
+    }
+
+    #[test]
+    fn membership_do_expire() {
+        let content = create_call_member_event_content();
+        let (now, one_second_ago, two_hours_ago) = timestamps();
+        assert_eq!(
+            content.active_memberships(Some(one_second_ago)),
+            content.memberships.iter().collect::<Vec<&Membership>>()
+        );
+        assert_eq!(
+            content.active_memberships(Some(now)),
+            content.memberships.iter().collect::<Vec<&Membership>>()
+        );
+        assert_eq!(content.active_memberships(Some(two_hours_ago)), vec![] as Vec<&Membership>);
+    }
+
+    #[test]
+    fn set_created_ts() {
+        let mut content_now = create_call_member_event_content();
+        let mut content_two_hours_ago = create_call_member_event_content();
+        let mut content_one_second_ago = create_call_member_event_content();
+        let (now, one_second_ago, two_hours_ago) = timestamps();
+
+        content_now.set_created_ts_if_none(now);
+        content_one_second_ago.set_created_ts_if_none(one_second_ago);
+        content_two_hours_ago.set_created_ts_if_none(two_hours_ago);
+        assert_eq!(
+            content_now.active_memberships(None),
+            content_now.memberships.iter().collect::<Vec<&Membership>>()
+        );
+
+        assert_eq!(content_two_hours_ago.active_memberships(None), vec![] as Vec<&Membership>);
+        assert_eq!(
+            content_one_second_ago.active_memberships(None),
+            content_one_second_ago.memberships.iter().collect::<Vec<&Membership>>()
+        );
+
+        // created_ts should not be overwritten.
+        content_two_hours_ago.set_created_ts_if_none(one_second_ago);
+        // There still should be no active membership.
+        assert_eq!(content_two_hours_ago.active_memberships(None), vec![] as Vec<&Membership>);
+    }
+}

crates/ruma-events/src/enums.rs

@@ -118,6 +118,9 @@ event_enum! {
         "m.room.topic" => super::room::topic,
         "m.space.child" => super::space::child,
         "m.space.parent" => super::space::parent,
+        #[cfg(feature = "unstable-msc3401")]
+        #[ruma_enum(alias = "m.call.member")]
+        "org.matrix.msc3401.call.member" => super::call::member,
     }
 
     /// Any to-device event.

crates/ruma/Cargo.toml

@@ -195,6 +195,7 @@ unstable-msc3245 = ["ruma-events?/unstable-msc3245"]
 unstable-msc3245-v1-compat = ["ruma-events?/unstable-msc3245-v1-compat"]
 unstable-msc3246 = ["ruma-events?/unstable-msc3246"]
 unstable-msc3381 = ["ruma-events?/unstable-msc3381"]
+unstable-msc3401 = ["ruma-events?/unstable-msc3401"]
 unstable-msc3488 = ["ruma-client-api?/unstable-msc3488", "ruma-events?/unstable-msc3488"]
 unstable-msc3551 = ["ruma-events?/unstable-msc3551"]
 unstable-msc3552 = ["ruma-events?/unstable-msc3552"]
@@ -239,6 +240,7 @@ __ci = [
     "unstable-msc3245-v1-compat",
     "unstable-msc3246",
     "unstable-msc3381",
+    "unstable-msc3401",
     "unstable-msc3488",
     "unstable-msc3551",
     "unstable-msc3552",