sync.rs

use std::{
	cmp::Ordering,
	collections::{hash_map::Entry, BTreeMap, BTreeSet, HashMap, HashSet},
	sync::Arc,
	time::Duration,
};

use ruma::{
	api::client::{
		filter::{FilterDefinition, LazyLoadOptions},
		sync::sync_events::{
			self,
			v3::{
				Ephemeral, Filter, GlobalAccountData, InviteState, InvitedRoom, JoinedRoom, LeftRoom, Presence,
				RoomAccountData, RoomSummary, Rooms, State, Timeline, ToDevice,
			},
			v4::SlidingOp,
			DeviceLists, UnreadNotificationsCount,
		},
		uiaa::UiaaResponse,
	},
	events::{
		presence::PresenceEvent,
		room::member::{MembershipState, RoomMemberEventContent},
		StateEventType, TimelineEventType,
	},
	serde::Raw,
	uint, DeviceId, OwnedDeviceId, OwnedUserId, RoomId, UInt, UserId,
};
use tokio::sync::watch::Sender;
use tracing::{debug, error};

use crate::{service::rooms::timeline::PduCount, services, Error, PduEvent, Result, Ruma, RumaResponse};

/// # `GET /_matrix/client/r0/sync`
///
/// Synchronize the client's state with the latest state on the server.
///
/// - This endpoint takes a `since` parameter which should be the `next_batch`
///   value from a
/// previous request for incremental syncs.
///
/// Calling this endpoint without a `since` parameter returns:
/// - Some of the most recent events of each timeline
/// - Notification counts for each room
/// - Joined and invited member counts, heroes
/// - All state events
///
/// Calling this endpoint with a `since` parameter from a previous `next_batch`
/// returns: For joined rooms:
/// - Some of the most recent events of each timeline that happened after since
/// - If user joined the room after since: All state events (unless lazy loading
///   is activated) and
/// all device list updates in that room
/// - If the user was already in the room: A list of all events that are in the
///   state now, but were
/// not in the state at `since`
/// - If the state we send contains a member event: Joined and invited member
///   counts, heroes
/// - Device list updates that happened after `since`
/// - If there are events in the timeline we send or the user send updated his
///   read mark: Notification counts
/// - EDUs that are active now (read receipts, typing updates, presence)
/// - TODO: Allow multiple sync streams to support Pantalaimon
///
/// For invited rooms:
/// - If the user was invited after `since`: A subset of the state of the room
///   at the point of the invite
///
/// For left rooms:
/// - If the user left after `since`: `prev_batch` token, empty state (TODO:
///   subset of the state at the point of the leave)
///
/// - Sync is handled in an async task, multiple requests from the same device
///   with the same
/// `since` will be cached
pub async fn sync_events_route(
	body: Ruma<sync_events::v3::Request>,
) -> Result<sync_events::v3::Response, RumaResponse<UiaaResponse>> {
	let sender_user = body.sender_user.expect("user is authenticated");
	let sender_device = body.sender_device.expect("user is authenticated");
	let body = body.body;

	let mut rx = match services()
		.globals
		.sync_receivers
		.write()
		.await
		.entry((sender_user.clone(), sender_device.clone()))
	{
		Entry::Vacant(v) => {
			let (tx, rx) = tokio::sync::watch::channel(None);

			v.insert((body.since.clone(), rx.clone()));

			tokio::spawn(sync_helper_wrapper(sender_user.clone(), sender_device.clone(), body, tx));

			rx
		},
		Entry::Occupied(mut o) => {
			if o.get().0 != body.since {
				let (tx, rx) = tokio::sync::watch::channel(None);

				o.insert((body.since.clone(), rx.clone()));

				debug!("Sync started for {sender_user}");

				tokio::spawn(sync_helper_wrapper(sender_user.clone(), sender_device.clone(), body, tx));

				rx
			} else {
				o.get().1.clone()
			}
		},
	};

	let we_have_to_wait = rx.borrow().is_none();
	if we_have_to_wait {
		if let Err(e) = rx.changed().await {
			error!("Error waiting for sync: {}", e);
		}
	}

	let result = match rx
		.borrow()
		.as_ref()
		.expect("When sync channel changes it's always set to some")
	{
		Ok(response) => Ok(response.clone()),
		Err(error) => Err(error.to_response()),
	};

	result
}

async fn sync_helper_wrapper(
	sender_user: OwnedUserId, sender_device: OwnedDeviceId, body: sync_events::v3::Request,
	tx: Sender<Option<Result<sync_events::v3::Response>>>,
) {
	let since = body.since.clone();

	let r = sync_helper(sender_user.clone(), sender_device.clone(), body).await;

	if let Ok((_, caching_allowed)) = r {
		if !caching_allowed {
			match services()
				.globals
				.sync_receivers
				.write()
				.await
				.entry((sender_user, sender_device))
			{
				Entry::Occupied(o) => {
					// Only remove if the device didn't start a different /sync already
					if o.get().0 == since {
						o.remove();
					}
				},
				Entry::Vacant(_) => {},
			}
		}
	}

	_ = tx.send(Some(r.map(|(r, _)| r)));
}

async fn sync_helper(
	sender_user: OwnedUserId,
	sender_device: OwnedDeviceId,
	body: sync_events::v3::Request,
	// bool = caching allowed
) -> Result<(sync_events::v3::Response, bool), Error> {
	// Presence update
	if services().globals.allow_local_presence() {
		services()
			.presence
			.ping_presence(&sender_user, body.set_presence)?;
	}

	// Setup watchers, so if there's no response, we can wait for them
	let watcher = services().globals.watch(&sender_user, &sender_device);

	let next_batch = services().globals.current_count()?;
	let next_batchcount = PduCount::Normal(next_batch);
	let next_batch_string = next_batch.to_string();

	// Load filter
	let filter = match body.filter {
		None => FilterDefinition::default(),
		Some(Filter::FilterDefinition(filter)) => filter,
		Some(Filter::FilterId(filter_id)) => services()
			.users
			.get_filter(&sender_user, &filter_id)?
			.unwrap_or_default(),
	};

	let (lazy_load_enabled, lazy_load_send_redundant) = match filter.room.state.lazy_load_options {
		LazyLoadOptions::Enabled {
			include_redundant_members: redundant,
		} => (true, redundant),