sync.rs

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

use axum::extract::State;
use conduit::{
	debug, error,
	utils::math::{ruma_from_u64, ruma_from_usize, usize_from_ruma, usize_from_u64_truncated},
	warn, Err, PduCount,
};
use ruma::{
	api::client::{
		error::ErrorKind,
		filter::{FilterDefinition, LazyLoadOptions},
		sync::sync_events::{
			self,
			v3::{
				Ephemeral, Filter, GlobalAccountData, InviteState, InvitedRoom, JoinedRoom, LeftRoom, Presence,
				RoomAccountData, RoomSummary, Rooms, State as RoomState, Timeline, ToDevice,
			},
			v4::{SlidingOp, SlidingSyncRoomHero},
			DeviceLists, UnreadNotificationsCount,
		},
		uiaa::UiaaResponse,
	},
	events::{
		presence::PresenceEvent,
		room::member::{MembershipState, RoomMemberEventContent},
		AnyRawAccountDataEvent, StateEventType, TimelineEventType,
	},
	room::RoomType,
	serde::Raw,
	state_res::Event,
	uint, DeviceId, EventId, MilliSecondsSinceUnixEpoch, OwnedRoomId, OwnedUserId, RoomId, UInt, UserId,
};
use tracing::{Instrument as _, Span};

use crate::{
	service::{pdu::EventHash, Services},
	utils, Error, PduEvent, Result, Ruma, RumaResponse,
};

const SINGLE_CONNECTION_SYNC: &str = "single_connection_sync";
const DEFAULT_BUMP_TYPES: &[TimelineEventType] = &[
	TimelineEventType::Message,
	TimelineEventType::Encrypted,
	TimelineEventType::Sticker,
	TimelineEventType::CallInvite,
	TimelineEventType::PollStart,
	TimelineEventType::Beacon,
];

macro_rules! extract_variant {
	($e:expr, $variant:path) => {
		match $e {
			$variant(value) => Some(value),
			_ => None,
		}
	};
}

/// # `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)
pub(crate) async fn sync_events_route(
	State(services): State<crate::State>, 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;

	// 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(),
	};

	// some clients, at least element, seem to require knowledge of redundant
	// members for "inline" profiles on the timeline to work properly
	let (lazy_load_enabled, lazy_load_send_redundant) = match filter.room.state.lazy_load_options {
		LazyLoadOptions::Enabled {
			include_redundant_members,
		} => (true, include_redundant_members),
		LazyLoadOptions::Disabled => (false, cfg!(feature = "element_hacks")),
	};

	let full_state = body.full_state;

	let mut joined_rooms = BTreeMap::new();
	let since = body
		.since
		.as_ref()
		.and_then(|string| string.parse().ok())
		.unwrap_or(0);
	let sincecount = PduCount::Normal(since);

	let mut presence_updates = HashMap::new();
	let mut left_encrypted_users = HashSet::new(); // Users that have left any encrypted rooms the sender was in
	let mut device_list_updates = HashSet::new();
	let mut device_list_left = HashSet::new();

	// Look for device list updates of this account
	device_list_updates.extend(
		services
			.users
			.keys_changed(sender_user.as_ref(), since, None)
			.filter_map(Result::ok),
	);

	if services.globals.allow_local_presence() {
		process_presence_updates(&services, &mut presence_updates, since, &sender_user).await?;
	}

	let all_joined_rooms = services
		.rooms
		.state_cache
		.rooms_joined(&sender_user)
		.collect::<Vec<_>>();

	// Coalesce database writes for the remainder of this scope.
	let _cork = services.db.cork_and_flush();

	for room_id in all_joined_rooms {
		let room_id = room_id?;
		if let Ok(joined_room) = load_joined_room(
			&services,
			&sender_user,
			&sender_device,
			&room_id,
			since,
			sincecount,
			next_batch,
			next_batchcount,
			lazy_load_enabled,
			lazy_load_send_redundant,
			full_state,
			&mut device_list_updates,
			&mut left_encrypted_users,
		)
		.await
		{
			if !joined_room.is_empty() {
				joined_rooms.insert(room_id.clone(), joined_room);
			}
		}
	}