arminius_lib_db_sync/

syncv4.rs

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

//! Queries related to sync v4 (Simplified Sliding Sync).

use std::collections::HashMap;

use arminius_lib_db::{Database, DatabaseError, ResultExt as _, TransactedDb};
use arminius_lib_events::{RoomEvent, Verified};
use arminius_lib_model::{event::EventVerifiedRow, syncv4::StreamUpperBound, types::StreamId};
use itertools::Itertools as _;
use ruma::{
	DeviceId, OwnedRoomId, RoomId, UserId,
	events::{AnyStrippedStateEvent, StateEventType},
	serde::Raw,
};

/// List of "bump events".
///
/// These events are used to compute the `bump_stamp` of rooms that are part of
/// sync responses (see <https://github.com/matrix-org/matrix-spec-proposals/blob/main/proposals/4186-simplified-sliding-sync.md#common-fields>).
///
/// For each room, we fetch the stream ID of the latest event of such type
/// (falls back to `0` if none), then we order the user's room using this stream
/// ID, and finally we convert that stream ID to just an index in the ordered
/// list.
///
/// ## Example
///
/// The user will get these 3 rooms as part of a sync request:
///
/// `room1(stream_id=4)`, `room2(stream_id=1)` and `room3(stream_id=8)`
///
/// The ordered list is therefore:
///
/// `[room2, room1, room3]`
///
/// And the answer will contain this array of rooms:
///
/// `[room2(bump_stamp=0), room1(bump_stamp=1), room3(bump_stamp=2)]`
pub const BUMP_EVENTS_TYPES: [&str; 7] = [
	"m.room.create",
	"m.room.message",
	"m.room.encrypted",
	"m.sticker",
	"m.call.invite",
	"m.poll.start",
	"m.beacon_info",
];

/// The various streams of our server.
#[derive(Debug, Clone, sqlx::Type, PartialEq, Eq, Hash)]
#[sqlx(type_name = "STREAM_NAME", rename_all = "snake_case")]
pub enum DbStreamName {
	/// Refers to the `eventids` sequence
	Events,
}

/// Represents whether the stream got fully synced or up to a specific point.
#[derive(Debug, Clone, sqlx::Type, PartialEq, Eq, Hash)]
#[sqlx(type_name = "STREAM_STATUS", rename_all = "UPPERCASE")]
pub enum DbStreamStatus {
	/// The stream was fully up-to-date with the last request's token.
	Live,
	/// The stream got synced up to a certain `up_to` token.
	Previously,
}

/// An existing connection in the database.
///
/// Connections are linked to `(user_id, device_id, conn_id)` triplets.
#[derive(Debug, Clone)]
pub struct DbSyncV4Connection {
	// FIXME(#2147): Add `last_used_at` to this struct, and update it.
}

/// Struct used to fetch the server's current set of stream IDs.
///
/// These are all the maximum allotted stream IDs, per each stream.
#[derive(Debug)]
pub struct DbMaxStreamIds {
	/// Maximum stream ID for the `events.stream_id`.
	pub max_event_stream_id: StreamId,
	/// Maximum stream ID for the `to_device_messages.stream_id`.
	pub max_todevice_stream_id: StreamId,
	/// Maximum stream ID for both `room_account_data.stream_id` and
	/// `global_account_data.stream_id`.
	pub max_account_data_stream_id: StreamId,
	/// Maximum stream ID for the `user_keys.stream_id`.
	pub max_user_keys_stream_id: StreamId,
	/// Maximum stream ID for the `read_receipts.stream_id`.
	pub max_read_receipts_stream_id: StreamId,
	/// Maximum stream ID for the `federated_invites.stream_id`.
	pub max_federated_invites_stream_id: StreamId,
	/// Maximum stream ID for the `federated_knocks.stream_id`.
	pub max_federated_knocks_stream_id: StreamId,
}

/// Information about a room, from a user's perspective.
#[derive(Debug, Clone)]
pub struct RoomInfo {
	/// Whether the user ever joined the room or not
	pub was_ever_joined: bool,
	/// Stream ID of the most recent event of the room.
	pub max_event_stream_id: StreamId,
	/// Stream ID of the most recent bump event of the room. See
	/// [`BUMP_EVENTS_TYPES`].
	pub bump_event_stream_id: StreamId,
}

