tuwunel_api_client/sync/v5/

range.rs

use std::{collections::BTreeMap, iter::empty, mem::take};

use futures::{FutureExt, StreamExt, TryFutureExt, TryStreamExt, future::try_join4};
use ruma::{
	OwnedRoomId, OwnedUserId, RoomId,
	api::client::sync::sync_events::v5::response,
	events::{
		AnyRawAccountDataEvent, AnyRoomAccountDataEvent, AnySyncEphemeralRoomEvent,
		GlobalAccountDataEventType, ignored_user_list::IgnoredUserListEvent,
		receipt::SyncReceiptEvent,
	},
	serde::Raw,
};
use tokio::sync::OnceCell;
use tuwunel_core::{
	Error, Result, at, err, error, extract_variant, implement,
	result::NotFound,
	utils::{BoolExt, IterStream, TryReadyExt, stream::BroadbandExt},
};
use tuwunel_service::{
	rooms::read_receipt::{PrivateReadEvents, pack_receipts_fallible},
	sync::{Connection, DeliveredConfig, Room, RoomUpdate},
};

use super::{
	SyncInfo, Window, WindowRoom,
	rooms::{
		Failure as RoomFailure,
		Failure::{Payload as PayloadFailure, Timeline as TimelineFailure},
		RoomDetails, handle_room, membership_allows_required_state, merged_room_details,
		room_config,
	},
};
use crate::is_empty_account_data_event;

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum Domain {
	Timeline,
	Payload,
	PublicReceipt,
	PrivateRead,
	RoomAccountData,
	ReceiptSerialization,
}

#[derive(Debug)]
struct Failure {
	domain: Domain,
	error: Error,
}

impl Failure {
	fn new(domain: Domain, error: Error) -> Self { Self { domain, error } }
}

impl From<RoomFailure> for Failure {
	fn from(failure: RoomFailure) -> Self {
		match failure {
			| TimelineFailure(error) => Self::new(Domain::Timeline, error),
			| PayloadFailure(error) => Self::new(Domain::Payload, error),
		}
	}
}

#[derive(Debug)]
struct CompleteRange {
	payload: Option<response::Room>,
	config: Option<DeliveredConfig>,
	receipts: Option<Raw<SyncReceiptEvent>>,
	account_data: Vec<Raw<AnyRoomAccountDataEvent>>,
}

#[derive(Default)]
pub(super) struct Results {
	ranges: BTreeMap<OwnedRoomId, CompleteRange>,
}

#[implement(Results)]
pub(super) fn room_updates(&mut self) -> impl Iterator<Item = RoomUpdate<'_>> {
	self.ranges
		.iter_mut()
		.map(|(room_id, range)| (room_id.as_ref(), range.config.take()))
}

#[implement(Results)]
pub(super) fn payload(&self, room_id: &RoomId) -> Option<&response::Room> {
	self.ranges
		.get(room_id)
		.and_then(|range| range.payload.as_ref())
}

#[implement(Results)]
pub(super) fn into_payloads(self) -> BTreeMap<OwnedRoomId, response::Room> {
	self.ranges
		.into_iter()
		.filter_map(|(room_id, range)| range.payload.map(|payload| (room_id, payload)))
		.collect()
}

#[implement(Results)]
pub(super) fn take_receipts(&mut self, room_id: &RoomId) -> Option<Raw<SyncReceiptEvent>> {
	self.ranges
		.get_mut(room_id)
		.and_then(|range| range.receipts.take())
}

#[implement(Results)]
pub(super) fn take_account_data(
	&mut self,
	room_id: &RoomId,
) -> Option<Vec<Raw<AnyRoomAccountDataEvent>>> {
	self.ranges
		.get_mut(room_id)
		.map(|range| take(&mut range.account_data))
		.filter(|events| !events.is_empty())
}

#[tracing::instrument(
	name = "ranges",
	level = "debug",
	skip_all,
	fields(
		next_batch = conn.next_batch,
		window = window.len(),
	),
)]
pub(super) async fn collect(
	sync_info: SyncInfo<'_>,
	conn: &Connection,
	window: &Window,
) -> Results {
	let ignored = OnceCell::new();
	let empty_room = Room::default();
	let ranges = window
		.iter()
		.stream()
		.broad_filter_map(async |(room_id, window_room)| {
			let room = conn.rooms.get(room_id).unwrap_or(&empty_room);

			let room_details = merged_room_details(conn, &window_room.lists, room_id);

			match collect_room(sync_info, conn, window_room, room, room_details, &ignored).await {
				| Ok(range) => Some((room_id.clone(), range)),
				| Err(Failure { domain, error }) => {
					error!(
						%room_id,
						?domain,
						roomsince = room.roomsince,
						next_batch = conn.next_batch,
						%error,
						"sliding sync range failed"
					);
					None
				},
			}
		})
		.collect()
		.await;

	Results { ranges }
}

