arminius_lib_db_sync/

lib.rs

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

//! Queries related to syncing.

pub mod syncv4;
mod tests;

use arminius_lib_db::{Database, DatabaseError, ResultExt as _};
use arminius_lib_events::{RoomEvent, Verified};
use arminius_lib_model::{event::EventVerifiedRow, types::StreamId};
use ruma::{
	EventId, OwnedEventId, OwnedRoomId, RoomId, UInt, UserId,
	api::client::filter::{RoomEventFilter, RoomFilter},
	events::room::member::MembershipState,
};

/// Maximum number (e.g. SQL `LIMIT`) of events retrieved by
/// [`SyncQueries::fetch_sync_room_events`].
pub const MAX_SYNCED_EVENTS: usize = 1000;

/// [`MAX_SYNCED_EVENTS`], converted to `i64`.
///
/// There's no reasonable way to do this in Rust in compile-time.
/// See <https://github.com/rust-lang/rust/issues/143874>
/// But there's a test case that checks they're identical at least.
const MAX_SYNCED_EVENTS_I64: i64 = 1000;

/// Default number (e.g. SQL `LIMIT`) of events retrieved by
/// [`SyncQueries::fetch_sync_room_events`].
pub const DEFAULT_SYNCED_EVENTS: i64 = 10;

/// Maximum number (e.g. SQL `LIMIT`) of messages retrieved by
/// [`SyncQueries::fetch_room_messages_backwards`] and
/// [`SyncQueries::fetch_room_messages_forwards`].
pub const MAX_MESSAGES: usize = 1000;

/// [`MAX_MESSAGES`], converted to `i64`.
///
/// There's no reasonable way to do this in Rust in compile-time.
/// See <https://github.com/rust-lang/rust/issues/143874>
/// But there's a test case that checks they're identical at least.
const MAX_MESSAGES_I64: i64 = 1000;

/// Structure used when querying events for joined rooms. Used for sync
#[derive(Debug)]
pub struct SyncEvent {
	/// The stream_id of the event
	pub stream_id: StreamId,
	/// The event_id of the event
	pub event_id: OwnedEventId,
	/// The ID of the room associated with the event.
	pub room_id: OwnedRoomId,
	/// The users membership of the room, at the point the sync event was
	/// fetched
	pub membership: MembershipState,
	/// The event JSON
	pub event: RoomEvent<Verified>,
}

/// Queries related to sync
// Queries are usually used as extension traits so suggest `import as _`.
#[rust_analyzer::prefer_underscore_import]
pub trait SyncQueries: Database {
	/// Fetches the earliest membership event for a user in a room, since
	/// `from_stream_id`
	#[tracing::instrument(skip(self))]
	async fn fetch_new_room_membership_event(
		&mut self,
		room_id: &RoomId,
		user_id: &UserId,
		from_stream_id: StreamId,
	) -> Result<Option<RoomEvent<Verified>>, DatabaseError> {
		sqlx::query_as!(
			EventVerifiedRow,
			r#"
			SELECT
				events.event_json AS "event_json: Box<sqlx::types::JsonRawValue>",
				rooms.room_version,
				verification.attempted_at,
				verification.signature_verification,
				verification.soft_failed
			FROM
				events
				JOIN rooms ON events.room_nid = rooms.nid
				JOIN verification_metadata verification ON verification.event_nid = events.nid
			WHERE
				rooms.room_id = $1
				AND (events.event_json ->> 'type') = 'm.room.member'
				AND (events.event_json ->> 'state_key') = $2
				AND events.stream_id > $3
				AND events.stream_id <= max_safe_seq ('eventids')
			ORDER BY
				stream_id ASC
			LIMIT 1
			"#,
			room_id.as_str(),
			user_id.as_str(),
			from_stream_id
		)
		.fetch_optional(self.connection().await?)
		.await?
		.map(EventVerifiedRow::try_into_event)
		.transpose()
	}