/// Struct used when fetching the timeline of synced room.
// TODO: Add required states and fetch them at once ?
#[derive(Debug, Default)]
pub struct SyncedRoomTimeline {
	/// The actual events of the timeline
	pub timeline: Vec<RoomEvent<Verified>>,
	/// Boolean indicating whether the room's `timeline_limit` was hit or not
	/// (i.e. if there were actually more events in there).
	pub limited: bool,
	/// Number of events that occurred after the given `from`.
	pub num_live: usize,
}

/// Contains the state of a room.
#[derive(Debug)]
pub struct RoomState {
	/// The events stream status applying to the last time it was synced.
	// TODO: When we add more than just the 'events' stream, this should become a
	// `HashMap<DbStreamName, StreamStatus>` containing all of these streams statuses
	pub events_stream_upper_bound: StreamUpperBound,
}

/// All the syncv4 related queries
// Queries are usually used as extension traits so suggest `import as _`.
#[rust_analyzer::prefer_underscore_import]
pub trait TransactedSyncV4Queries: Database {
	/// Inserts a new sync v4 connection given its `(user ID, device ID, connection ID)` triplet.
	#[tracing::instrument(skip(self))]
	async fn insert_syncv4_connection(
		&mut self,
		user_id: &UserId,
		device_id: &DeviceId,
		conn_id: Option<&str>,
	) -> Result<(), DatabaseError> {
		sqlx::query!(
			r#"
			INSERT INTO syncv4_connections (device_nid, conn_id)
				VALUES ((
						SELECT
							devices.nid
						FROM
							devices
							JOIN users ON devices.user_nid = users.nid
						WHERE
							users.user_id = $1
							AND devices.device_id = $2),
						$3)
			"#,
			user_id.to_string(),
			device_id.to_string(),
			conn_id
		)
		.execute(self.connection().await?)
		.await?;

		Ok(())
	}

	/// Deletes all the data associated to a connection.
	#[tracing::instrument(skip(self))]
	async fn reset_connection(
		&mut self,
		user_id: &UserId,
		device_id: &DeviceId,
		conn_id: Option<&str>,
	) -> Result<(), DatabaseError> {
		sqlx::query!(
			r#"
			DELETE FROM syncv4_positions USING syncv4_connections, devices, users
			WHERE syncv4_positions.connection_nid = syncv4_connections.nid
				AND syncv4_connections.device_nid = devices.nid
				AND devices.user_nid = users.nid
				AND users.user_id = $1
				AND devices.device_id = $2
				AND syncv4_connections.conn_id IS NOT DISTINCT FROM $3
			"#,
			user_id.as_str(),
			device_id.as_str(),
			conn_id
		)
		.execute(self.connection().await?)
		.await?;

		Ok(())
	}

	/// Tries to fetch a user's connection given its `(user ID, device ID,
	/// connection ID)` triplet.
	#[tracing::instrument(skip(self))]
	async fn fetch_syncv4_connection(
		&mut self,
		user_id: &UserId,
		device_id: &DeviceId,
		conn_id: Option<&str>,
	) -> Result<Option<DbSyncV4Connection>, DatabaseError> {
		// FIXME(#2147): Update `last_used_at` of the connection if found.
		Ok(sqlx::query_scalar!(
			r#"
			SELECT EXISTS (
				SELECT 1
				FROM syncv4_connections
					JOIN devices ON syncv4_connections.device_nid = devices.nid
					JOIN users ON devices.user_nid = users.nid
				WHERE
					users.user_id = $1
					AND devices.device_id = $2
					AND syncv4_connections.conn_id IS NOT DISTINCT FROM $3
			) AS "exists!"
			"#,
			user_id.to_string(),
			device_id.to_string(),
			conn_id,
		)
		.fetch_one(self.connection().await?)
		.await?
		.then_some(DbSyncV4Connection {}))
	}

