arminius_lib_sync/

sync.rs

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

//! Implements APIs for client sync API
//!
//! See <https://bundesmessenger.gitlab.io/backend/neo/arminius/arminius/architecture/sync.html> for information
//! about how the stream IDs work
//!
//! For limited timelines: We generally fetch up to `limit+1` events from the
//! database, where `limit` is the amount of events the user asked for, or a
//! sensible default. If there are actually `limit+1` events we can sync to the
//! user, we throw away the last event, and mark the timeline as `limited`,
//! indicating to the user that if they use the `/messages` endpoint, there are
//! more events they could fetch.

#![allow(
	clippy::allow_attributes,
	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."
)]

use std::{
	collections::{BTreeMap, HashMap, HashSet},
	time::Duration,
};

use arminius_lib_authentication::{AuthenticatedClaims, SessionInfo};
use arminius_lib_component::{
	MessagesToken, ServerState,
	error::{InternalServerError, MatrixApiError, ResultExt as _},
	ruma_requests::{ArminiusResponse, RumaRequest},
};
use arminius_lib_db::{Database as _, DefaultDb, TransactedDb};
use arminius_lib_db_account_data::{
	AccountDataQueries as _, SyncGlobalAccountDataEvent, SyncRoomAccountDataEvent,
};
use arminius_lib_db_filters::FilterQueries as _;
use arminius_lib_db_keys::{ChangedUserKeys, FallbackKeysQueries as _, UserKeysQueries as _};
use arminius_lib_db_read_receipts::{ReadReceiptsQueries as _, SyncRoomReceipts};
use arminius_lib_db_room::{EventQueries as _, StateQueries as _};
use arminius_lib_db_stripped_state::{FederatedInvite, StrippedStateQueries as _};
use arminius_lib_db_sync::{SyncEvent, SyncQueries as _};
use arminius_lib_db_todevice::{SyncToDeviceEvent, ToDeviceQueries as _};
use arminius_lib_events::{
	ClientContext, ContextualizedFields as _, DraftFields as _, RoomEvent, Verified,
};
use arminius_lib_macros::{FromDbError, FromNatsError};
use arminius_lib_model::{read_receipts::SyncReceiptsMap, types::StreamId};
use axum::{extract::State, response::IntoResponse};
use futures::StreamExt as _;
use reqwest::StatusCode;
use ruma::{
	OneTimeKeyAlgorithm, OwnedEventId, OwnedRoomId, OwnedUserId, UInt, UserId,
	api::{
		client::{
			filter::{EventFormat, FilterDefinition},
			sync::sync_events::{self, v3::Filter},
		},
		error::ErrorKind,
	},
	events::{
		AnyGlobalAccountDataEvent, AnySyncEphemeralRoomEvent, AnySyncStateEvent,
		AnySyncTimelineEvent, StateEventType, SyncEphemeralRoomEvent, TimelineEventType,
		receipt::{Receipt, ReceiptEventContent, ReceiptType},
		room::member::{MembershipState, RoomMemberEventContent},
	},
	serde::Raw,
};
use serde_json::{json, value::to_raw_value};
use ulid::Ulid;

use crate::sync_token::{SyncToken, SyncTokenParseError};

/// A room that is part of the sync response.
#[derive(Debug)]
enum Room {
	/// The requesting user is currently a member of the room.
	Join(sync_events::v3::JoinedRoom),
	/// The requesting user has left the room.
	Leave(sync_events::v3::LeftRoom),
	/// The requesting user is invited to the room.
	Invite {
		/// The user inviting us to a room
		inviter: Option<OwnedUserId>,
		/// The actual room information
		room: sync_events::v3::InvitedRoom,
	},
	/// The requesting user is knocking on the room.
	Knock(sync_events::v3::KnockedRoom),
}

/// A synchronized room
#[derive(Debug)]
struct SyncedRoom {
	/// The actual sync response room. See [`Room`].
	room: Room,
	/// For rooms that contain timelines, we also store these timelines
	/// minimal stream ID, so that we can put it in the response's `prev_batch`.
	min_stream_id: StreamId,
	/// The list of state events that were already added to that room. This is
	/// used to not send duplicates back.
	state_event_ids: HashSet<OwnedEventId>,
	/// The user performing the sync
	user: OwnedUserId,
}

impl SyncedRoom {
	/// Creates a new [`SyncedRoom`] which [`Room`] is of the correct type.
	pub fn new(user: OwnedUserId, state: &MembershipState) -> Result<Self, InternalServerError> {
		let room = match state {
			MembershipState::Invite => {
				Room::Invite { inviter: None, room: sync_events::v3::InvitedRoom::new() }
			}
			MembershipState::Join => Room::Join(sync_events::v3::JoinedRoom::new()),
			MembershipState::Leave | MembershipState::Ban => {
				Room::Leave(sync_events::v3::LeftRoom::new())
			}
			MembershipState::Knock => Room::Knock(sync_events::v3::KnockedRoom::new()),
			_ => None.unrecoverable_error("Unknown membership state")?,
		};

		Ok(Self { room, min_stream_id: StreamId::MAX, state_event_ids: HashSet::new(), user })
	}

