arminius_lib_sync/syncv4/plan/

apply_state.rs

// SPDX-FileCopyrightText: 2026 BWI GmbH
//
// SPDX-License-Identifier: Apache-2.0

//! Contains logic to apply the per-connection state to an already filtered list of rooms (see
//! [MatchedRoom]).

mod tests;

use std::collections::HashMap;

use arminius_lib_db_sync::syncv4::{RoomInfo, RoomState};
use arminius_lib_model::{
	syncv4::{RoomConfig, StreamUpperBound},
	types::StreamId,
};
use ruma::OwnedRoomId;

use crate::syncv4::plan::MatchedRoom;

/// A room to sync, produced by [`apply_state`].
///
/// We differentiate between rooms the user is or has been a member of and the ones they never have.
#[derive(Debug)]
pub enum RoomToSync {
	/// A room to sync that the user is or has been a member of.
	Joined {
		/// From which point this room should be synced.
		from_stream_id: StreamId,
		/// Configuration to apply when fetching this room.
		room_config: RoomConfig,
		/// Whether it is the first time this room is synced or not.
		initial: bool,
	},
	/// Never joined rooms will only have their stripped state fetched.
	NeverJoined {
		/// Whether it is the first time this room is synced or not.
		initial: bool,
	},
}

impl RoomToSync {
	/// Helper to construct a [RoomToSync].
	///
	/// If `was_ever_joined` is `true`, this will produce a [RoomToSync::Joined], otherwise a
	/// [RoomToSync::NeverJoined].
	#[must_use]
	pub fn new_conditional(
		was_ever_joined: bool,
		room_config: RoomConfig,
		sync_room_from: StreamId,
		initial: bool,
	) -> Self {
		match was_ever_joined {
			true => RoomToSync::Joined { from_stream_id: sync_room_from, room_config, initial },
			false => RoomToSync::NeverJoined { initial },
		}
	}
}

/// Result produced by [`apply_state`]
#[derive(Debug)]
pub(super) struct AppliedState {
	/// The rooms that will be part of the current sync.
	pub(super) rooms_to_sync: HashMap<OwnedRoomId, RoomToSync>,
	/// The rooms belonging to the previous position and that should be marked to PREVIOUSLY during
	/// the current sync.
	pub(super) stale_rooms: Vec<OwnedRoomId>,
}

/// Apply the current connection's state (i.e. rooms streams and previously used configs) to the
/// rooms matched by the request.
///
/// This function will also update the state to reflect what is being synced.
pub(super) fn apply_state(
	prev_events_stream_id: StreamId,
	matched_rooms: &HashMap<OwnedRoomId, MatchedRoom>,
	user_rooms: &HashMap<OwnedRoomId, RoomInfo>,
	rooms_states: &mut HashMap<OwnedRoomId, RoomState>,
) -> AppliedState {
	let rooms_to_sync = mark_rooms_to_sync(prev_events_stream_id, matched_rooms, rooms_states);
	let stale_rooms =
		mark_stale_rooms(rooms_states, user_rooms, &rooms_to_sync, prev_events_stream_id);

	AppliedState { rooms_to_sync, stale_rooms }
}

/// Computes the list of rooms to sync by applying the rooms states to the rooms matched by the
/// request.
///
/// This will also update the `rooms_states` by marking all the rooms to sync as LIVE.
fn mark_rooms_to_sync(
	prev_events_stream_id: StreamId,
	matched_rooms: &HashMap<OwnedRoomId, MatchedRoom>,
	rooms_states: &mut HashMap<OwnedRoomId, RoomState>,
) -> HashMap<OwnedRoomId, RoomToSync> {
	let mut rooms_to_sync = HashMap::new();

	for (room_id, MatchedRoom { room_info, room_config }) in matched_rooms {
		// Check if we have a state for that room (i.e. if it was synced before within the
		// current connection)
		match rooms_states.get(room_id) {
			// The room has been synced previously
			Some(prev_room_state) => {
				if let Some(sync_from) =
					// If the room was LIVE and there are new events in it, sync it from the
					// last token's point
					if prev_room_state.events_stream_upper_bound
						== StreamUpperBound::Live
						&& room_info.max_event_stream_id > prev_events_stream_id
					{
						Some(prev_events_stream_id)
					}
					// If the room was PREVIOUSLY(up_to) and there are new events past
					// `up_to`, sync it from `up_to`
					else if let StreamUpperBound::Previously(up_to) =
						prev_room_state.events_stream_upper_bound
						&& room_info.max_event_stream_id > up_to
					{
						Some(up_to)
					}
					// Otherwise, don't sync it
					else {
						None
					} {
					rooms_to_sync.insert(
						room_id.clone(),
						RoomToSync::new_conditional(
							room_info.was_ever_joined,
							room_config.clone(),
							sync_from,
							false,
						),
					);
				} else {
					continue;
				}
			}
			// If the room was never synced, perform initial sync of it
			None => {
				rooms_to_sync.insert(
					room_id.clone(),
					RoomToSync::new_conditional(
						room_info.was_ever_joined,
						room_config.clone(),
						0,
						true,
					),
				);
			}
		}

		// If we reach here, we're syncing the current room, so we need to mark its state as
		// LIVE (overwrites any existing state for it if any).
		// TODO: Also update the room config we store for it
		rooms_states.insert(
			room_id.clone(),
			RoomState { events_stream_upper_bound: StreamUpperBound::Live },
		);
	}

	rooms_to_sync
}

/// Update and get the list of stale rooms.
///
/// A stale room is a room that was previously marked as LIVE, and contains new events in it
/// that will not be returned as part of the current request. These rooms therefore need to be
/// marked as `PREVIOUSLY` with the `up_to` set to the last request's events stream ID (because
/// that is the point up to which those rooms were LIVE).
pub(super) fn mark_stale_rooms(
	rooms_states: &mut HashMap<OwnedRoomId, RoomState>,
	user_rooms: &HashMap<OwnedRoomId, RoomInfo>,
	rooms_to_sync: &HashMap<OwnedRoomId, RoomToSync>,
	prev_events_stream_id: StreamId,
) -> Vec<OwnedRoomId> {
	rooms_states
		.iter_mut()
		.filter(|(room_id, room_state)| {
			// Only consider rooms that were marked as LIVE,
			room_state.events_stream_upper_bound == StreamUpperBound::Live
					// That were not part of the current sync
					&& !rooms_to_sync.contains_key(*room_id)
					// And contains new data
					&& user_rooms.get(*room_id).is_some_and(|room_info| {
						room_info.max_event_stream_id > prev_events_stream_id
					})
		})
		.map(|(room_id, room_state)| {
			// Mark the room as PREVIOUSLY(prev_max_event_stream_id)
			room_state.events_stream_upper_bound =
				StreamUpperBound::Previously(prev_events_stream_id);

			room_id.clone()
		})
		.collect()
}