async fn collect_room(
	sync_info: SyncInfo<'_>,
	conn: &Connection,
	window_room: &WindowRoom,
	room: &Room,
	room_details: RoomDetails,
	ignored: &OnceCell<Option<IgnoredUserListEvent>>,
) -> Result<CompleteRange, Failure> {
	let room_id = &window_room.room_id;
	let (config_hash, required_state) = room_config(&room_details);
	let config_changed = room.config_hash != config_hash;
	let payload_is_fresh = window_room.payload_is_fresh(room.roomsince) || config_changed;

	let delivery = payload_is_fresh
		.then_async(|| {
			handle_room(sync_info, conn, window_room, room, config_changed, room_details)
		})
		.map(Option::transpose)
		.map_err(Failure::from);

	if window_room.forgotten {
		let (payload, state_read) = delivery.map_ok(Option::unzip).await?;
		let (shortstatehash, unread_state) = state_read.unwrap_or_default();
		let shortstatehash = shortstatehash.unwrap_or(room.shortstatehash);

		return assemble(
			payload,
			empty(),
			PrivateReadEvents::new(),
			Vec::new(),
			(config_hash, Default::default(), shortstatehash, unread_state),
		);
	}

	let public_receipts = public_receipts(sync_info, conn, room_id, room.roomsince, ignored)
		.map_err(|error| Failure::new(Domain::PublicReceipt, error));

	let private_receipts = private_receipts(sync_info, conn, room_id, room.roomsince)
		.map_err(|error| Failure::new(Domain::PrivateRead, error));

	let account_data = room_account_data(sync_info, conn, room_id, room.roomsince)
		.map_err(|error| Failure::new(Domain::RoomAccountData, error));

	let (delivery, public_receipts, private_receipts, account_data) =
		try_join4(delivery, public_receipts, private_receipts, account_data).await?;

	let required_state = membership_allows_required_state(window_room.membership.as_ref())
		.then_some(required_state)
		.unwrap_or_default();

	// Without freshly read state, the range keeps the snapshot recorded at its last delivery.
	let (payload, state_read) = delivery.unzip();
	let (shortstatehash, unread_state) = state_read.unwrap_or_default();
	let shortstatehash = shortstatehash.unwrap_or(room.shortstatehash);

	assemble(
		payload,
		public_receipts,
		private_receipts,
		account_data,
		(config_hash, required_state, shortstatehash, unread_state),
	)
}

async fn public_receipts(
	SyncInfo { services, sender_user, .. }: SyncInfo<'_>,
	conn: &Connection,
	room_id: &RoomId,
	roomsince: u64,
	ignored: &OnceCell<Option<IgnoredUserListEvent>>,
) -> Result<impl Iterator<Item = Raw<AnySyncEphemeralRoomEvent>>> {
	let mut receipts: Vec<(OwnedUserId, Raw<AnySyncEphemeralRoomEvent>)> = services
		.read_receipt
		.readreceipts_since_fallible(room_id, roomsince, Some(conn.next_batch))
		.map_ok(|(user_id, _ts, event)| (user_id.to_owned(), event))
		.try_collect()
		.await?;

	if !receipts.is_empty() {
		let ignored = ignored
			.get_or_try_init(async || {
				services
					.account_data
					.get_global(sender_user, GlobalAccountDataEventType::IgnoredUserList)
					.await
					.optional()
			})
			.await?;

		if let Some(ignored) = ignored {
			receipts.retain(|(user_id, _)| {
				!ignored
					.content
					.ignored_users
					.contains_key(user_id)
			});
		}
	}

	Ok(receipts.into_iter().map(at!(1)))
}

async fn private_receipts(
	SyncInfo { services, sender_user, .. }: SyncInfo<'_>,
	conn: &Connection,
	room_id: &RoomId,
	roomsince: u64,
) -> Result<PrivateReadEvents> {
	let update = services
		.read_receipt
		.last_privateread_update_fallible(sender_user, room_id)
		.await?;

	match update {
		| _ if update <= roomsince => Ok(PrivateReadEvents::new()),
		| _ if update > conn.next_batch =>
			Err(err!(Database("Private read advanced beyond the bounded sync range."))),
		| _ =>
			services
				.read_receipt
				.private_read_get_fallible(room_id, sender_user, update)
				.await,
	}
}