	/// Adds the given `event` to the room.
	#[expect(
		clippy::expect_used,
		reason = "TODO(#2003): This exception predates us enabling `clippy::allow_attributes_without_reason`, so we don't know why it was enabled"
	)]
	pub fn add_event(
		&mut self,
		event: SyncEvent,
		client_context: ClientContext,
		max_synced_events: usize,
		event_format: &EventFormat,
	) -> Result<(), InternalServerError> {
		fn add_event_to_timeline(
			timeline: &mut sync_events::v3::Timeline,
			min_stream_id: &mut i64,
			event: SyncEvent,
			client_context: ClientContext,
			max_synced_events: usize,
			event_format: &EventFormat,
		) -> Result<(), InternalServerError> {
			if timeline.events.len() < max_synced_events - 1 {
				timeline.prev_batch = Some(String::from(&MessagesToken(event.event_id)));
				*min_stream_id = std::cmp::min(*min_stream_id, event.stream_id);
				let formatted_event = event.event.format_as(event_format, client_context);
				timeline.events.insert(0, Raw::<AnySyncTimelineEvent>::from_json(formatted_event));
			}
			// In case we reached the `arminius_lib_db_sync::MAX_SYNCED_EVENTS` limit, we set
			// the `prev_batch` of the response.
			else {
				timeline.limited = true;
				// SAFETY:  We just verified that we have several elements in
				// the timeline
				timeline.prev_batch = Some(String::from(&MessagesToken(
					timeline
						.events
						.first()
						.expect("We just verified that we have several elements in the timeline")
						.deserialize()
						.unrecoverable_error("Could not parse event in database")?
						.event_id()
						.to_owned(),
				)));
			}

			Ok(())
		}

		match &mut self.room {
			Room::Join(joined_room) => {
				add_event_to_timeline(
					&mut joined_room.timeline,
					&mut self.min_stream_id,
					event,
					client_context,
					max_synced_events,
					event_format,
				)?;
			}
			Room::Leave(left_room) => {
				add_event_to_timeline(
					&mut left_room.timeline,
					&mut self.min_stream_id,
					event,
					client_context,
					max_synced_events,
					event_format,
				)?;
			}
			Room::Invite { inviter, room } => {
				if event.event.event_type() == &TimelineEventType::RoomMember
					&& event
						.event
						.state_key()
						.is_some_and(|state_key| state_key == self.user.as_str())
				{
					*inviter = Some(event.event.sender().to_owned());
				}
				if event.event.state_key().is_some()
					&& ruma::events::RECOMMENDED_STRIPPED_STATE_EVENT_TYPES
						.contains(&StateEventType::from(event.event.event_type().to_string()))
				{
					room.invite_state.events.push(
						event
							.event
							.to_stripped_state_event()
							.expect("This must be a state event, because we just checked this"),
					);
				}
			}
			Room::Knock(knocked_room) => {
				if event.event.state_key().is_some()
					&& ruma::events::RECOMMENDED_STRIPPED_STATE_EVENT_TYPES
						.contains(&StateEventType::from(event.event.event_type().to_string()))
				{
					knocked_room.knock_state.events.push(
						event
							.event
							.to_stripped_state_event()
							.expect("This must be a state event, because we just checked this"),
					);
				}
			}
		}

		Ok(())
	}

	/// Adds the given `state_event` to the room.
	pub fn add_state_event(
		&mut self,
		event: &RoomEvent<Verified>,
		event_format: &EventFormat,
	) -> Result<(), InternalServerError> {
		if self.state_event_ids.contains(event.event_id()) {
			return Ok(());
		}
		let client_event = event.format_as(event_format, ClientContext::default());

		match &mut self.room {
			Room::Join(joined_room) => match &mut joined_room.state {
				sync_events::v3::State::Before(state_events) => {
					joined_room.timeline.limited = true;
					let event = Raw::<AnySyncStateEvent>::from_json(client_event);
					state_events.events.push(event);
				}
				sync_events::v3::State::After(_after) => {
					tracing::warn!("Parameter `use_state_after` not yet supported, skipping.");
				}
				_unknown => {
					tracing::error!("Unknown state changes type, skipping.");
				}
			},
			Room::Leave(left_room) => match &mut left_room.state {
				sync_events::v3::State::Before(state_events) => {
					left_room.timeline.limited = true;
					let event = Raw::<AnySyncStateEvent>::from_json(client_event);
					state_events.events.push(event);
				}
				sync_events::v3::State::After(_after) => {
					tracing::warn!("Parameter `use_state_after` not yet supported, skipping.");
				}
				_unknown => {
					tracing::error!("Unknown state changes type, skipping.");
				}
			},
			Room::Invite { inviter, room } => {
				if let Some(state_key) = event.state_key() {
					let is_recommended_state_event =
						ruma::events::RECOMMENDED_STRIPPED_STATE_EVENT_TYPES
							.contains(&StateEventType::from(event.event_type().to_string()));

					let state_key_is_user = state_key == self.user.as_str();
					let state_key_is_inviter =
						inviter.as_ref().is_some_and(|inviter| state_key == inviter);
					let is_recommended_member_event = event.event_type()
						== &TimelineEventType::RoomMember
						&& (state_key_is_user || state_key_is_inviter);

					if is_recommended_state_event || is_recommended_member_event {
						#[expect(
							clippy::expect_used,
							reason = "`to_stripped_state_event()` fails on non-state events. But we check for a state key directly above, so we know that this call is infallible"
						)]
						room.invite_state.events.push(
							event
								.to_stripped_state_event()
								.expect("This must be a state event, because we just checked this"),
						);
					}
				}
			}
			Room::Knock(knocked_room) => {
				if event.state_key().is_some()
					&& ruma::events::RECOMMENDED_STRIPPED_STATE_EVENT_TYPES
						.contains(&StateEventType::from(event.event_type().to_string()))
					|| (event.state_key() == Some(self.user.as_str())
						&& event.event_type() == &TimelineEventType::RoomMember)
				{
					#[expect(
						clippy::expect_used,
						reason = "`to_stripped_state_event()` fails on non-state events. But we check for a state key directly above, so we know that this call is infallible"
					)]
					knocked_room.knock_state.events.push(
						event
							.to_stripped_state_event()
							.expect("This must be a state event, because we just checked this"),
					);
				}
			}
		}

		self.state_event_ids.insert(event.event_id().to_owned());

		Ok(())
	}

	/// Adds the given room account data event to the room.
	///
	/// This throws an error if the rooms in self is neither a `Joined` nor a
	/// `Left` room (as such rooms do cannot contain account data in sync
	/// responses), or if we fail to deserialize the given event into a
	/// [`ruma::events::AnyRoomAccountDataEvent`].
	pub fn add_account_data_event(
		&mut self,
		event: &SyncRoomAccountDataEvent,
	) -> Result<(), InternalServerError> {
		match &mut self.room {
			Room::Join(joined_room) => {
				joined_room.account_data.events.push(Raw::from_json(
					to_raw_value(&event.event).unrecoverable_error(
						"Could not parse room account data event from database",
					)?,
				));
			}
			Room::Leave(left_room) => {
				left_room.account_data.events.push(Raw::from_json(
					to_raw_value(&event.event).unrecoverable_error(
						"Could not parse room account data event from database",
					)?,
				));
			}
			_ => {
				None.unrecoverable_error(&format!(
					"Impossible to add room account data events to '{}' rooms",
					event.membership
				))?;
			}
		}

		Ok(())
	}

	/// Adds the given read receipts to the room.
	pub fn add_read_receipts(
		&mut self,
		receipts: BTreeMap<OwnedEventId, BTreeMap<ReceiptType, BTreeMap<OwnedUserId, Receipt>>>,
	) -> Result<(), InternalServerError> {
		let Room::Join(joined_room) = &mut self.room else {
			return None.unrecoverable_error("Cannot add read receipts to non-joined rooms.")?;
		};

		let any_ephemeral: AnySyncEphemeralRoomEvent = AnySyncEphemeralRoomEvent::Receipt(
			SyncEphemeralRoomEvent::new(ReceiptEventContent(receipts)),
		);

		let event = json!({
			"type": any_ephemeral.event_type(),
			"content": any_ephemeral.content(),
		});

		joined_room.ephemeral.events.push(Raw::from_json(
			to_raw_value(&event).unrecoverable_error("Failed to convert to raw value")?,
		));

		Ok(())
	}
}

