arminius_lib_sync/syncv4/connect/
mod.rs

1// SPDX-FileCopyrightText: 2026 BWI GmbH
2//
3// SPDX-License-Identifier: Apache-2.0
4
5//! Contains logic to resolve a sync v4 connection and produce a [ResolvedConnection].
6
7use arminius_lib_db::DatabaseError;
8use arminius_lib_db_sync::syncv4::TransactedSyncV4Queries;
9use ruma::{OwnedDeviceId, OwnedUserId};
10
11use crate::{sync_token, syncv4::error::SyncV4Error};
12
13/// Every sync v4 connection is uniquely identified by this `(user ID, device ID, connection ID)`
14/// triplet.
15#[derive(Debug)]
16pub struct ConnectionKey {
17	/// User owning the connection.
18	pub(crate) user_id: OwnedUserId,
19	/// Device of the user owning the connection.
20	pub(crate) device_id: OwnedDeviceId,
21	/// An optional string to identify this connection to the server.
22	pub(crate) conn_id: Option<String>,
23}
24
25impl ConnectionKey {
26	/// Helper to create a [ConnectionKey].
27	pub(crate) fn new(
28		user_id: OwnedUserId,
29		device_id: OwnedDeviceId,
30		conn_id: Option<String>,
31	) -> Self {
32		Self { user_id, device_id, conn_id }
33	}
34}
35
36/// An actual sync v4 connection, tied to a [ConnectionKey].
37#[derive(Debug)]
38pub struct ResolvedConnection {
39	/// The connection's `(user ID, device ID, connection ID)` triplet.
40	pub(crate) key: ConnectionKey,
41
42	/// The token we sent back to the client the last time they synced.
43	///
44	/// This token is used to
45	/// - a) Know which position we recorded (if any) when they last synced (using the
46	///   `syncv4_position` field of the token).
47	/// - b) Know up to which point the user already synced in the current connection.
48	pub(crate) prev_token: Option<sync_token::SyncToken>,
49	/// The token we will send back to the client after handling the current
50	/// sync request. This token represents the latest point in every streams
51	/// of the server.
52	pub(crate) next_token: sync_token::SyncToken,
53}
54
55impl ResolvedConnection {
56	/// Helper to get the previous token's `syncv4_position`, but only if it actually corresponds to
57	/// a stored position (i.e. it is `> 0`).
58	pub fn previous_stored_position(&self) -> Option<i64> {
59		self.prev_token
60			.as_ref()
61			.filter(|prev_token| prev_token.syncv4_position > 0)
62			.map(|prev_token| prev_token.syncv4_position)
63	}
64
65	/// Helper to get the event stream position of the previous token, if any.
66	#[must_use]
67	pub fn prev_events_stream_id(&self) -> Option<i64> {
68		self.prev_token.as_ref().map(|prev_token| prev_token.event_stream_position)
69	}
70
71	/// Helper to get a [String] from the next sync token.
72	#[must_use]
73	pub fn next_pos_as_string(&self) -> String {
74		String::from(&self.next_token)
75	}
76}
77
78/// Resolves a connection from its `(user ID, device ID, connection ID)` triplet.
79///
80/// This will also insert/delete a connection when needed.
81///
82/// # Errors
83///
84/// - [`SyncV4Error::SyncTokenParseError`]: Well, obviously, if we failed to parse the request's
85///   token.
86/// - [`SyncV4Error::InternalServerError`]: If anything went wrong on our side (most likely while
87///   communicating with the database).
88/// - [`SyncV4Error::UnknownPos`]: If the request contains a `pos` token but we did not find any
89///   connection stored for the given triplet, or that position is not found in the connection.
90/// - [`SyncV4Error::PosFromFuture`]: If the given `pos` contains a token that is past the server's
91///   current token (see [`sync_token::get_current_token`]).
92pub(super) async fn resolve<T: TransactedSyncV4Queries>(
93	transaction: &mut T,
94	key: ConnectionKey,
95	pos: Option<String>,
96) -> Result<ResolvedConnection, SyncV4Error> {
97	// Parse the `pos` token
98	let prev_token = pos.map(|pos| sync_token::SyncToken::try_from(pos.as_str())).transpose()?;
99
100	let connection = transaction
101		.fetch_syncv4_connection(&key.user_id, &key.device_id, key.conn_id.as_deref())
102		.await?;
103
104	match (connection, prev_token.as_ref()) {
105		// The connection does not exist and `pos` is unset -> New connection
106		(None, None) => {
107			tracing::debug!("Connection created");
108
109			transaction
110				.insert_syncv4_connection(&key.user_id, &key.device_id, key.conn_id.as_deref())
111				.await?;
112		}
113		// The connection does not exist but `pos` is set -> `400 - M_UNKNOWN_POS`
114		(None, Some(token)) => {
115			tracing::debug!("Request with `pos={:?}` without an existing connection", token);
116
117			return Err(SyncV4Error::UnknownPos);
118		}
119		// The connection exists but the `pos` is `None` -> Connection reset
120		(Some(_connection), None) => {
121			tracing::debug!("Connection reset");
122
123			transaction
124				.reset_connection(&key.user_id, &key.device_id, key.conn_id.as_deref())
125				.await?;
126		}
127		// The connection exists, and the `pos` is set -> Check that the position exists in the
128		// connection and is owned by the user
129		(Some(_connection), Some(prev_token)) => {
130			tracing::debug!("Handling request with `pos` = {:?}", prev_token);
131
132			if let Some(prev_position) =
133				(prev_token.syncv4_position > 0).then_some(prev_token.syncv4_position)
134			{
135				match transaction
136					.fetch_syncv4_position(
137						&key.user_id,
138						&key.device_id,
139						key.conn_id.as_deref(),
140						prev_position,
141					)
142					.await
143				{
144					Err(DatabaseError::SqlxError(sqlx::Error::RowNotFound)) => {
145						return Err(SyncV4Error::UnknownPos);
146					}
147					Err(other_error) => {
148						return Err(other_error.into());
149					}
150					Ok(()) => (),
151				}
152			}
153		}
154	}
155
156	let next_token = sync_token::get_current_token(
157		transaction,
158		prev_token.as_ref().map(|prev_token| prev_token.syncv4_position),
159	)
160	.await?;
161
162	// Make sure the request's token is not past the server's one (i.e. "in
163	// the future")
164	if prev_token.as_ref().is_some_and(|token| token.is_after(&next_token)) {
165		return Err(SyncV4Error::PosFromFuture);
166	}
167
168	Ok(ResolvedConnection { key, prev_token, next_token })
169}
arminius_lib_sync/syncv4/connect/mod.rs

arminius_lib_sync/syncv4/connect/
mod.rs