	/// Fetches a position that belongs to a specific connection.
	///
	/// Errors if the position was not found.
	#[tracing::instrument(skip(self))]
	async fn fetch_syncv4_position(
		&mut self,
		user_id: &UserId,
		device_id: &DeviceId,
		conn_id: Option<&str>,
		position: i64,
	) -> Result<(), DatabaseError> {
		sqlx::query!(
			r#"
			SELECT
				1 AS "exists"
			FROM
				syncv4_positions
				JOIN syncv4_connections ON syncv4_positions.connection_nid = syncv4_connections.nid
				JOIN devices ON syncv4_connections.device_nid = devices.nid
				JOIN users ON devices.user_nid = users.nid
			WHERE
				users.user_id = $1
				AND devices.device_id = $2
				AND syncv4_connections.conn_id IS NOT DISTINCT FROM $3
				AND syncv4_positions.position = $4
			"#,
			user_id.to_string(),
			device_id.to_string(),
			conn_id,
			position
		)
		.fetch_one(self.connection().await?)
		.await?;

		Ok(())
	}

	/// Inserts a new position in the database, after having deleted any sibling
	/// of its previous position, as well as the parent of that previous
	/// position.
	#[tracing::instrument(skip(self))]
	async fn new_syncv4_position(
		&mut self,
		user_id: &UserId,
		device_id: &DeviceId,
		conn_id: Option<&str>,
		prev_position: Option<i64>,
	) -> Result<StreamId, DatabaseError> {
		// Delete the sibling(s) and parent of the previous connection if any
		sqlx::query!(
			"
   			DELETE FROM syncv4_positions USING syncv4_connections, users, devices
   			WHERE syncv4_connections.nid = syncv4_positions.connection_nid
						AND syncv4_connections.device_nid = devices.nid
						AND devices.user_nid = users.nid
						AND users.user_id = $1
						AND devices.device_id = $2
   			    AND conn_id IS NOT DISTINCT FROM $3
   			    AND prev_position IS DISTINCT FROM $4
   			    AND position IS DISTINCT FROM $4
   			",
			user_id.to_string(),
			device_id.to_string(),
			conn_id,
			prev_position
		)
		.execute(self.connection().await?)
		.await?;

		let new_position = sqlx::query_scalar!(
			r#"
			INSERT INTO syncv4_positions (connection_nid, prev_position)
			VALUES (
				(
					SELECT syncv4_connections.nid
					FROM syncv4_connections
						JOIN devices ON syncv4_connections.device_nid = devices.nid
						JOIN users ON devices.user_nid = users.nid
					WHERE
						users.user_id = $1
						AND devices.device_id = $2
						AND syncv4_connections.conn_id IS NOT DISTINCT FROM $3
				),
				$4
			)
			RETURNING position
			"#,
			user_id.to_string(),
			device_id.to_string(),
			conn_id,
			prev_position
		)
		.fetch_one(self.connection().await?)
		.await?;

		Ok(new_position)
	}

	/// Fetches all the maximum stream IDs of the various streams of Arminius.
	///
	/// This is used to compute the `next-pos` of sliding sync.
	#[tracing::instrument(skip(self))]
	async fn fetch_max_stream_ids(&mut self) -> Result<DbMaxStreamIds, DatabaseError> {
		let row = sqlx::query!(
			r#"
			SELECT
				(SELECT MAX(events.stream_id) FROM events) AS max_events_stream_id,
				(SELECT MAX(to_device_messages.stream_id) FROM to_device_messages) AS max_todevice_stream_id,
				(SELECT GREATEST(
					(SELECT MAX(global_account_data.stream_id) FROM global_account_data),
					(SELECT MAX(room_account_data.stream_id) FROM room_account_data)
				)) AS max_account_data_stream_id,
				(SELECT MAX(user_keys.stream_id) FROM user_keys) AS max_user_keys_stream_id,
				(SELECT MAX(read_receipts.stream_id) FROM read_receipts) AS max_read_receipts_stream_id,
				(SELECT MAX(federated_invites.stream_id) FROM federated_invites) AS max_federated_invites_stream_id,
				(SELECT MAX(federated_knocks.stream_id) FROM federated_knocks) AS max_federated_knocks_stream_id
			"#
		)
		.fetch_one(self.connection().await?)
		.await?;

		Ok(DbMaxStreamIds {
			max_event_stream_id: row.max_events_stream_id.unwrap_or(0),
			max_todevice_stream_id: row.max_todevice_stream_id.unwrap_or(0),
			max_account_data_stream_id: row.max_account_data_stream_id.unwrap_or(0),
			max_user_keys_stream_id: row.max_user_keys_stream_id.unwrap_or(0),
			max_read_receipts_stream_id: row.max_read_receipts_stream_id.unwrap_or(0),
			max_federated_invites_stream_id: row.max_federated_invites_stream_id.unwrap_or(0),
			max_federated_knocks_stream_id: row.max_federated_knocks_stream_id.unwrap_or(0),
		})
	}