/// A wrapper around a sync response
#[derive(Debug)]
pub(crate) struct SyncResponse {
	/// The sync token to return
	sync_token: SyncToken,
	/// The user for who we are syncing data
	user: OwnedUserId,
	/// Mapping of all the rooms involved in the `/sync` request.
	rooms: HashMap<OwnedRoomId, SyncedRoom>,
	/// The global account data events of the user.
	account_data_events: Vec<Raw<AnyGlobalAccountDataEvent>>,
	/// To device messages
	to_device_messages: sync_events::v3::ToDevice,
	/// List of users who changed their keys or newly share a room since the
	/// last sync.
	changed_user_keys: Vec<OwnedUserId>,
	/// List of users who no longer share any encrypted room since the last
	/// sync.
	left_user_keys: Vec<OwnedUserId>,
	/// For each key algorithm, the number of unclaimed one-time keys currently
	/// held on the server for this device
	device_one_time_keys_count: BTreeMap<OneTimeKeyAlgorithm, UInt>,
	/// The unused fallback key algorithms.
	device_unused_fallback_key_types: Vec<OneTimeKeyAlgorithm>,
}

impl IntoResponse for SyncResponse {
	fn into_response(self) -> axum::response::Response {
		let mut response = sync_events::v3::Response::new(String::from(&self.sync_token));

		response.account_data.events = self.account_data_events;

		for (room_id, synced_room) in self.rooms {
			match synced_room.room {
				Room::Join(joined_room) => {
					response.rooms.join.insert(room_id, joined_room);
				}
				Room::Leave(left_room) => {
					response.rooms.leave.insert(room_id, left_room);
				}
				Room::Invite { room, .. } => {
					response.rooms.invite.insert(room_id, room);
				}
				Room::Knock(knocked_room) => {
					response.rooms.knock.insert(room_id, knocked_room);
				}
			}
		}

		response.to_device = self.to_device_messages;

		response.device_lists.changed = self.changed_user_keys;
		response.device_lists.left = self.left_user_keys;
		response.device_one_time_keys_count = self.device_one_time_keys_count;
		response.device_unused_fallback_key_types = Some(self.device_unused_fallback_key_types);

		ArminiusResponse(response).into_response()
	}
}