	/// Fetches all room events the user is able to see, starting at
	/// `from_stream_id`.
	#[tracing::instrument(skip(self))]
	async fn fetch_sync_room_events(
		&mut self,
		user: &UserId,
		from_stream_id: StreamId,
		room_filters: Option<&RoomFilter>,
	) -> Result<Vec<SyncEvent>, DatabaseError> {
		let limit = room_filters
			.and_then(|room_filter| room_filter.timeline.limit.map(i64::from))
			.map_or(DEFAULT_SYNCED_EVENTS, |client_requested_limit| {
				i64::min(MAX_SYNCED_EVENTS_I64, client_requested_limit)
			});
		let included_rooms = room_filters.and_then(|room_filter| {
			room_filter.rooms.as_ref().map(|included_rooms| {
				included_rooms.iter().map(OwnedRoomId::to_string).collect::<Vec<_>>()
			})
		});
		let excluded_rooms = room_filters
			.map(|room_filter| {
				room_filter.not_rooms.iter().map(OwnedRoomId::to_string).collect::<Vec<_>>()
			})
			.unwrap_or_default();
		let included_event_types =
			room_filters.and_then(|room_filter| room_filter.timeline.types.clone());
		let blocked_event_types = room_filters
			.map(|room_filter| room_filter.timeline.not_types.clone())
			.unwrap_or_default();

		let included_senders = room_filters.and_then(|room_filter| {
			room_filter
				.clone()
				.timeline
				.senders
				.map(|senders| senders.iter().map(ToString::to_string).collect::<Vec<_>>())
		});
		let excluded_senders = room_filters
			.map(|room_filter| {
				room_filter.timeline.not_senders.iter().map(ToString::to_string).collect::<Vec<_>>()
			})
			.unwrap_or_default();

		sqlx::query_file!(
			"src/sync.sql",
			user.as_str(),
			from_stream_id,
			limit,
			included_rooms.as_deref(),
			&excluded_rooms,
			included_event_types.as_deref(),
			&blocked_event_types,
			included_senders.as_deref(),
			&excluded_senders
		)
		.fetch_all(self.connection().await?)
		.await?
		.into_iter()
		.map(|row| -> Result<SyncEvent, DatabaseError> {
			let event_row = EventVerifiedRow {
				event_json: row.event_json,
				room_version: row.room_version,
				attempted_at: row.attempted_at,
				signature_verification: row.signature_verification,
				soft_failed: row.soft_failed,
			};

			let event = event_row.try_into_event()?;

			Ok(SyncEvent {
				stream_id: row.stream_id,
				event_id: row.event_id.parse().invalid_data("Invalid event ID in database")?,
				room_id: row.room_id.parse().invalid_data("Invalid room ID in database")?,
				membership: MembershipState::from(row.membership),
				event,
			})
		})
		.collect::<Result<Vec<_>, _>>()
	}

	/// Fetch messages forwards starting from the given event ID. If no event ID
	/// is given, the starting point is the earliest event visible to the given
	/// user.
	///
	/// The returned boolean is `true` if there are more than `limit` messages.
	#[tracing::instrument(skip(self))]
	async fn fetch_room_messages_forwards(
		&mut self,
		user: &UserId,
		room: &RoomId,
		from_event_id: Option<&EventId>,
		limit: UInt,
		room_filters: &RoomEventFilter,
	) -> Result<(Vec<RoomEvent<Verified>>, bool), DatabaseError> {
		// We try to get n+1  messages to know if there is a need to paginate.
		let limit = i64::min(i64::from(limit), MAX_MESSAGES_I64) + 1;
		let included_senders = room_filters
			.clone()
			.senders
			.map(|senders| senders.iter().map(ToString::to_string).collect::<Vec<_>>());
		let excluded_senders =
			room_filters.not_senders.iter().map(ToString::to_string).collect::<Vec<_>>();

		// NOTE: This query is hand-formatted because the formatter chokes on
		// the comments
		let mut events = sqlx::query_as!(
			EventVerifiedRow,
			"
			SELECT
				event_json AS \"event_json!: Box<sqlx::types::JsonRawValue>\",
				rooms.room_version,
				verification.attempted_at,
				verification.signature_verification,
				verification.soft_failed
			FROM
				visible_timeline_events ($2, $1) events
				JOIN event_ordering ON events.nid = event_ordering.event_nid
				JOIN rooms ON rooms.nid = events.room_nid
				JOIN verification_metadata verification ON verification.event_nid = events.nid
			WHERE
				-- The ordering value for the event should be greater than the one of from_event_id
				-- if given
				($3::text IS NULL OR ordering > (
					SELECT
						ordering
					FROM
						event_ordering
						JOIN events ON event_ordering.event_nid = events.nid
					WHERE
						events.event_id = $3::text))
				-- The type should match the allowlist in the room filter if given
				AND ($5::text[] IS NULL
					OR events.event_json ->> 'type' = ANY ($5::text[]))
				-- The type should not be in the denylist from the room filter
				AND NOT (events.event_json ->> 'type' = ANY ($6::text[]))
				-- The sender should match the allowlist given in the room filter
				AND ($7::text[] IS NULL
					OR events.event_json ->> 'sender' = ANY ($7::text[]))
				-- The sender should not be in the denylist given in the room filter
				AND NOT (events.event_json ->> 'sender' = ANY ($8::text[]))
				AND events.stream_id <= max_safe_seq ('eventids')
			ORDER BY
				ordering ASC
			LIMIT $4
			",
			room.as_str(),
			user.as_str(),
			from_event_id.map(EventId::as_str),
			limit,
			room_filters.types.as_deref(),
			&room_filters.not_types,
			included_senders.as_deref(),
			&excluded_senders
		)
		.fetch_all(self.connection().await?)
		.await?
		.into_iter()
		.map(EventVerifiedRow::try_into_event)
		.collect::<Result<Vec<_>, _>>()?;

		let there_are_more = events.len() == limit as usize;
		if there_are_more {
			events.pop();
		}

		Ok((events, there_are_more))
	}

