arminius_lib_model/
syncv4.rs

1// SPDX-FileCopyrightText: 2025 BWI GmbH
2//
3// SPDX-License-Identifier: Apache-2.0
4
5//! Models related to sync v4 (Simplified Sliding Sync).
6
7mod tests;
8
9use std::collections::{HashMap, HashSet};
10
11use ruma::{api::client::sync::sync_events, events::StateEventType};
12
13use crate::types::StreamId;
14
15/// Configuration for a single room. See
16/// <https://github.com/matrix-org/matrix-spec-proposals/blob/erikj/sss/proposals/4186-simplified-sliding-sync.md#combining-room-configs>
17#[derive(Debug, Clone, Default)]
18pub struct RoomConfig {
19	/// The maximum number of timeline events to return per response.
20	pub timeline_limit: i64,
21	/// The request's configured required states.
22	pub required_state: Vec<(StateEventType, String)>,
23}
24
25/// Conversion from ruma's type used in lists.
26impl From<sync_events::v5::request::RoomDetails> for RoomConfig {
27	fn from(room_config: sync_events::v5::request::RoomDetails) -> Self {
28		Self {
29			timeline_limit: room_config.timeline_limit.into(),
30			required_state: room_config.required_state,
31		}
32	}
33}
34
35/// Conversion from ruma's type used in subscriptions.
36impl From<sync_events::v5::request::RoomSubscription> for RoomConfig {
37	fn from(room_config: sync_events::v5::request::RoomSubscription) -> Self {
38		Self {
39			timeline_limit: room_config.timeline_limit.into(),
40			required_state: room_config.required_state,
41		}
42	}
43}
44
45/// Sync v4 room configuration
46impl RoomConfig {
47	/// Function to perform room config combination (See
48	/// <https://github.com/matrix-org/matrix-spec-proposals/blob/main/proposals/4186-simplified-sliding-sync.md#combining-room-configs>
49	/// and
50	/// <https://github.com/matrix-org/matrix-spec-proposals/blob/main/proposals/4186-simplified-sliding-sync.md#combining-required_state>).
51	///
52	/// Returns the combined room config.
53	pub fn combine(&mut self, other: &Self) {
54		self.timeline_limit = self.timeline_limit.max(other.timeline_limit);
55
56		let mut other_required_state = other.required_state.clone();
57		self.required_state.append(&mut other_required_state);
58
59		self.required_state = self
60			.required_state
61			.iter()
62			.fold(
63				HashMap::<StateEventType, HashSet<String>>::new(),
64				|mut the_map, (state_event_type, state_key)| {
65					if the_map
66						.get(&StateEventType::from("*"))
67						.is_some_and(|the_state_keys| the_state_keys.get("*").is_some())
68						|| state_event_type.to_string() == "*" && state_key == "*"
69					{
70						return vec![(StateEventType::from("*"), ["*".to_owned()].into())]
71							.into_iter()
72							.collect();
73					};
74
75					if state_key == "*" {
76						// replace every other existing key
77						the_map.insert(state_event_type.clone(), [state_key.clone()].into());
78						return the_map;
79					}
80
81					// if there is already a more general type that covers this
82					// case, we don't need to add the more specific.
83					if the_map
84						.get(&StateEventType::from("*"))
85						.is_some_and(|keys| keys.contains(state_key))
86					{
87						return the_map;
88					}
89					let keys = the_map.entry(state_event_type.clone()).or_default();
90					if !keys.contains("*") {
91						keys.insert(state_key.clone());
92					}
93
94					the_map
95				},
96			)
97			.into_iter()
98			.fold(vec![], |mut the_list, (state_event_type, keys)| {
99				keys.into_iter().for_each(|key| the_list.push((state_event_type.clone(), key)));
100				the_list
101			});
102	}
103}
104
105/// Tracks the upper bound of a stream. That is, up to which point it was synced.
106///
107/// These states are stored in the `syncv4_room_streams` table.
108#[derive(Debug, Clone, PartialEq, Eq)]
109pub enum StreamUpperBound {
110	/// A status of `LIVE` means the stream is fully caught up to the last sync
111	/// v4 request's token.
112	Live,
113	/// A status of `PREVIOUSLY(StreamId)` means the stream was sent down to the
114	/// client up to the `up_to` stream ID. During a request, if a room that was
115	/// `LIVE` before is not part of that request's matched data and we detect
116	/// that that stream has had new data, we change that `LIVE` to
117	/// `PREVIOUSLY(last token)` to know that whenever that is matched again, we
118	/// can send data from that point onwards.
119	Previously(StreamId),
120}
arminius_lib_model/syncv4.rs

arminius_lib_model/
syncv4.rs