impl SyncResponse {
	/// Constructor
	fn new(sync_token: SyncToken, user: OwnedUserId) -> Self {
		Self {
			sync_token,
			user,
			rooms: HashMap::new(),
			account_data_events: Vec::new(),
			to_device_messages: sync_events::v3::ToDevice::new(),
			changed_user_keys: Vec::new(),
			left_user_keys: Vec::new(),
			device_one_time_keys_count: BTreeMap::new(),
			device_unused_fallback_key_types: Vec::new(),
		}
	}

	/// Add an event to the sync response
	pub fn add_event(
		&mut self,
		event: SyncEvent,
		client_context: ClientContext,
		max_synced_events: usize,
		event_format: &EventFormat,
	) -> Result<(), InternalServerError> {
		self.sync_token.event_stream_position =
			std::cmp::max(self.sync_token.event_stream_position, event.stream_id);

		let room = self
			.rooms
			.entry(event.room_id.clone())
			.or_insert(SyncedRoom::new(self.user.clone(), &event.membership)?);

		room.add_event(event, client_context, max_synced_events, event_format)
	}

	/// This adds the given room account data to the sync response.
	///
	/// Throws an error if the room type is not supported.
	pub fn add_room_account_data_event(
		&mut self,
		event: &SyncRoomAccountDataEvent,
	) -> Result<(), InternalServerError> {
		let room = self
			.rooms
			.entry(event.room_id.clone())
			.or_insert(SyncedRoom::new(self.user.clone(), &event.membership)?);

		room.add_account_data_event(event)
	}

	/// Add a to_device message to the sync response
	pub fn add_to_device_message(
		&mut self,
		event: &SyncToDeviceEvent,
	) -> Result<(), InternalServerError> {
		self.sync_token.todevice_stream_position =
			std::cmp::max(event.stream_id, self.sync_token.todevice_stream_position);

		let event = json!({
			"type": event.event_type.to_string(),
			"sender": event.sender,
			"content": event.content,
		});

		self.to_device_messages.events.push(
			Raw::from_json_string(event.to_string())
				.unrecoverable_error("Could not parse to-device message in database")?,
		);

		Ok(())
	}

	/// Adds the given read receipts map to all their corresponding rooms of the
	/// sync response.
	pub fn add_read_receipts(
		&mut self,
		receipts: SyncReceiptsMap,
	) -> Result<(), InternalServerError> {
		for (room_id, room_receipts) in receipts {
			let room = self.rooms.entry(room_id.clone()).or_insert(SyncedRoom::new(
				self.user.clone(),
				// Read receipts are only for joined rooms.
				&MembershipState::Join,
			)?);

			room.add_read_receipts(room_receipts)?;
		}

		Ok(())
	}
}