	/// Fetch messages backwards starting from the given event ID. If no event
	/// ID is given, the starting point is the last event visible to the given
	/// user.
	///
	/// The returned boolean is `true` if there are more than `limit` messages.
	#[tracing::instrument(skip(self))]
	async fn fetch_room_messages_backwards(
		&mut self,
		user: &UserId,
		room: &RoomId,
		from_event_id: Option<&EventId>,
		limit: UInt,
		room_filters: &RoomEventFilter,
	) -> Result<(Vec<RoomEvent<Verified>>, bool), DatabaseError> {
		// We try to get n+1  messages to know if there is a need to paginate.
		let limit = i64::min(i64::from(limit), MAX_MESSAGES_I64) + 1;
		let included_senders = room_filters
			.clone()
			.senders
			.map(|senders| senders.iter().map(ToString::to_string).collect::<Vec<_>>());
		let excluded_senders =
			room_filters.not_senders.iter().map(ToString::to_string).collect::<Vec<_>>();

		// NOTE: This query is hand-formatted because the formatter chokes on
		// the comments
		let mut events = sqlx::query_as!(
			EventVerifiedRow,
			"
			SELECT
			    event_json AS \"event_json!: Box<sqlx::types::JsonRawValue>\",
			    rooms.room_version,
			    verification.attempted_at,
			    verification.signature_verification,
			    verification.soft_failed
			FROM
				visible_timeline_events ($2, $1) events
				JOIN event_ordering ON events.nid = event_ordering.event_nid
				JOIN rooms ON rooms.nid = events.room_nid
				JOIN verification_metadata verification ON verification.event_nid = events.nid
			WHERE
			    -- The ordering value for the event should be less than the one of from_event_id
			    -- if given
				($3::text IS NULL OR ordering < (
					SELECT
						ordering
					FROM
						event_ordering
						JOIN events ON event_ordering.event_nid = events.nid
					WHERE
						events.event_id = $3::text))
				-- The type should match the allowlist in the room filter if given
				AND ($5::text[] IS NULL
					OR events.event_json ->> 'type' = ANY ($5::text[]))
				-- The type should not be in the denylist from the room filter
				AND NOT (events.event_json ->> 'type' = ANY ($6::text[]))
				-- The sender should match the allowlist given in the room filter
				AND ($7::text[] IS NULL
					OR events.event_json ->> 'sender' = ANY ($7::text[]))
				-- The sender should not be in the denylist given in the room filter
				AND NOT (events.event_json ->> 'sender' = ANY ($8::text[]))
				AND events.stream_id <= max_safe_seq ('eventids')
			ORDER BY
				ordering DESC
			LIMIT $4
			",
			room.as_str(),
			user.as_str(),
			from_event_id.map(EventId::as_str),
			limit,
			room_filters.types.as_deref(),
			&room_filters.not_types,
			included_senders.as_deref(),
			&excluded_senders
		)
		.fetch_all(self.connection().await?)
		.await?
		.into_iter()
		.map(EventVerifiedRow::try_into_event)
		.collect::<Result<Vec<_>, _>>()?;

		let there_are_more = events.len() == limit as usize;
		if there_are_more {
			events.pop();
		}

		Ok((events, there_are_more))
	}

	/// Get all state changes between two points in the timeline
	#[tracing::instrument(skip(self))]
	async fn fetch_incremental_room_state_for_user(
		&mut self,
		room_id: &RoomId,
		from_event_id: StreamId,
		to_stream_id: StreamId,
	) -> Result<Vec<RoomEvent<Verified>>, DatabaseError> {
		// FIXME(#370): Implement event visibility rules (passing an additional
		// `user_id`).
		sqlx::query_as!(
			EventVerifiedRow,
			r#"
			SELECT DISTINCT ON (state_events.nid)
				state_events.event_json AS "event_json!: Box<sqlx::types::JsonRawValue>",
				rooms.room_version,
				verification.attempted_at,
				verification.signature_verification,
				verification.soft_failed
			FROM
				events
				JOIN rooms ON rooms.nid = events.room_nid
				CROSS JOIN LATERAL (
					SELECT
						event_state.nid,
						event_state.event_id,
						event_state.depth,
						event_state.event_json::text,
						event_state.received_ts
					FROM
						state_before (events.nid) event_state
					WHERE
						event_state.stream_id >= $2) AS state_events
				JOIN verification_metadata verification ON verification.event_nid = state_events.nid
			WHERE
				events.room_nid = (
					SELECT
						nid
					FROM
						rooms
					WHERE
						room_id = $1)
				AND events.stream_id >= $2
				AND events.stream_id <= $3
			ORDER BY
				state_events.nid
			"#,
			room_id.to_string(),
			from_event_id,
			to_stream_id,
		)
		.fetch_all(self.connection().await?)
		.await?
		.into_iter()
		.map(EventVerifiedRow::try_into_event)
		.collect()
	}
}

/// Make the query available on types implementing the
/// Database trait.
impl<T> SyncQueries for T where T: Database {}