	/// Fetches all the [`RoomInfo`] for all the rooms the user has a membership
	/// event in.
	#[tracing::instrument(skip(self))]
	async fn fetch_user_room_info(
		&mut self,
		user_id: &UserId,
	) -> Result<HashMap<OwnedRoomId, RoomInfo>, DatabaseError> {
		// TODO: For rooms the user is not joined to, `bump_stream_id` should be that of the
		// relevant membership event (e.g. when the user left the room).
		sqlx::query!(
			r#"
			SELECT
				room_membership_state.room_id AS "room_id!",
				(room_membership_state.membership = 'join'
					OR bool_or((events.event_json ->> 'type') = 'm.room.member'
					AND (events.event_json ->> 'state_key') = $1
					AND (events.event_json #>> '{content,membership}' = 'join'))) AS "has_ever_joined!",
				MAX(events.stream_id) AS "max_event_stream_id!",
				MAX(bump_events.stream_id) AS "bump_stream_id"
			FROM
				room_membership_state ($1)
				JOIN events ON room_membership_state.room_nid = events.room_nid
				-- PERF: This might need a pre-populated table to be less costly to fetch.
				LEFT JOIN events bump_events ON room_membership_state.room_nid = bump_events.room_nid
					AND (bump_events.event_json ->> 'type' = ANY ($2))
			GROUP BY
				room_membership_state.room_id,
				room_membership_state.membership
			"#,
			user_id.to_string(),
			&BUMP_EVENTS_TYPES as _
		)
		.fetch_all(self.connection().await?)
		.await?
		.into_iter()
		.map(|row| {
			let room_id = RoomId::parse(row.room_id).invalid_data("Invalid room ID in database")?;

			Ok((
				room_id,
				RoomInfo {
					was_ever_joined: row.has_ever_joined,
					max_event_stream_id: row.max_event_stream_id,
					bump_event_stream_id: row.bump_stream_id.unwrap_or(0),
				},
			))
		})
		.collect::<Result<_, DatabaseError>>()
	}

	/// Fetches rooms timelines for the given user.
	///
	/// Parameters:
	/// - `user_id`: The user for whom to fetch the visible timeline events.
	/// - `rooms`: A map from room ID to `(start, limit)` where `start` is the stream ID to start
	///   fetching events from, and `limit` is the maximum number of events to fetch for that room.
	/// - `last_sync_stream_id`: The last sync's highest events stream ID. This is used to compute
	///   `num_live` for each room.
	/// - `to_stream_id`: The maximum stream ID to not go past by when fetching events.
	///
	/// Returns a mapping from room IDs to [`SyncedRoomTimeline`].
	// TODO: Also fetch required states here !
	#[tracing::instrument(skip(self))]
	async fn fetch_syncv4_joined_rooms_timelines(
		&mut self,
		user_id: &UserId,
		rooms: &HashMap<OwnedRoomId, (StreamId, i64)>,
		last_sync_stream_id: StreamId,
		to_stream_id: StreamId,
	) -> Result<HashMap<OwnedRoomId, SyncedRoomTimeline>, DatabaseError> {
		let (rooms_ids, rooms_from, rooms_limits): (Vec<String>, Vec<i64>, Vec<i64>) = rooms
			.iter()
			// We add 1 to the timeline limits here so that we can tell if we reached them or
			// not
			.map(|(room_id, (from_stream_id, timeline_limit))| {
				(room_id.to_string(), from_stream_id, timeline_limit + 1)
			})
			.multiunzip();

		let mut room_timelines = sqlx::query!(
			r#"
			SELECT
				events.room_id AS "room_id!",
				events.room_limit AS "room_limit!",
				events.stream_id AS "stream_id!",
				events.event_json::text AS "event_json!: Box<sqlx::types::JsonRawValue>",
				events.room_version,
				events.attempted_at,
				events.signature_verification,
				events.soft_failed
			FROM
				UNNEST($2::text[], $3::bigint[], $4::bigint[]) AS room_params (room_id,
					room_from,
					room_limit)
				CROSS JOIN LATERAL (
					SELECT
						rooms.room_id,
						room_params.room_limit,
						events.stream_id,
						events.event_json,
						rooms.room_version,
						verification_metadata.attempted_at,
						verification_metadata.signature_verification,
						verification_metadata.soft_failed
					FROM
						visible_timeline_events ($1, room_params.room_id) events
						JOIN verification_metadata ON verification_metadata.event_nid = events.nid
						JOIN rooms ON rooms.room_id = room_params.room_id
					WHERE
						events.stream_id > room_params.room_from
						AND events.stream_id <= max_safe_seq ('eventids')
						AND events.stream_id <= $5
					ORDER BY
						events.stream_id DESC
					LIMIT room_params.room_limit) events
			ORDER BY
				events.stream_id DESC
			"#,
			user_id.to_string(),
			&rooms_ids,
			&rooms_from,
			&rooms_limits,
			to_stream_id
		)
		.fetch_all(self.connection().await?)
		.await?
		.into_iter()
		.try_fold(
			HashMap::new(),
			|mut result: HashMap<OwnedRoomId, SyncedRoomTimeline>, row| {
				let room_id =
					RoomId::parse(row.room_id).invalid_data("Invalid room ID in database")?;

				let room_timeline = result.entry(room_id).or_default();

				// If we've hit the limit, ignore the event and set `limited` to
				// true
				if room_timeline.timeline.len() == (row.room_limit - 1) as usize {
					room_timeline.limited = true;
					return Ok(result);
				}

				room_timeline.timeline.push(
					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,
					}
					.try_into_event()?,
				);

				if row.stream_id > last_sync_stream_id {
					room_timeline.num_live += 1;
				}

				Ok::<_, DatabaseError>(result)
			},
		)?;

		// Reverse every timelines, as their last events is expected to be the
		// most recent ones
		for timeline in room_timelines.values_mut() {
			timeline.timeline.reverse();
		}

		Ok(room_timelines)
	}

	/// Fetches the counts of 'join' and 'invite' users for a single room.
	#[tracing::instrument(skip(self))]
	async fn fetch_room_membership_counts(
		&mut self,
		room_id: &RoomId,
	) -> Result<(u64, u64), DatabaseError> {
		let counts = sqlx::query!(
			r#"
			SELECT
				COUNT(*) FILTER (WHERE events.event_json #>> '{content,membership}' = 'join') AS "joined_count!",
				COUNT(*) FILTER (WHERE events.event_json #>> '{content,membership}' = 'invite') AS "invited_count!"
			FROM
				rooms
				JOIN room_state_snapshots ON room_state_snapshots.room_nid = rooms.nid
				JOIN state_snapshot_state_events ON state_snapshot_state_events.state_snapshot_nid = room_state_snapshots.state_snapshot_nid
				JOIN events ON events.nid = state_snapshot_state_events.state_event_nid
			WHERE
				events.event_json ->> 'type' = 'm.room.member'
				AND rooms.room_id = $1
				AND events.event_json #>> '{content, membership}' IN ('join', 'invite')
			"#,
			room_id.to_string()
		).fetch_one(self.connection().await?).await?;

		Ok((counts.joined_count as u64, counts.invited_count as u64))
	}

	/// Fetches the list of stripped state events consisting the state of all
	/// the given `rooms`.
	#[tracing::instrument(skip(self))]
	async fn fetch_rooms_stripped_state(
		&mut self,
		rooms: &[OwnedRoomId],
	) -> Result<HashMap<OwnedRoomId, Vec<Raw<AnyStrippedStateEvent>>>, DatabaseError> {
		sqlx::query!(
			r#"
			SELECT
				rooms.room_id,
				rooms.room_version,
				events.event_json::text AS "event_json!: Box<sqlx::types::JsonRawValue>",
				verification_metadata.attempted_at,
				verification_metadata.signature_verification,
				verification_metadata.soft_failed
			FROM
				room_state_snapshots
				JOIN state_snapshot_state_events ON state_snapshot_state_events.state_snapshot_nid = room_state_snapshots.state_snapshot_nid
				JOIN events ON state_snapshot_state_events.state_event_nid = events.nid
				JOIN verification_metadata ON verification_metadata.event_nid = events.nid
				JOIN rooms ON events.room_nid = rooms.nid
			WHERE
				rooms.room_id = ANY ($1::text[])
			"#,
			&rooms.into_iter().map(ToString::to_string).collect::<Vec<_>>()
		).fetch_all(self.connection().await?)
			.await?
			.into_iter()
			.try_fold(HashMap::new(), |mut result, row| -> Result<_, DatabaseError> {
				let room_id = RoomId::parse(row.room_id).invalid_data("Invalid room ID in database")?;

				let verified_event = 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
				}.try_into_event()?;


				let stripped_state_event = verified_event.to_stripped_state_event().invalid_data("Non state event stored as one in the database")?;

				result.entry(room_id)
					.and_modify(|room_events: &mut Vec<_>| room_events.push(stripped_state_event.clone()))
					.or_insert_with(|| vec![stripped_state_event]);

				Ok(result)
		})
	}

	/// Fetches the list of required state events for a room.
	#[tracing::instrument(skip(self))]
	async fn fetch_rooms_required_states(
		&mut self,
		room_id: &RoomId,
		required_state_config: &Vec<(StateEventType, String)>,
		from_stream_id: StreamId,
		to_stream_id: StreamId,
	) -> Result<Vec<RoomEvent<Verified>>, DatabaseError> {
		let (event_types, state_keys): (Vec<String>, Vec<String>) = required_state_config
			.iter()
			.map(|(event_type, state_key)| (event_type.to_string(), state_key.to_owned()))
			.unzip();

		sqlx::query_as!(
			EventVerifiedRow,
			r#"
			WITH wanted_events AS (
				SELECT
					*
				FROM
					unnest($2::text[], $3::text[]) AS t (event_type,
						state_key))
			SELECT DISTINCT ON (events.nid)
				events.event_json AS "event_json: Box<sqlx::types::JsonRawValue>",
				rooms.room_version,
				verification_metadata.attempted_at,
				verification_metadata.signature_verification,
				verification_metadata.soft_failed
			FROM
				room_state_snapshots
				JOIN state_snapshot_state_events s ON room_state_snapshots.state_snapshot_nid = s.state_snapshot_nid
				JOIN events ON s.state_event_nid = events.nid
				JOIN verification_metadata ON verification_metadata.event_nid = events.nid
				JOIN rooms ON events.room_nid = rooms.nid
				JOIN wanted_events ON (wanted_events.event_type = '*'
						OR wanted_events.event_type = event_json ->> 'type')
					AND (wanted_events.state_key = '*'
						OR wanted_events.state_key = '$LAZY'
						OR wanted_events.state_key = event_json ->> 'state_key')
			WHERE
				rooms.room_id = $1
				AND events.stream_id > $4
				AND events.stream_id <= max_safe_seq ('eventids')
				AND events.stream_id <= $5
			"#,
				room_id.as_str(),
				&event_types,
				&state_keys,
				from_stream_id,
				to_stream_id
			)
		.fetch_all(self.connection().await?)
		.await?
		.into_iter()
		.map(EventVerifiedRow::try_into_event)
		.collect()
	}

	/// Fetches all of a user's stored room streams, but only for the 'events'
	/// stream.
	#[tracing::instrument(skip(self))]
	async fn fetch_events_room_streams(
		&mut self,
		position: i64,
	) -> Result<HashMap<OwnedRoomId, StreamUpperBound>, DatabaseError> {
		sqlx::query!(
			r#"
			SELECT
				room_id,
				stream_status AS "stream_status:DbStreamStatus",
				up_to
			FROM
				syncv4_room_streams
				JOIN rooms ON syncv4_room_streams.room_nid = rooms.nid
				JOIN syncv4_positions ON syncv4_room_streams.position = syncv4_positions.position
			WHERE
				syncv4_positions.position = $1
			"#,
			position
		)
		.fetch_all(self.connection().await?)
		.await?
		.into_iter()
		.map(|row| {
			let room_id = RoomId::parse(row.room_id).invalid_data("Invalid room ID in database")?;

			let event_stream_status = if row.stream_status == DbStreamStatus::Live {
				StreamUpperBound::Live
			} else {
				let up_to = row.up_to.ok_or(DatabaseError::InvalidData(
					"Entry with status 'previously' without its `up_to` set".to_owned(),
				))?;
				StreamUpperBound::Previously(up_to)
			};

			Ok((room_id, event_stream_status))
		})
		.collect::<Result<_, DatabaseError>>()
	}

	/// Updates the given room's `syncv4_room_streams` entries where
	/// `stream_name='events'` to become `PREVIOUSLY(up_to)`.
	#[tracing::instrument(skip(self))]
	async fn update_rooms_events_streams_to_previously(
		&mut self,
		position: i64,
		rooms_to_update: Vec<OwnedRoomId>,
		new_up_to: StreamId,
	) -> Result<(), DatabaseError> {
		let room_ids: Vec<String> =
			rooms_to_update.into_iter().map(|room_id| room_id.to_string()).collect();

		sqlx::query!(
			r#"
			UPDATE
				syncv4_room_streams
			SET
				stream_status = $4,
				up_to = $5
			FROM
				rooms
			WHERE
				syncv4_room_streams.position = $1
				AND syncv4_room_streams.stream_name = $2
				AND rooms.room_id = ANY ($3::text[])
			"#,
			position,
			DbStreamName::Events as _,
			&room_ids,
			DbStreamStatus::Previously as _,
			new_up_to
		)
		.execute(self.connection().await?)
		.await?;

		Ok(())
	}

	/// Insert all the room streams associated to a position.
	#[tracing::instrument(skip(self))]
	async fn insert_rooms_events_streams(
		&mut self,
		new_position: i64,
		rooms_streams: HashMap<OwnedRoomId, StreamUpperBound>,
	) -> Result<(), DatabaseError> {
		let (room_ids, stream_statuses, up_tos): (
			Vec<String>,
			Vec<DbStreamStatus>,
			Vec<Option<i64>>,
		) = rooms_streams
			.into_iter()
			.map(|(room_id, event_stream_status)| {
				let (stream_status, up_to) = match event_stream_status {
					StreamUpperBound::Previously(up_to) => {
						(DbStreamStatus::Previously, Some(up_to))
					}
					StreamUpperBound::Live => (DbStreamStatus::Live, None),
				};

				(room_id.to_string(), stream_status, up_to)
			})
			.multiunzip();

		sqlx::query!(
			r#"
			INSERT INTO syncv4_room_streams (position, stream_name, room_nid, stream_status, up_to)
			SELECT
				$1,
				$2,
				rooms.nid,
				params.stream_status,
				params.up_to
			FROM
				UNNEST($3::text[], $4::STREAM_STATUS[], $5::bigint[]) AS params (room_id,
					stream_status,
					up_to)
				JOIN rooms ON params.room_id = rooms.room_id
			"#,
			new_position,
			DbStreamName::Events as _,
			&room_ids,
			stream_statuses as _,
			up_tos as _
		)
		.execute(self.connection().await?)
		.await?;

		Ok(())
	}
}

impl<'c> TransactedSyncV4Queries for TransactedDb<'c> {}