#[allow(
	missing_docs,
	clippy::missing_docs_in_private_items,
	reason = "TODO(#2003): This exception predates us enabling `clippy::allow_attributes_without_reason`, so we don't know why it was enabled"
)]
#[derive(thiserror::Error, Debug, MatrixApiError, FromDbError, FromNatsError)]
pub(crate) enum SyncError {
	#[error(transparent)]
	InternalServerError(#[from] InternalServerError),
	#[error(transparent)]
	SyncTokenParseError(#[from] SyncTokenParseError),
	/// Filter ID on our homeserver is the nid of the row which is `i64` number.
	/// if it fails to parse, it means that the provided filter ID is not a
	/// valid one.
	#[matrix_api_error(status_code = StatusCode::BAD_REQUEST, error_kind = ErrorKind::InvalidParam)]
	#[error("The filter ID provided is not a valid filter ID")]
	InvalidFilterId,
	/// Filter that was requested cannot be found.
	#[error("Filter not found")]
	#[matrix_api_error(status_code = StatusCode::NOT_FOUND, error_kind = ErrorKind::NotFound)]
	UnknownFilter,
}

/// This structure contains any data that needs to be synchronized.
#[derive(Debug)]
pub struct SyncData {
	/// The global account data that needs to be synchronized.
	pub global_account_data_events: Vec<SyncGlobalAccountDataEvent>,
	/// The rooms account data events that needs to be synchronized.
	pub room_account_data_events: Vec<SyncRoomAccountDataEvent>,
	/// The room events that need to be synchronized.
	pub room_events: Vec<SyncEvent>,
	/// The to-device events that needs to be synchronized.
	pub todevice_events: Vec<SyncToDeviceEvent>,
	/// The list of users who either changed their keys or newly joined a room
	/// shared with the synchronizing user.
	pub changed_user_keys: Vec<ChangedUserKeys>,
	/// The list of users who either changed their keys or newly joined a room
	/// shared with the synchronizing user.
	pub left_user_keys: Vec<OwnedUserId>,
	/// The new and forward read receipts that were received since the last
	/// sync.
	pub read_receipts: SyncRoomReceipts,
	/// The invites the syncing user has received over federation
	pub federated_invites: Vec<(StreamId, FederatedInvite)>,
	/// The invites the syncing user has received over federation
	pub federated_knocks: Vec<(StreamId, FederatedInvite)>,
}

impl SyncData {
	/// Builds a [`SyncResponse`] from a pair of a [`SyncToken`] and some
	/// [`SyncData`].
	///
	/// It also fetches any needed additional data from the transaction (e.g.
	/// events state history, etc)
	async fn build_sync_response<'c>(
		self,
		transaction: &mut TransactedDb<'c>,
		user_id: &UserId,
		device_id: &ruma::DeviceId,
		sync_token: SyncToken,
		filter_definition: Option<&FilterDefinition>,
	) -> Result<SyncResponse, InternalServerError> {
		let mut sync_response = SyncResponse::new(sync_token.clone(), user_id.to_owned());

		for account_data_event in self.global_account_data_events {
			sync_response.sync_token.account_data_stream_position = sync_response
				.sync_token
				.account_data_stream_position
				.max(account_data_event.stream_id);

			let raw_event = Raw::from_json(
				to_raw_value(&account_data_event.event)
					.unrecoverable_error("Converting global account data to raw")?,
			);
			sync_response.account_data_events.push(raw_event);
		}

		let max_fetched_timeline_events: usize = filter_definition
			.and_then(|fd| {
				fd.room
					.timeline
					.limit
					.and_then(|limit| usize::try_from(limit).map(|limit| limit + 1).ok())
			})
			.unwrap_or(arminius_lib_db_sync::MAX_SYNCED_EVENTS);

		let event_format = filter_definition
			.map(|filter_definition| filter_definition.event_format.clone())
			.unwrap_or_default();

		for room_event in self.room_events {
			let client_context = transaction
				.fetch_client_context(&room_event.room_id, &room_event.event_id, user_id, device_id)
				.await?;
			sync_response.add_event(
				room_event,
				client_context,
				max_fetched_timeline_events,
				&event_format,
			)?;
		}

		for (room_id, synced_room) in &mut sync_response.rooms {
			let state_events = SyncData::fetch_state_for_room(
				transaction,
				synced_room,
				user_id,
				room_id,
				&sync_token,
			)
			.await?;
			for state_event in state_events {
				synced_room.add_state_event(&state_event, &event_format)?;
			}
		}

		for account_data_event in self.room_account_data_events {
			sync_response.sync_token.account_data_stream_position = sync_response
				.sync_token
				.account_data_stream_position
				.max(account_data_event.stream_id);

			sync_response.add_room_account_data_event(&account_data_event)?;
		}

		for todevice_event in &self.todevice_events {
			sync_response.add_to_device_message(todevice_event)?;
		}

		for changed_user_keys in self.changed_user_keys {
			sync_response.sync_token.user_keys_stream_position =
				sync_response.sync_token.user_keys_stream_position.max(changed_user_keys.stream_id);

			sync_response.changed_user_keys.push(changed_user_keys.user_id);
		}

		for (stream_id, invite) in self.federated_invites {
			let mut room = sync_events::v3::InvitedRoom::new();
			room.invite_state.events = invite.state_events.into_iter().map(Into::into).collect();
			sync_response.rooms.insert(
				invite.room_id,
				SyncedRoom {
					room: Room::Invite { inviter: None, room },
					min_stream_id: stream_id,
					state_event_ids: HashSet::new(),
					user: invite.invited_user,
				},
			);
			sync_response.sync_token.federated_invites_stream_position =
				sync_token.federated_invites_stream_position.max(stream_id + 1);
		}

		for (stream_id, knock) in self.federated_knocks {
			let mut room = sync_events::v3::KnockedRoom::new();
			room.knock_state.events = knock.state_events.into_iter().map(Into::into).collect();
			sync_response.rooms.insert(
				knock.room_id,
				SyncedRoom {
					room: Room::Knock(room),
					min_stream_id: stream_id,
					state_event_ids: HashSet::new(),
					user: knock.invited_user,
				},
			);
			sync_response.sync_token.federated_knocks_stream_position =
				sync_token.federated_knocks_stream_position.max(stream_id + 1);
		}

		sync_response.left_user_keys = self.left_user_keys;

		sync_response.sync_token.read_receipts_stream_position =
			sync_token.read_receipts_stream_position.max(self.read_receipts.max_stream_id);
		sync_response.add_read_receipts(self.read_receipts.receipts)?;

		Ok(sync_response)
	}

	/// Given a synced room, fill out the `state` field
	///
	/// If too many messages have been sent since the last `/sync` call, we need
	/// to still send all the state events that the user will miss when the
	/// returned timeline has a gap
	///
	/// If this is the first time this room is synced since we joined or were
	/// invited to the room, we also need to get the entire state of the room
	async fn fetch_state_for_room<'c>(
		transaction: &mut TransactedDb<'c>,
		synced_room: &SyncedRoom,
		user_id: &UserId,
		room_id: &ruma::RoomId,
		sync_token: &SyncToken,
	) -> Result<Vec<RoomEvent<Verified>>, InternalServerError> {
		match synced_room.room {
			// For a room we're invited to, just fetch the current state
			Room::Invite { .. } | Room::Knock(_) => {
				// FIXME(#2071): The stripped state fetched should only depend
				// on the invite event found in the original sync events,
				// and not on the current state of the room
				let invite_event = transaction
					.fetch_room_state(room_id)
					.await?
					.remove(&(StateEventType::RoomMember, user_id.to_string()))
					.unrecoverable_error(
						"Room state was recorded as invite, but the last membership event is not invite",
					)?;
				let mut state: Vec<_> = transaction
					.fetch_state_before(room_id, invite_event.event_id())
					.await?
					.into_values()
					// Manually remove the users previous member event, we only want the current
					// member event
					.filter(|event| {
						event.event_type() != &TimelineEventType::RoomMember
							|| event.state_key() != Some(user_id.as_str())
					})
					.collect();
				// The actual membership event isn't in the `state_before` of
				// the membership event, and must be added manually
				state.push(invite_event);
				Ok(state)
			}
			Room::Join(_) | Room::Leave(_) => {
				// Checks whether we joined the room recently
				//
				// Specifically, check if we start the timeline with a join
				// event, and the state before our timeline starts is not
				// another join event. If we're overwriting a join event
				// with another join event, we're not actually joining the room,
				// just modifying a displayname, avatar, or similar
				if let Some(first_membership_event) = transaction
					.fetch_new_room_membership_event(
						room_id,
						user_id,
						sync_token.event_stream_position,
					)
					.await?
				{
					let first_membership = serde_json::from_str::<RoomMemberEventContent>(
						first_membership_event.content().get(),
					)
					.unrecoverable_error("Could not parse membership event in DB")?
					.membership;
					let previous_membership = transaction
						.fetch_membership_before(
							room_id,
							first_membership_event.event_id(),
							user_id,
						)
						.await?
						.map(|event| {
							serde_json::from_str::<RoomMemberEventContent>(event.content().get())
								.unrecoverable_error("Invalid member event in database")
						})
						.transpose()?
						.map(|event| event.membership);

					if first_membership == MembershipState::Join
						&& previous_membership != Some(MembershipState::Join)
					{
						// If we just joined the room since last sync, we return
						// the complete room state, up until the timeline
						// starts
						return Ok(transaction
							.fetch_incremental_room_state_for_user(
								room_id,
								0,
								synced_room.min_stream_id,
							)
							.await?);
					}
				}
				// If there's no new membership event, or the new membership
				// event is just an update to a previous membership event
				// with the same membership state, return the standard
				// incremental room state
				Ok(transaction
					.fetch_incremental_room_state_for_user(
						room_id,
						sync_token.event_stream_position,
						synced_room.min_stream_id,
					)
					.await?)
			}
		}
	}
}