async fn room_account_data(
	SyncInfo { services, sender_user, .. }: SyncInfo<'_>,
	conn: &Connection,
	room_id: &RoomId,
	roomsince: u64,
) -> Result<Vec<Raw<AnyRoomAccountDataEvent>>> {
	services
		.account_data
		.changes_since_fallible(Some(room_id), sender_user, roomsince, Some(conn.next_batch))
		.ready_try_filter_map(|event| Ok(extract_variant!(event, AnyRawAccountDataEvent::Room)))
		.ready_try_filter(move |event| roomsince != 0 || !is_empty_account_data_event(event))
		.try_collect()
		.await
}

fn assemble<PublicReceipts>(
	payload: Option<response::Room>,
	public_receipts: PublicReceipts,
	private_receipts: PrivateReadEvents,
	account_data: Vec<Raw<AnyRoomAccountDataEvent>>,
	config: DeliveredConfig,
) -> Result<CompleteRange, Failure>
where
	PublicReceipts: Iterator<Item = Raw<AnySyncEphemeralRoomEvent>>,
{
	let mut receipts = public_receipts.chain(private_receipts).peekable();
	let receipts = receipts
		.peek()
		.is_some()
		.then(|| pack_receipts_fallible(receipts))
		.transpose()
		.map_err(|error| Failure::new(Domain::ReceiptSerialization, error))?;

	let config = payload.as_ref().map(|_| config);

	Ok(CompleteRange { payload, config, receipts, account_data })
}

#[cfg(test)]
mod tests {
	use std::iter::once;

	use ruma::{api::client::sync::sync_events::v5::response::Room as ResponseRoom, room_id};
	use serde_json::{json, value::to_raw_value};

	use super::*;

	#[test]
	fn malformed_receipt_withholds_the_complete_range() {
		let room_id = room_id!("!receipt:example.com");
		let malformed = Raw::from_json(
			to_raw_value(&json!({"content": 5})).expect("test JSON should serialize"),
		);

		let range = assemble(
			Some(ResponseRoom::default()),
			vec![malformed].into_iter(),
			PrivateReadEvents::new(),
			Vec::new(),
			(7, once(11).collect(), 13, Default::default()),
		);

		let error = range.expect_err("malformed receipt must fail the complete range");

		assert_eq!(error.domain, Domain::ReceiptSerialization);
		assert!(
			!publish(room_id, Err(error))
				.ranges
				.contains_key(room_id)
		);
	}

	#[test]
	fn extension_only_range_commits_without_a_room_payload() {
		let room_id = room_id!("!extension-only:example.com");
		let range = assemble(
			None,
			Vec::new().into_iter(),
			PrivateReadEvents::new(),
			Vec::new(),
			(7, once(11).collect(), 13, Default::default()),
		);

		let mut range = publish(room_id, range); // room_updates consumes the configuration.

		assert_eq!(range.room_updates().collect::<Vec<_>>(), [(room_id, None)]);
		assert!(range.into_payloads().is_empty());
	}

	#[test]
	fn extension_outputs_are_taken_once_without_removing_the_complete_range() {
		let room_id = room_id!("!extension-output:example.com");
		let receipt = Raw::from_json(
			to_raw_value(&json!({"content": {}})).expect("test receipt should serialize"),
		);

		let account_data = Raw::from_json(
			to_raw_value(&json!({"type": "m.tag", "content": {"tags": {}}}))
				.expect("test account data should serialize"),
		);

		let range = CompleteRange {
			payload: None,
			config: None,
			receipts: Some(receipt),
			account_data: vec![account_data],
		};

		let ranges = [(room_id.to_owned(), range)].into();
		let mut results = Results { ranges };

		assert!(results.take_receipts(room_id).is_some());
		assert!(results.take_receipts(room_id).is_none());

		let count = results
			.take_account_data(room_id)
			.map(|events| events.len());

		assert_eq!(Some(1), count);
		assert!(results.take_account_data(room_id).is_none());
		assert!(results.ranges.contains_key(room_id));
	}

	fn publish(room_id: &RoomId, range: Result<CompleteRange, Failure>) -> Results {
		let ranges = range
			.ok()
			.map(|range| (room_id.to_owned(), range))
			.into_iter()
			.collect();

		Results { ranges }
	}
}