sync.rs

use crate::{
    service::rooms::timeline::PduCount, services, Error, PduEvent, Result, Ruma, RumaResponse,
};
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::{
        room::member::{MembershipState, RoomMemberEventContent},
        StateEventType, TimelineEventType,
    },
    serde::Raw,
    uint, DeviceId, OwnedDeviceId, OwnedUserId, RoomId, UInt, UserId,
};
use std::{
    collections::{hash_map::Entry, BTreeMap, BTreeSet, HashMap, HashSet},
    sync::Arc,
    time::Duration,
};
use tokio::sync::watch::Sender;
use tracing::error;

/// # `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()
        .unwrap()
        .entry((sender_user.clone(), sender_device.clone()))
    {
        Entry::Vacant(v) => {
            let (tx, rx) = tokio::sync::watch::channel(None);

            v.insert((body.since.to_owned(), 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()));

                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()
                .unwrap()
                .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(_) => {}
            }
        }
    }

    let _ = 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> {
    // TODO: match body.set_presence {
    services().rooms.edus.presence.ping_presence(&sender_user)?;

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

    let full_state = body.full_state;