/// This function is a central point in which we query every possible data that
/// needs synchronization.
async fn fetch_sync_data<'c>(
	transaction: &mut TransactedDb<'c>,
	session: &SessionInfo,
	initial: bool,
	sync_token: &mut SyncToken,
	filter_definition: Option<&FilterDefinition>,
) -> Result<Option<SyncData>, InternalServerError> {
	let room_filters = filter_definition.map(|filter_definition| &filter_definition.room);
	let new_room_events = transaction
		.fetch_sync_room_events(&session.user_id, sync_token.event_stream_position, room_filters)
		.await?;

	let new_global_account_data_events = transaction
		.fetch_sync_global_account_data_events(
			&session.user_id,
			sync_token.account_data_stream_position,
			filter_definition.map(|filter_definition| &filter_definition.account_data),
		)
		.await?;

	let new_room_account_data_events = transaction
		.fetch_sync_room_account_data_events(
			&session.user_id,
			sync_token.account_data_stream_position,
			filter_definition.map(|filter_definition| &filter_definition.room.account_data),
		)
		.await?;

	let new_todevice_events = transaction
		.fetch_sync_todevice_messages(
			&session.user_id,
			&session.device_id,
			sync_token.todevice_stream_position,
		)
		.await?;

	// Only populate device lists changes on incremental sync
	let (changed_user_keys, users_that_left_rooms) = match initial {
		// If it's the first request, just need to push the sync token's user keys stream ID to the
		// server's latest one.
		true => {
			sync_token.user_keys_stream_position =
				transaction.fetch_current_user_keys_stream_id().await?;
			Default::default()
		}
		// Otherwise, fetch the key changes that occurred since the last request's token
		false => (
			transaction
				.fetch_sync_user_keys_changes(
					&session.user_id,
					sync_token.event_stream_position,
					sync_token.user_keys_stream_position,
				)
				.await?,
			transaction
				.fetch_sync_user_keys_left(&session.user_id, sync_token.event_stream_position)
				.await?,
		),
	};

	let new_read_receipts = transaction
		.fetch_sync_read_receipts(
			&session.user_id,
			sync_token.read_receipts_stream_position,
			room_filters.map(|room_filter| &room_filter.ephemeral),
		)
		.await?;

	let new_federated_invites = transaction
		.fetch_federated_invites(&session.user_id, sync_token.federated_invites_stream_position)
		.await?;

	let new_federated_knocks = transaction
		.fetch_federated_knocks(&session.user_id, sync_token.federated_knocks_stream_position)
		.await?;

	Ok((!new_room_events.is_empty()
		|| !new_todevice_events.is_empty()
		|| !new_global_account_data_events.is_empty()
		|| !new_room_account_data_events.is_empty()
		|| !changed_user_keys.is_empty()
		|| !users_that_left_rooms.is_empty()
		|| !new_read_receipts.receipts.is_empty()
		|| !new_federated_invites.is_empty()
		|| !new_federated_knocks.is_empty())
	.then_some(SyncData {
		global_account_data_events: new_global_account_data_events,
		room_account_data_events: new_room_account_data_events,
		room_events: new_room_events,
		todevice_events: new_todevice_events,
		changed_user_keys,
		left_user_keys: users_that_left_rooms,
		read_receipts: new_read_receipts,
		federated_invites: new_federated_invites,
		federated_knocks: new_federated_knocks,
	}))
}

