arminius_lib_sync/
sync_token.rs

1// SPDX-FileCopyrightText: 2025 BWI GmbH
2//
3// SPDX-License-Identifier: Apache-2.0
4
5//! Sync tokens allow us to index our position into the stream of events and
6//! EDUs.
7//!
8//! See <https://bundesmessenger.gitlab.io/backend/neo/arminius/arminius/architecture/sync.html> for information
9//! on what we're indexing into.
10
11#![allow(
12	clippy::allow_attributes,
13	reason = "This source file contains enums which allow `clippy::missing_docs_in_private_items`. But an exception to this lint will mistakenly report that it's unfulfilled when used with expect, even though it is not."
14)]
15
16mod tests;
17
18use std::num::ParseIntError;
19
20use arminius_lib_component::error::MatrixApiError;
21use arminius_lib_db::DatabaseError;
22use arminius_lib_db_sync::syncv4::{DbMaxStreamIds, TransactedSyncV4Queries};
23use arminius_lib_model::types::StreamId;
24use itertools::Itertools as _;
25use reqwest::StatusCode;
26use ruma::api::error::ErrorKind;
27
28/// Error that occurs when parsing a sync token
29#[allow(
30	missing_docs,
31	clippy::missing_docs_in_private_items,
32	reason = "TODO(#2003): This exception predates us enabling `clippy::allow_attributes_without_reason`, so we don't know why it was enabled"
33)]
34#[derive(thiserror::Error, Debug, MatrixApiError, PartialEq, Eq)]
35#[matrix_api_error(status_code = StatusCode::BAD_REQUEST, error_kind = ErrorKind::InvalidParam)]
36#[error(transparent)]
37pub enum SyncTokenParseError {
38	#[error("Missing sync token prefix")]
39	MissingSyncTokenPrefix,
40
41	#[error("Missing event stream position")]
42	MissingEventStreamPosition,
43	#[error("Could not parse event stream position: {0}")]
44	InvalidEventStreamPosition(ParseIntError),
45
46	#[error("Missing to-device stream position")]
47	MissingToDeviceStreamPosition,
48	#[error("Could not parse to-device stream position: {0}")]
49	InvalidToDeviceStreamPosition(ParseIntError),
50
51	#[error("Missing account data stream position")]
52	MissingAccountDataStreamPosition,
53	#[error("Could not parse account data stream position: {0}")]
54	InvalidAccountDataStreamPosition(ParseIntError),
55
56	#[error("Missing user keys stream position")]
57	MissingUserKeysStreamPosition,
58	#[error("Could not parse user keys stream position: {0}")]
59	InvalidUserKeysStreamPosition(ParseIntError),
60
61	#[error("Missing read receipts stream position")]
62	MissingReadReceiptsStreamPosition,
63	#[error("Could not parse read receipts stream position: {0}")]
64	InvalidReadReceiptsStreamPosition(ParseIntError),
65
66	#[error("Missing federated invites stream position")]
67	MissingFederatedInvitesStreamPosition,
68	#[error("Could not parse federated invites stream position: {0}")]
69	InvalidFederatedInvitesStreamPosition(ParseIntError),
70
71	#[error("Missing sync v4 position")]
72	MissingSyncV4Position,
73	#[error("Could not parse sync v4 position: {0}")]
74	InvalidSyncV4Position(ParseIntError),
75
76	#[error("Missing federated invites stream position")]
77	MissingFederatedKnocksStreamPosition,
78	#[error("Could not parse federated invites stream position: {0}")]
79	InvalidFederatedKnocksStreamPosition(ParseIntError),
80}
81
82/// Prefix that correctly identifies a sync token
83const SYNC_TOKEN_PREFIX: &str = "s";
84
85/// A deserialized token which represents the state of sync.
86///
87/// NOTE: This token format is backwards compatible due to the parser
88/// ignoring junk at the end it does not understand
89#[derive(Clone, Debug, PartialEq, Eq, Default)]
90pub struct SyncToken {
91	/// Position in the event stream
92	pub event_stream_position: StreamId,
93	/// Position in the to_device stream
94	pub todevice_stream_position: StreamId,
95	/// Position in the account data stream
96	pub account_data_stream_position: StreamId,
97	/// Position in the user keys stream
98	pub user_keys_stream_position: StreamId,
99	/// Position in the read receipts stream
100	pub read_receipts_stream_position: StreamId,
101	/// Position in the federated invites stream
102	pub federated_invites_stream_position: StreamId,
103	/// ID of the sync v4 position
104	pub syncv4_position: StreamId,
105	/// Position in the federated knocks stream
106	pub federated_knocks_stream_position: StreamId,
107}
108
109impl SyncToken {
110	/// Returns `true` if `self > other` by comparing each **stream position**
111	/// fields one by one (not taking into account `syncv4_position`).
112	///
113	/// NOTE: We do not implement this as [`Ord`] nor [`PartialOrd`] to avoid
114	/// the possible footgun of comparing two tokens when unwanted.
115	#[must_use]
116	pub fn is_after(&self, other: &Self) -> bool {
117		self.event_stream_position > other.event_stream_position
118			|| self.todevice_stream_position > other.todevice_stream_position
119			|| self.account_data_stream_position > other.account_data_stream_position
120			|| self.user_keys_stream_position > other.user_keys_stream_position
121			|| self.read_receipts_stream_position > other.read_receipts_stream_position
122			|| self.federated_invites_stream_position > other.federated_invites_stream_position
123	}
124}
125
126impl TryFrom<&str> for SyncToken {
127	type Error = SyncTokenParseError;
128
129	fn try_from(input: &str) -> Result<Self, Self::Error> {
130		let mut parted = input.split_terminator("_");
131
132		if parted.next().is_none_or(|prefix| prefix != SYNC_TOKEN_PREFIX) {
133			return Err(SyncTokenParseError::MissingSyncTokenPrefix);
134		}
135
136		let event_stream_position: i64 = parted
137			.next()
138			.ok_or(SyncTokenParseError::MissingEventStreamPosition)?
139			.parse()
140			.map_err(SyncTokenParseError::InvalidEventStreamPosition)?;
141
142		let todevice_stream_position = parted
143			.next()
144			.ok_or(SyncTokenParseError::MissingToDeviceStreamPosition)?
145			.parse()
146			.map_err(SyncTokenParseError::InvalidToDeviceStreamPosition)?;
147
148		let account_data_stream_position = parted
149			.next()
150			.ok_or(SyncTokenParseError::MissingAccountDataStreamPosition)?
151			.parse()
152			.map_err(SyncTokenParseError::InvalidAccountDataStreamPosition)?;
153
154		let user_keys_stream_position = parted
155			.next()
156			.ok_or(SyncTokenParseError::MissingUserKeysStreamPosition)?
157			.parse()
158			.map_err(SyncTokenParseError::InvalidUserKeysStreamPosition)?;
159
160		let read_receipts_stream_position = parted
161			.next()
162			.ok_or(SyncTokenParseError::MissingReadReceiptsStreamPosition)?
163			.parse()
164			.map_err(SyncTokenParseError::InvalidReadReceiptsStreamPosition)?;
165
166		let federated_invites_stream_position = parted
167			.next()
168			.ok_or(SyncTokenParseError::MissingFederatedInvitesStreamPosition)?
169			.parse()
170			.map_err(SyncTokenParseError::InvalidFederatedInvitesStreamPosition)?;
171
172		let syncv4_position = parted
173			.next()
174			.ok_or(SyncTokenParseError::MissingSyncV4Position)?
175			.parse()
176			.map_err(SyncTokenParseError::InvalidSyncV4Position)?;
177
178		let federated_knocks_stream_position = parted
179			.next()
180			.ok_or(SyncTokenParseError::MissingFederatedKnocksStreamPosition)?
181			.parse()
182			.map_err(SyncTokenParseError::InvalidFederatedKnocksStreamPosition)?;
183
184		Ok(SyncToken {
185			event_stream_position,
186			todevice_stream_position,
187			account_data_stream_position,
188			user_keys_stream_position,
189			read_receipts_stream_position,
190			federated_invites_stream_position,
191			syncv4_position,
192			federated_knocks_stream_position,
193		})
194	}
195}
196
197impl From<&SyncToken> for String {
198	fn from(value: &SyncToken) -> Self {
199		let SyncToken {
200			event_stream_position,
201			todevice_stream_position,
202			account_data_stream_position,
203			user_keys_stream_position,
204			read_receipts_stream_position,
205			federated_invites_stream_position,
206			syncv4_position,
207			federated_knocks_stream_position,
208		} = value;
209
210		[
211			SYNC_TOKEN_PREFIX.to_owned(),
212			event_stream_position.to_string(),
213			todevice_stream_position.to_string(),
214			account_data_stream_position.to_string(),
215			user_keys_stream_position.to_string(),
216			read_receipts_stream_position.to_string(),
217			federated_invites_stream_position.to_string(),
218			syncv4_position.to_string(),
219			federated_knocks_stream_position.to_string(),
220		]
221		.iter()
222		.join("_")
223	}
224}
225
226/// Fetches the current token of the server.
227///
228/// That token is composed of all the maximum stream IDs ever emitted for each stream. The
229/// `syncv4_position` field of the token gets copied from the previous token's one if any, or
230/// defaulted to 0.
231pub async fn get_current_token<T: TransactedSyncV4Queries>(
232	transaction: &mut T,
233	prev_position: Option<i64>,
234) -> Result<SyncToken, DatabaseError> {
235	let DbMaxStreamIds {
236		max_event_stream_id,
237		max_todevice_stream_id,
238		max_account_data_stream_id,
239		max_user_keys_stream_id,
240		max_read_receipts_stream_id,
241		max_federated_invites_stream_id,
242		max_federated_knocks_stream_id,
243	} = transaction.fetch_max_stream_ids().await?;
244
245	Ok(SyncToken {
246		event_stream_position: max_event_stream_id,
247		todevice_stream_position: max_todevice_stream_id,
248		account_data_stream_position: max_account_data_stream_id,
249		user_keys_stream_position: max_user_keys_stream_id,
250		read_receipts_stream_position: max_read_receipts_stream_id,
251		federated_invites_stream_position: max_federated_invites_stream_id,
252		federated_knocks_stream_position: max_federated_knocks_stream_id,
253		// Keep the last request's position if any
254		syncv4_position: prev_position.unwrap_or(0),
255	})
256}
arminius_lib_sync/sync_token.rs

arminius_lib_sync/
sync_token.rs
