arminius_lib_sync/syncv4/execute/
persist.rs

1// SPDX-FileCopyrightText: 2026 BWI GmbH
2//
3// SPDX-License-Identifier: Apache-2.0
4
5//! Contains the logic to persist a sync v4 request handling result.
6
7use std::collections::{BTreeMap, HashMap};
8
9use arminius_lib_db::{DatabaseError, TransactedDb};
10use arminius_lib_db_sync::syncv4::TransactedSyncV4Queries;
11use arminius_lib_model::syncv4::StreamUpperBound;
12use ruma::OwnedRoomId;
13
14use crate::syncv4::{
15	connect,
16	error::SyncV4Error,
17	execute::{Persisted, fetch},
18	plan,
19};
20
21/// Persist the result of a sync v4 request handling.
22///
23/// In any case, this function will update the previous position's state to mark its stale rooms to
24/// PREVIOUSLY (if any).
25///
26/// If `fetched` is `Some`, it will also create a new position storing the new state for this
27/// fetched data.
28///
29/// Finally, this function will commit the given `transaction`, finalizing the request handling and
30/// producing a [Persisted].
31pub(super) async fn persist(
32	mut transaction: TransactedDb<'_>,
33	plan: plan::SyncV4Plan,
34	fetched: Option<fetch::Fetched>,
35) -> Result<Persisted, SyncV4Error> {
36	let plan::SyncV4Plan { mut connection, updated_state, lists_counts, stale_rooms, .. } = plan;
37	update_stale_rooms(&mut transaction, &connection, stale_rooms).await?;
38
39	// A new position is allocated only when we actually return data — an empty
40	// response leaves the connection's position where it already was.
41	let rooms = match fetched {
42		Some(fetch::Fetched { rooms }) => {
43			let new_position = new_position(&mut transaction, &mut connection).await?;
44
45			let new_streams: HashMap<OwnedRoomId, StreamUpperBound> = updated_state
46				.into_iter()
47				.map(|(room_id, state)| (room_id, state.events_stream_upper_bound))
48				.collect();
49			transaction.insert_rooms_events_streams(new_position, new_streams).await?;
50
51			rooms
52		}
53		None => BTreeMap::new(),
54	};
55
56	transaction.commit().await?;
57
58	Ok(Persisted { pos: connection.next_pos_as_string(), lists_counts, rooms })
59}
60
61/// Update the previous position's rooms to PREVIOUSLY(prev_events_stream_id) if any.
62///
63/// This helper is callable even if there is nothing to update (i.e. `stale_rooms` is empty).
64async fn update_stale_rooms<T: TransactedSyncV4Queries>(
65	transaction: &mut T,
66	connection: &connect::ResolvedConnection,
67	stale_rooms: Vec<OwnedRoomId>,
68) -> Result<(), DatabaseError> {
69	// Nothing to do on an initial request (no stored position to update), or if there are no new
70	// stale rooms.
71	let (Some(prev_position), Some(prev_events_stream_id), false) = (
72		connection.previous_stored_position(),
73		connection.prev_events_stream_id(),
74		stale_rooms.is_empty(),
75	) else {
76		return Ok(());
77	};
78
79	transaction
80		.update_rooms_events_streams_to_previously(
81			prev_position,
82			stale_rooms,
83			prev_events_stream_id,
84		)
85		.await
86}
87
88/// Creates a new position tied to the given connection.
89///
90/// NOTE: This also updates `connection.next_token` to point to that new position.
91///
92/// Returns the ID of the new position.
93async fn new_position<T: TransactedSyncV4Queries>(
94	transaction: &mut T,
95	connection: &mut connect::ResolvedConnection,
96) -> Result<i64, DatabaseError> {
97	let prev_position = connection.previous_stored_position();
98
99	let new_position = transaction
100		.new_syncv4_position(
101			&connection.key.user_id,
102			&connection.key.device_id,
103			connection.key.conn_id.as_deref(),
104			prev_position,
105		)
106		.await?;
107
108	connection.next_token.syncv4_position = new_position;
109
110	Ok(new_position)
111}
arminius_lib_sync/syncv4/execute/persist.rs

arminius_lib_sync/syncv4/execute/
persist.rs