/// Given a [`arminius_lib_nats::Subscriber`] instanced from
/// `sync_data_subscriber`, wait until it receives a NATS message about a new
/// sync response.
///
/// Then, it checks in the database if there are any new data to synchronize
/// whenever a new message is received by that subscriber, using a call to
/// [`fetch_sync_data`].
async fn wait_for_sync_data<'c>(
	mut sync_update_subscriber: arminius_lib_nats::Subscriber,
	transaction: &mut TransactedDb<'c>,
	session: &SessionInfo,
	initial: bool,
	sync_token: &mut SyncToken,
	filter_definition: Option<&FilterDefinition>,
	timeout: Duration,
) -> Result<Option<SyncData>, InternalServerError> {
	match tokio::time::timeout(timeout, async {
		loop {
			if sync_update_subscriber.next().await.is_some() {
				let sync_data =
					fetch_sync_data(transaction, session, initial, sync_token, filter_definition)
						.await?;

				if sync_data.is_some() {
					return Ok::<_, InternalServerError>(sync_data);
				}
			}
		}
	})
	.await
	{
		// Whether we found something or we failed at it, return immediately
		Ok(sync_data) => Ok(sync_data?),
		// We still try one last time, just in case we missed a notification.
		Err(_timeout) => {
			fetch_sync_data(transaction, session, initial, sync_token, filter_definition).await
		}
	}
}

/// Since the client can send us a filter on the fly we need to normalize the
/// the request.
///
/// We do not need this inside a transaction since we don't need the same sort
/// of integrity as other sync queries.
async fn normalize_filter(
	database: &mut DefaultDb,
	user_id: &UserId,
	filter: Option<Filter>,
) -> Result<Option<FilterDefinition>, SyncError> {
	match filter {
		Some(Filter::FilterDefinition(filter_definition)) => Ok(Some(filter_definition)),
		Some(Filter::FilterId(filter_id_raw)) => {
			let filter_id =
				filter_id_raw.parse::<Ulid>().map_err(|_| SyncError::InvalidFilterId)?;

			Ok(Some(
				database
					.fetch_filter_with_id(filter_id, user_id)
					.await?
					.ok_or(SyncError::UnknownFilter)?,
			))
		}
		_ => Ok(None),
	}
}

/// Handler for `GET /_matrix/client/v3/sync` (See
/// <https://spec.matrix.org/v1.14/client-server-api/#get_matrixclientv3sync>).
///
/// Endpoint that returns state of the data and is used by clients to sync their
/// data with the homeserver.
///
/// **Warning:** Whenever you add support for any other data to be synced in
/// this endpoint (e.g account_data, keys, etc), make sure to call
/// `arminius_lib_nats::connection::NatsClient::notify_new_sync_data` on
/// the creator of such data that will get synced.
pub(crate) async fn sync_events_handler(
	state: State<ServerState>,
	AuthenticatedClaims(claims): AuthenticatedClaims,
	RumaRequest(request): RumaRequest<sync_events::v3::Request>,
) -> Result<SyncResponse, SyncError> {
	// FIXME(#680): Remove before release, very GDPR unfriendly
	tracing::info!(?request, "Received sync request");
	let initial = request.since.is_none();
	let mut sync_token =
		request.since.as_deref().map(SyncToken::try_from).transpose()?.unwrap_or_default();

	let mut database = state.database.clone();
	let filter_definition =
		normalize_filter(&mut database, &claims.user_id, request.filter).await?;

	let mut transaction = database.begin_transaction().await?;

	// We can already delete past synced to-device messages (only called when it's not an initial
	// sync).
	if !initial {
		transaction
			.delete_synced_todevice_messages(
				&claims.user_id,
				&claims.device_id,
				sync_token.todevice_stream_position,
			)
			.await?;
	}

	// Set up the subscriber that listens for sync notifications, before doing
	// the first DB call Doing it afterwards, we risk the following flow of
	// events:
	//  - We check the database for new stuff to sync, find nothing
	//  - A new message to sync gets added to the database, and a NATS message gets sent which no
	//    one hears
	//  - We start listening for NATS messages. There are none.
	//  - We eventually time out.
	let subscriber = state.nats_connection.sync_data_subscriber().await?;

	let mut sync_data = fetch_sync_data(
		&mut transaction,
		&claims,
		initial,
		&mut sync_token,
		filter_definition.as_ref(),
	)
	.await?;

	// NOTE: This default/fallback timeout is defined in the Matrix
	// specifications.
	let timeout = request.timeout.unwrap_or(Duration::ZERO);

	// Try fetching data in a loop if we did not find any on the first try and
	// we have a timeout
	if sync_data.is_none() && timeout != Duration::ZERO {
		sync_data = wait_for_sync_data(
			subscriber,
			&mut transaction,
			&claims,
			initial,
			&mut sync_token,
			filter_definition.as_ref(),
			timeout,
		)
		.await?;
	}

	// No matter if we found data or not, convert it to a response
	let mut sync_response = match sync_data {
		Some(sync_data) => {
			sync_data
				.build_sync_response(
					&mut transaction,
					&claims.user_id,
					&claims.device_id,
					sync_token,
					filter_definition.as_ref(),
				)
				.await?
		}
		None => SyncResponse::new(sync_token, claims.user_id.clone()),
	};

	// Add the data that will always be in the sync response, no matter if it is
	// syncing anything new
	sync_response.device_one_time_keys_count =
		transaction.fetch_otk_counts(&claims.user_id, &claims.device_id).await?;

	sync_response.device_unused_fallback_key_types = transaction
		.fetch_unused_fallback_keys_algorithms(&claims.user_id, &claims.device_id)
		.await?;

	transaction.commit().await?;

	// FIXME(#680): Remove before release, very GDPR unfriendly
	tracing::info!(?sync_response, "Made sync response");

	Ok(sync_response)
}