matrix_sdk/latest_events/

mod.rs

// Copyright 2025 The Matrix.org Foundation C.I.C.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! The Latest Events API provides a lazy, reactive and efficient way to compute
//! the latest event for a room or a thread.
//!
//! The latest event represents the last displayable and relevant event a room
//! or a thread has received. It is usually displayed in a _summary_, e.g.
//! below the room title in a room list.
//!
//! The entry point is [`LatestEvents`]. It is preferable to get a reference to
//! it from [`Client::latest_events`][crate::Client::latest_events], which
//! already plugs everything to build it. [`LatestEvents`] is using the
//! [`EventCache`] and the [`SendQueue`] to respectively get known remote events
//! (i.e. synced from the server), or local events (i.e. ones being sent).
//!
//! ## Laziness
//!
//! [`LatestEvents`] is lazy, it means that, despite [`LatestEvents`] is
//! listening to all [`EventCache`] or [`SendQueue`] updates, it will only do
//! something if one is expected to get the latest event for a particular room
//! or a particular thread. Concretely, it means that until
//! [`LatestEvents::listen_to_room`] is called for a particular room, no latest
//! event will ever be computed for that room (and similarly with
//! [`LatestEvents::listen_to_thread`]).
//!
//! If one is no longer interested to get the latest event for a particular room
//! or thread, the [`LatestEvents::forget_room`] and
//! [`LatestEvents::forget_thread`] methods must be used.
//!
//! ## Reactive
//!
//! [`LatestEvents`] is designed to be reactive. Using
//! [`LatestEvents::listen_and_subscribe_to_room`] will provide a
//! [`Subscriber`], which brings all the tooling to get the current value or the
//! future values with a stream.

mod error;
mod latest_event;
mod room_latest_events;

use std::{
    collections::HashMap,
    ops::{ControlFlow, DerefMut, Not},
    sync::Arc,
};

pub use error::LatestEventsError;
use eyeball::{AsyncLock, Subscriber};
pub(crate) use latest_event::filter_timeline_event;
use latest_event::{LatestEvent, With};
pub use latest_event::{LatestEventValue, LocalLatestEventValue, RemoteLatestEventValue};
use matrix_sdk_base::{RoomInfoNotableUpdate, RoomInfoNotableUpdateReasons, timer};
use matrix_sdk_common::executor::{AbortOnDrop, JoinHandleExt as _, spawn};
use room_latest_events::{RoomLatestEvents, RoomLatestEventsWriteGuard};
use ruma::{EventId, OwnedRoomId, RoomId};
use tokio::{
    select,
    sync::{RwLock, RwLockReadGuard, RwLockWriteGuard, broadcast, mpsc},
};
use tracing::{info, warn};

use crate::{
    client::WeakClient,
    event_cache::{EventCache, RoomEventCacheGenericUpdate},
    room::WeakRoom,
    send_queue::{RoomSendQueueUpdate, SendQueue, SendQueueUpdate},
};

/// The entry point to fetch the [`LatestEventValue`] for rooms or threads.
#[derive(Clone, Debug)]
pub struct LatestEvents {
    state: Arc<LatestEventsState>,
}

/// The state of [`LatestEvents`].
#[derive(Debug)]
struct LatestEventsState {
    /// All the registered rooms, i.e. rooms the latest events are computed for.
    registered_rooms: Arc<RegisteredRooms>,

    /// The task handle of the [`listen_to_updates_task`].
    _listen_task_handle: AbortOnDrop<()>,

    /// The task handle of the [`compute_latest_events_task`].
    _computation_task_handle: AbortOnDrop<()>,
}

impl LatestEvents {
    /// Create a new [`LatestEvents`].
    pub(crate) fn new(
        weak_client: WeakClient,
        event_cache: EventCache,
        send_queue: SendQueue,
        room_info_updates: broadcast::Receiver<RoomInfoNotableUpdate>,
    ) -> Self {
        let (latest_event_queue_sender, latest_event_queue_receiver) = mpsc::unbounded_channel();

        let registered_rooms =
            Arc::new(RegisteredRooms::new(weak_client, &event_cache, &latest_event_queue_sender));

        // The task listening to the event cache, the send queue, and the room
        // infos updates.
        //
        // Note: we spawn the send queue observer here, instead of holding onto
        // the `SendQueue` in the `listen_to_updates_task`, which would
        // lead to a `Client` leak (since it's a field of
        // the `SendQueue`).
        let send_queue_generic_updates_subscriber = send_queue.subscribe();
        let listen_task_handle = spawn(listen_to_updates_task(
            registered_rooms.clone(),
            event_cache,
            send_queue_generic_updates_subscriber,
            room_info_updates,
            latest_event_queue_sender,
        ))
        .abort_on_drop();

        // The task computing the new latest events.
        let computation_task_handle = spawn(compute_latest_events_task(
            registered_rooms.clone(),
            latest_event_queue_receiver,
        ))
        .abort_on_drop();

        Self {
            state: Arc::new(LatestEventsState {
                registered_rooms,
                _listen_task_handle: listen_task_handle,
                _computation_task_handle: computation_task_handle,
            }),
        }
    }

    /// Start listening to updates (if not already) for a particular room.
    ///
    /// It returns `true` if the room exists, `false` otherwise.
    pub async fn listen_to_room(&self, room_id: &RoomId) -> Result<bool, LatestEventsError> {
        Ok(self.state.registered_rooms.for_room(room_id).await?.is_some())
    }

    /// Check whether the system listens to a particular room.
    ///
    /// Note: It's a test only method.
    #[cfg(test)]
    pub async fn is_listening_to_room(&self, room_id: &RoomId) -> bool {
        self.state.registered_rooms.rooms.read().await.contains_key(room_id)
    }

    /// Start listening to updates (if not already) for a particular room, and
    /// return a [`Subscriber`] to get the current and future
    /// [`LatestEventValue`]s.
    ///
    /// It returns `Some` if the room exists, `None` otherwise.
    pub async fn listen_and_subscribe_to_room(
        &self,
        room_id: &RoomId,
    ) -> Result<Option<Subscriber<LatestEventValue, AsyncLock>>, LatestEventsError> {
        let Some(room_latest_events) = self.state.registered_rooms.for_room(room_id).await? else {
            return Ok(None);
        };

        let room_latest_events = room_latest_events.read().await;
        let latest_event = room_latest_events.for_room();

        Ok(Some(latest_event.subscribe().await))
    }

    /// Start listening to updates (if not already) for a particular room and a
    /// particular thread in this room.
    ///
    /// It returns `true` if the room and the thread exists, `false` otherwise.
    pub async fn listen_to_thread(
        &self,
        room_id: &RoomId,
        thread_id: &EventId,
    ) -> Result<bool, LatestEventsError> {
        Ok(self.state.registered_rooms.for_thread(room_id, thread_id).await?.is_some())
    }

    /// Start listening to updates (if not already) for a particular room and a
    /// particular thread in this room, and return a [`Subscriber`] to get the
    /// current and future [`LatestEventValue`]s.
    ///
    /// It returns `Some` if the room and the thread exists, `None` otherwise.
    pub async fn listen_and_subscribe_to_thread(
        &self,
        room_id: &RoomId,
        thread_id: &EventId,
    ) -> Result<Option<Subscriber<LatestEventValue, AsyncLock>>, LatestEventsError> {
        let Some(room_latest_events) =
            self.state.registered_rooms.for_thread(room_id, thread_id).await?
        else {
            return Ok(None);
        };

        let room_latest_events = room_latest_events.read().await;
        let latest_event = room_latest_events
            .for_thread(thread_id)
            .expect("The `LatestEvent` for the thread must have been created");

        Ok(Some(latest_event.subscribe().await))
    }

    /// Forget a room.
    ///
    /// It means that [`LatestEvents`] will stop listening to updates for the
    /// `LatestEvent`s of the room and all its threads.
    ///
    /// If [`LatestEvents`] is not listening for `room_id`, nothing happens.
    pub async fn forget_room(&self, room_id: &RoomId) {
        self.state.registered_rooms.forget_room(room_id).await;
    }

    /// Forget a thread.
    ///
    /// It means that [`LatestEvents`] will stop listening to updates for the
    /// `LatestEvent` of the thread.
    ///
    /// If [`LatestEvents`] is not listening for `room_id` or `thread_id`,
    /// nothing happens.
    pub async fn forget_thread(&self, room_id: &RoomId, thread_id: &EventId) {
        self.state.registered_rooms.forget_thread(room_id, thread_id).await;
    }
}

#[derive(Debug)]
struct RegisteredRooms {
    /// All the registered [`RoomLatestEvents`].
    rooms: RwLock<HashMap<OwnedRoomId, RoomLatestEvents>>,

    /// The (weak) client.
    weak_client: WeakClient,

    /// The event cache.
    event_cache: EventCache,

    /// The sender part of the channel used by [`compute_latest_events_task`].
    ///
    /// This is used to _trigger_ a computation of a `LatestEventValue` if the
    /// restored value is `None`.
    latest_event_queue_sender: mpsc::UnboundedSender<LatestEventQueueUpdate>,
}

impl RegisteredRooms {
    fn new(
        weak_client: WeakClient,
        event_cache: &EventCache,
        latest_event_queue_sender: &mpsc::UnboundedSender<LatestEventQueueUpdate>,
    ) -> Self {
        Self {
            rooms: RwLock::new(HashMap::default()),
            weak_client,
            event_cache: event_cache.clone(),
            latest_event_queue_sender: latest_event_queue_sender.clone(),
        }
    }

    /// Get a read lock guard to a [`RoomLatestEvents`] given a room ID and an
    /// optional thread ID.
    ///
    /// The [`RoomLatestEvents`], and the associated [`LatestEvent`], will be
    /// created if missing. It means that write lock is taken if necessary, but
    /// it's always downgraded to a read lock at the end.
    async fn room_latest_event(
        &self,
        room_id: &RoomId,
        thread_id: Option<&EventId>,
    ) -> Result<Option<RwLockReadGuard<'_, RoomLatestEvents>>, LatestEventsError> {
        fn create_and_insert_room_latest_events(
            room_id: &RoomId,
            rooms: &mut HashMap<OwnedRoomId, RoomLatestEvents>,
            weak_client: &WeakClient,
            event_cache: &EventCache,
            latest_event_queue_sender: &mpsc::UnboundedSender<LatestEventQueueUpdate>,
        ) {
            let (room_latest_events, is_latest_event_value_none) =
                With::unzip(RoomLatestEvents::new(
                    WeakRoom::new(weak_client.clone(), room_id.to_owned()),
                    event_cache,
                ));

            // Insert the new `RoomLatestEvents`.
            rooms.insert(room_id.to_owned(), room_latest_events);

            // If the `LatestEventValue` restored by `RoomLatestEvents` is of
            // kind `None`, let's try to re-compute it without waiting on the
            // Event Cache (so the sync usually) or the Send Queue. Maybe the
            // system has migrated to a new version and the `LatestEventValue`
            // has been erased, while it is still possible to compute a correct
            // value.
            if is_latest_event_value_none {
                let _ = latest_event_queue_sender
                    .send(LatestEventQueueUpdate::EventCache { room_id: room_id.to_owned() });
            }
        }

        Ok(match thread_id {
            // Get the room latest event with the aim of fetching the latest
            // event for a particular thread.
            //
            // We need to take a write lock immediately, in case the thead
            // latest event doesn't exist.
            Some(thread_id) => {
                let mut rooms = self.rooms.write().await;

                // The `RoomLatestEvents` doesn't exist. Let's create and insert
                // it.
                if rooms.contains_key(room_id).not() {
                    create_and_insert_room_latest_events(
                        room_id,
                        rooms.deref_mut(),
                        &self.weak_client,
                        &self.event_cache,
                        &self.latest_event_queue_sender,
                    );
                }

                if let Some(room_latest_event) = rooms.get(room_id) {
                    let mut room_latest_event = room_latest_event.write().await;

                    // In `RoomLatestEvents`, the `LatestEvent` for this thread
                    // doesn't exist. Let's create and insert it.
                    if room_latest_event.has_thread(thread_id).not() {
                        room_latest_event.create_and_insert_latest_event_for_thread(thread_id);
                    }
                }

                RwLockWriteGuard::try_downgrade_map(rooms, |rooms| rooms.get(room_id)).ok()
            }

            // Get the room latest event with the aim of fetching the latest
            // event for a particular room.
            None => {
                match RwLockReadGuard::try_map(self.rooms.read().await, |rooms| rooms.get(room_id))
                    .ok()
                {
                    value @ Some(_) => value,
                    None => {
                        let _timer = timer!(
                            tracing::Level::DEBUG,
                            format!("Creating `RoomLatestEvents` for {room_id:?}"),
                        );

                        let mut rooms = self.rooms.write().await;

                        if rooms.contains_key(room_id).not() {
                            create_and_insert_room_latest_events(
                                room_id,
                                rooms.deref_mut(),
                                &self.weak_client,
                                &self.event_cache,
                                &self.latest_event_queue_sender,
                            );
                        }

                        RwLockWriteGuard::try_downgrade_map(rooms, |rooms| rooms.get(room_id)).ok()
                    }
                }
            }
        })
    }

    /// Start listening to updates (if not already) for a particular room.
    ///
    /// It returns `None` if the room doesn't exist.
    pub async fn for_room(
        &self,
        room_id: &RoomId,
    ) -> Result<Option<RwLockReadGuard<'_, RoomLatestEvents>>, LatestEventsError> {
        self.room_latest_event(room_id, None).await
    }

    /// Start listening to updates (if not already) for a particular room.
    ///
    /// It returns `None` if the room or the thread doesn't exist.
    pub async fn for_thread(
        &self,
        room_id: &RoomId,
        thread_id: &EventId,
    ) -> Result<Option<RwLockReadGuard<'_, RoomLatestEvents>>, LatestEventsError> {
        self.room_latest_event(room_id, Some(thread_id)).await
    }

    /// Forget a room.
    ///
    /// It means that [`LatestEvents`] will stop listening to updates for the
    /// `LatestEvent`s of the room and all its threads.
    ///
    /// If [`LatestEvents`] is not listening for `room_id`, nothing happens.
    pub async fn forget_room(&self, room_id: &RoomId) {
        {
            let mut rooms = self.rooms.write().await;

            // Remove the whole `RoomLatestEvents`.
            rooms.remove(room_id);
        }
    }

    /// Forget a thread.
    ///
    /// It means that [`LatestEvents`] will stop listening to updates for the
    /// `LatestEvent` of the thread.
    ///
    /// If [`LatestEvents`] is not listening for `room_id` or `thread_id`,
    /// nothing happens.
    pub async fn forget_thread(&self, room_id: &RoomId, thread_id: &EventId) {
        let rooms = self.rooms.read().await;

        // If the `RoomLatestEvents`, remove the `LatestEvent` in `per_thread`.
        if let Some(room_latest_event) = rooms.get(room_id) {
            let mut room_latest_event = room_latest_event.write().await;

            // Release the lock on `self.rooms`.
            drop(rooms);

            room_latest_event.forget_thread(thread_id);
        }
    }
}

/// Represents the kind of updates the [`compute_latest_events_task`] will have
/// to deal with.
#[derive(Debug)]
enum LatestEventQueueUpdate {
    /// An update from the [`EventCache`] happened.
    EventCache {
        /// The ID of the room that has triggered the update.
        room_id: OwnedRoomId,
    },

    /// An update from the [`SendQueue`] happened.
    SendQueue {
        /// The ID of the room that has triggered the update.
        room_id: OwnedRoomId,

        /// The update itself.
        update: RoomSendQueueUpdate,
    },

    /// An update from the [`RoomInfo`] happened.
    ///
    /// [`RoomInfo`]: crate::RoomInfo
    RoomInfo {
        /// The ID of the room that has triggered the update.
        room_id: OwnedRoomId,

        /// The notable update reasons.
        reasons: RoomInfoNotableUpdateReasons,
    },
}

/// The task responsible to listen to the [`EventCache`], the [`SendQueue`] and
/// the [`RoomInfoNotableUpdate`].
///
/// When an update is received and is considered relevant, a message is sent to
/// the [`compute_latest_events_task`] to compute a new [`LatestEvent`].
///
/// When an update is considered relevant, a message is sent over the
/// `latest_event_queue_sender` channel. See [`compute_latest_events_task`].
async fn listen_to_updates_task(
    registered_rooms: Arc<RegisteredRooms>,
    event_cache: EventCache,
    mut send_queue_generic_updates_subscriber: broadcast::Receiver<SendQueueUpdate>,
    room_info_updates: broadcast::Receiver<RoomInfoNotableUpdate>,
    latest_event_queue_sender: mpsc::UnboundedSender<LatestEventQueueUpdate>,
) {
    let mut event_cache_generic_updates_subscriber =
        event_cache.subscribe_to_room_generic_updates();
    let mut room_info_updates_subscriber = room_info_updates.resubscribe();

    loop {
        if listen_to_updates(
            &registered_rooms.rooms,
            &mut event_cache_generic_updates_subscriber,
            &mut send_queue_generic_updates_subscriber,
            &mut room_info_updates_subscriber,
            &latest_event_queue_sender,
        )
        .await
        .is_break()
        {
            warn!("`listen_to_updates_task` has stopped");

            break;
        }
    }
}

/// The core of [`listen_to_updates_task`].
///
/// Having this function detached from its task is helpful for testing and for
/// state isolation.
async fn listen_to_updates(
    registered_rooms: &RwLock<HashMap<OwnedRoomId, RoomLatestEvents>>,
    event_cache_generic_updates_subscriber: &mut broadcast::Receiver<RoomEventCacheGenericUpdate>,
    send_queue_generic_updates_subscriber: &mut broadcast::Receiver<SendQueueUpdate>,
    room_info_updates_subscriber: &mut broadcast::Receiver<RoomInfoNotableUpdate>,
    latest_event_queue_sender: &mpsc::UnboundedSender<LatestEventQueueUpdate>,
) -> ControlFlow<()> {
    select! {
        room_event_cache_generic_update = event_cache_generic_updates_subscriber.recv() => {
            if let Ok(RoomEventCacheGenericUpdate { room_id }) = room_event_cache_generic_update {
                if registered_rooms.read().await.contains_key(&room_id) {
                    let _ = latest_event_queue_sender.send(LatestEventQueueUpdate::EventCache {
                        room_id
                    });
                }
            } else {
                warn!("`event_cache_generic_updates` channel has been closed");

                return ControlFlow::Break(());
            }
        }

        send_queue_generic_update = send_queue_generic_updates_subscriber.recv() => {
            if let Ok(SendQueueUpdate { room_id, update }) = send_queue_generic_update {
                if registered_rooms.read().await.contains_key(&room_id) {
                    let _ = latest_event_queue_sender.send(LatestEventQueueUpdate::SendQueue {
                        room_id,
                        update
                    });
                }
            } else {
                warn!("`send_queue_generic_updates` channel has been closed");

                return ControlFlow::Break(());
            }
        }

        room_info_update = room_info_updates_subscriber.recv() => {
            if let Ok(RoomInfoNotableUpdate { room_id, reasons }) = room_info_update {
                // Filter the update reasons we are interested by.
                if
                    // Be careful: the `LATEST_EVENT` reason alone must always
                    // be ignored! Otherwise it can create a loop.
                    reasons == RoomInfoNotableUpdateReasons::LATEST_EVENT ||

                    // We are interested by `MEMBERSHIP` so that it captures
                    // when the user is invited, is joining, is knocking, or is
                    // leaving a room.
                    !reasons.contains(RoomInfoNotableUpdateReasons::MEMBERSHIP)
                {
                    return ControlFlow::Continue(())
                }

                if registered_rooms.read().await.contains_key(&room_id) {
                    let _ = latest_event_queue_sender.send(LatestEventQueueUpdate::RoomInfo {
                        room_id,
                        reasons,
                    });
                }
            } else {
                warn!("`room_info_updates` channel has been closed");

                return ControlFlow::Break(());
            }
        }
    }

    ControlFlow::Continue(())
}

/// The task responsible to compute new [`LatestEvent`] for a particular room or
/// thread.
///
/// The messages are coming from [`listen_to_updates_task`].
async fn compute_latest_events_task(
    registered_rooms: Arc<RegisteredRooms>,
    mut latest_event_queue_receiver: mpsc::UnboundedReceiver<LatestEventQueueUpdate>,
) {
    const BUFFER_SIZE: usize = 16;

    let mut buffer = Vec::with_capacity(BUFFER_SIZE);

    while latest_event_queue_receiver.recv_many(&mut buffer, BUFFER_SIZE).await > 0 {
        compute_latest_events(&registered_rooms, &buffer).await;
        buffer.clear();
    }

    warn!("`compute_latest_events_task` has stopped");
}

async fn compute_latest_events(
    registered_rooms: &RegisteredRooms,
    latest_event_queue_updates: &[LatestEventQueueUpdate],
) {
    async fn room_latest_events_write_guard(
        registered_rooms: &RegisteredRooms,
        room_id: &OwnedRoomId,
    ) -> ControlFlow<RoomLatestEventsWriteGuard, ()> {
        let rooms = registered_rooms.rooms.read().await;

        if let Some(room_latest_events) = rooms.get(room_id) {
            let room_latest_events = room_latest_events.write().await;

            // Release the lock on `registered_rooms`. It is possible because
            // `room_latest_events` is an owned lock guard.
            drop(rooms);

            ControlFlow::Break(room_latest_events)
        } else {
            info!(?room_id, "Failed to find the room");

            ControlFlow::Continue(())
        }
    }

    for latest_event_queue_update in latest_event_queue_updates {
        match latest_event_queue_update {
            LatestEventQueueUpdate::EventCache { room_id } => {
                let ControlFlow::Break(mut room_latest_events) =
                    room_latest_events_write_guard(registered_rooms, room_id).await
                else {
                    continue;
                };

                room_latest_events.update_with_event_cache().await;
            }

            LatestEventQueueUpdate::SendQueue { room_id, update } => {
                let ControlFlow::Break(mut room_latest_events) =
                    room_latest_events_write_guard(registered_rooms, room_id).await
                else {
                    continue;
                };

                room_latest_events.update_with_send_queue(update).await;
            }

            LatestEventQueueUpdate::RoomInfo { room_id, reasons } => {
                let ControlFlow::Break(mut room_latest_events) =
                    room_latest_events_write_guard(registered_rooms, room_id).await
                else {
                    continue;
                };

                room_latest_events.update_with_room_info(*reasons).await;
            }
        }
    }
}

#[cfg(test)]
fn local_room_message(body: &str) -> LocalLatestEventValue {
    use matrix_sdk_base::store::SerializableEventContent;
    use ruma::{
        MilliSecondsSinceUnixEpoch,
        events::{AnyMessageLikeEventContent, room::message::RoomMessageEventContent},
    };

    LocalLatestEventValue {
        timestamp: MilliSecondsSinceUnixEpoch::now(),
        content: SerializableEventContent::new(&AnyMessageLikeEventContent::RoomMessage(
            RoomMessageEventContent::text_plain(body),
        ))
        .unwrap(),
    }
}

#[cfg(all(test, not(target_family = "wasm")))]
mod tests {
    use std::{collections::HashMap, ops::Not, time::Duration};

    use assert_matches::assert_matches;
    use matrix_sdk_base::{
        RoomState,
        cross_process_lock::CrossProcessLockConfig,
        deserialized_responses::TimelineEventKind,
        event_cache::store::{EventCacheStore, MemoryStore},
        linked_chunk::{ChunkIdentifier, LinkedChunkId, Position, Update},
        store::StoreConfig,
    };
    use matrix_sdk_test::{
        InvitedRoomBuilder, JoinedRoomBuilder, async_test, event_factory::EventFactory,
    };
    use ruma::{
        MilliSecondsSinceUnixEpoch, OwnedTransactionId, event_id,
        events::{
            AnySyncMessageLikeEvent, AnySyncStateEvent, AnySyncTimelineEvent, SyncMessageLikeEvent,
            room::member::{MembershipState, SyncRoomMemberEvent},
        },
        owned_event_id, owned_room_id, room_id, user_id,
    };
    use stream_assert::assert_pending;
    use tokio::{
        task::yield_now,
        time::{sleep, timeout},
    };

    use super::{
        LatestEventValue, RegisteredRooms, RemoteLatestEventValue, RoomEventCacheGenericUpdate,
        RoomInfoNotableUpdate, RoomInfoNotableUpdateReasons, RoomLatestEvents, RoomSendQueueUpdate,
        RwLock, SendQueueUpdate, WeakRoom, With, broadcast, listen_to_updates, mpsc,
    };
    use crate::{
        latest_events::{LatestEventQueueUpdate, local_room_message},
        test_utils::{client::MockClientBuilder, mocks::MatrixMockServer},
    };

    #[async_test]
    async fn test_latest_events_are_lazy() {
        let room_id_0 = room_id!("!r0");
        let room_id_1 = room_id!("!r1");
        let room_id_2 = room_id!("!r2");
        let thread_id_1_0 = event_id!("$ev1.0");
        let thread_id_2_0 = event_id!("$ev2.0");

        let server = MatrixMockServer::new().await;
        let client = server.client_builder().build().await;

        client.base_client().get_or_create_room(room_id_0, RoomState::Joined);
        client.base_client().get_or_create_room(room_id_1, RoomState::Joined);
        client.base_client().get_or_create_room(room_id_2, RoomState::Joined);

        let latest_events = client.latest_events().await.unwrap();

        // Despite there are many rooms, zero `RoomLatestEvents` are created.
        assert!(latest_events.state.registered_rooms.rooms.read().await.is_empty());

        // Now let's listen to two rooms.
        assert!(latest_events.listen_to_room(room_id_0).await.unwrap());
        assert!(latest_events.listen_to_room(room_id_1).await.unwrap());

        {
            let rooms = latest_events.state.registered_rooms.rooms.read().await;
            // There are two rooms…
            assert_eq!(rooms.len(), 2);
            // … which are room 0 and room 1.
            assert!(rooms.contains_key(room_id_0));
            assert!(rooms.contains_key(room_id_1));

            // Room 0 contains zero thread latest events.
            assert!(rooms.get(room_id_0).unwrap().read().await.per_thread().is_empty());
            // Room 1 contains zero thread latest events.
            assert!(rooms.get(room_id_1).unwrap().read().await.per_thread().is_empty());
        }

        // Now let's listen to one thread respectively for two rooms.
        assert!(latest_events.listen_to_thread(room_id_1, thread_id_1_0).await.unwrap());
        assert!(latest_events.listen_to_thread(room_id_2, thread_id_2_0).await.unwrap());

        {
            let rooms = latest_events.state.registered_rooms.rooms.read().await;
            // There are now three rooms…
            assert_eq!(rooms.len(), 3);
            // … yup, room 2 is now created.
            assert!(rooms.contains_key(room_id_0));
            assert!(rooms.contains_key(room_id_1));
            assert!(rooms.contains_key(room_id_2));

            // Room 0 contains zero thread latest events.
            assert!(rooms.get(room_id_0).unwrap().read().await.per_thread().is_empty());
            // Room 1 contains one thread latest event…
            let room_1 = rooms.get(room_id_1).unwrap().read().await;
            assert_eq!(room_1.per_thread().len(), 1);
            // … which is thread 1.0.
            assert!(room_1.per_thread().contains_key(thread_id_1_0));
            // Room 2 contains one thread latest event…
            let room_2 = rooms.get(room_id_2).unwrap().read().await;
            assert_eq!(room_2.per_thread().len(), 1);
            // … which is thread 2.0.
            assert!(room_2.per_thread().contains_key(thread_id_2_0));
        }
    }

    #[async_test]
    async fn test_forget_room() {
        let room_id_0 = room_id!("!r0");
        let room_id_1 = room_id!("!r1");

        let server = MatrixMockServer::new().await;
        let client = server.client_builder().build().await;

        client.base_client().get_or_create_room(room_id_0, RoomState::Joined);
        client.base_client().get_or_create_room(room_id_1, RoomState::Joined);

        let latest_events = client.latest_events().await.unwrap();

        // Now let's fetch one room.
        assert!(latest_events.listen_to_room(room_id_0).await.unwrap());

        {
            let rooms = latest_events.state.registered_rooms.rooms.read().await;
            // There are one room…
            assert_eq!(rooms.len(), 1);
            // … which is room 0.
            assert!(rooms.contains_key(room_id_0));

            // Room 0 contains zero thread latest events.
            assert!(rooms.get(room_id_0).unwrap().read().await.per_thread().is_empty());
        }

        // Now let's forget about room 0.
        latest_events.forget_room(room_id_0).await;

        {
            let rooms = latest_events.state.registered_rooms.rooms.read().await;
            // There are now zero rooms.
            assert!(rooms.is_empty());
        }
    }

    #[async_test]
    async fn test_forget_thread() {
        let room_id_0 = room_id!("!r0");
        let room_id_1 = room_id!("!r1");
        let thread_id_0_0 = event_id!("$ev0.0");

        let server = MatrixMockServer::new().await;
        let client = server.client_builder().build().await;

        client.base_client().get_or_create_room(room_id_0, RoomState::Joined);
        client.base_client().get_or_create_room(room_id_1, RoomState::Joined);

        let latest_events = client.latest_events().await.unwrap();

        // Now let's fetch one thread .
        assert!(latest_events.listen_to_thread(room_id_0, thread_id_0_0).await.unwrap());

        {
            let rooms = latest_events.state.registered_rooms.rooms.read().await;
            // There is one room…
            assert_eq!(rooms.len(), 1);
            // … which is room 0.
            assert!(rooms.contains_key(room_id_0));

            // Room 0 contains one thread latest event…
            let room_0 = rooms.get(room_id_0).unwrap().read().await;
            assert_eq!(room_0.per_thread().len(), 1);
            // … which is thread 0.0.
            assert!(room_0.per_thread().contains_key(thread_id_0_0));
        }

        // Now let's forget about the thread.
        latest_events.forget_thread(room_id_0, thread_id_0_0).await;

        {
            let rooms = latest_events.state.registered_rooms.rooms.read().await;
            // There is still one room…
            assert_eq!(rooms.len(), 1);
            // … which is room 0.
            assert!(rooms.contains_key(room_id_0));

            // But the thread has been removed.
            assert!(rooms.get(room_id_0).unwrap().read().await.per_thread().is_empty());
        }
    }

    #[async_test]
    async fn test_inputs_task_can_listen_to_room_event_cache() {
        let room_id = owned_room_id!("!r0");

        let server = MatrixMockServer::new().await;
        let client = server.client_builder().build().await;
        let weak_client = client.downgrade();
        let weak_room = WeakRoom::new(weak_client, room_id.clone());

        let event_cache = client.event_cache().unwrap();

        let registered_rooms = RwLock::new(HashMap::new());
        let (room_event_cache_generic_update_sender, mut room_event_cache_generic_update_receiver) =
            broadcast::channel(1);
        let (_send_queue_generic_update_sender, mut send_queue_generic_update_receiver) =
            broadcast::channel(1);
        let (_room_info_update_sender, mut room_info_update_receiver) = broadcast::channel(1);
        let (latest_event_queue_sender, latest_event_queue_receiver) = mpsc::unbounded_channel();

        // New event cache update, but the `LatestEvents` isn't listening to it.
        {
            room_event_cache_generic_update_sender
                .send(RoomEventCacheGenericUpdate { room_id: room_id.clone() })
                .unwrap();

            // Run the task.
            assert!(
                listen_to_updates(
                    &registered_rooms,
                    &mut room_event_cache_generic_update_receiver,
                    &mut send_queue_generic_update_receiver,
                    &mut room_info_update_receiver,
                    &latest_event_queue_sender,
                )
                .await
                .is_continue()
            );

            // No latest event computation has been triggered.
            assert!(latest_event_queue_receiver.is_empty());
        }

        // New event cache update, but this time, the `LatestEvents` is
        // listening to it.
        {
            registered_rooms.write().await.insert(
                room_id.clone(),
                With::inner(RoomLatestEvents::new(weak_room, event_cache)),
            );
            room_event_cache_generic_update_sender
                .send(RoomEventCacheGenericUpdate { room_id: room_id.clone() })
                .unwrap();

            assert!(
                listen_to_updates(
                    &registered_rooms,
                    &mut room_event_cache_generic_update_receiver,
                    &mut send_queue_generic_update_receiver,
                    &mut room_info_update_receiver,
                    &latest_event_queue_sender,
                )
                .await
                .is_continue()
            );

            // A latest event computation has been triggered!
            assert!(latest_event_queue_receiver.is_empty().not());
        }
    }

    #[async_test]
    async fn test_inputs_task_can_listen_to_send_queue() {
        let room_id = owned_room_id!("!r0");

        let server = MatrixMockServer::new().await;
        let client = server.client_builder().build().await;
        let weak_client = client.downgrade();
        let weak_room = WeakRoom::new(weak_client, room_id.clone());

        let event_cache = client.event_cache().unwrap();

        let registered_rooms = RwLock::new(HashMap::new());

        let (_room_event_cache_generic_update_sender, mut room_event_cache_generic_update_receiver) =
            broadcast::channel(1);
        let (send_queue_generic_update_sender, mut send_queue_generic_update_receiver) =
            broadcast::channel(1);
        let (_room_info_update_sender, mut room_info_update_receiver) = broadcast::channel(1);
        let (latest_event_queue_sender, latest_event_queue_receiver) = mpsc::unbounded_channel();

        // New send queue update, but the `LatestEvents` isn't listening to it.
        {
            send_queue_generic_update_sender
                .send(SendQueueUpdate {
                    room_id: room_id.clone(),
                    update: RoomSendQueueUpdate::SentEvent {
                        transaction_id: OwnedTransactionId::from("txnid0"),
                        event_id: owned_event_id!("$ev0"),
                    },
                })
                .unwrap();

            // Run the task.
            assert!(
                listen_to_updates(
                    &registered_rooms,
                    &mut room_event_cache_generic_update_receiver,
                    &mut send_queue_generic_update_receiver,
                    &mut room_info_update_receiver,
                    &latest_event_queue_sender,
                )
                .await
                .is_continue()
            );

            // No latest event computation has been triggered.
            assert!(latest_event_queue_receiver.is_empty());
        }

        // New send queue update, but this time, the `LatestEvents` is listening
        // to it.
        {
            registered_rooms.write().await.insert(
                room_id.clone(),
                With::inner(RoomLatestEvents::new(weak_room, event_cache)),
            );
            send_queue_generic_update_sender
                .send(SendQueueUpdate {
                    room_id: room_id.clone(),
                    update: RoomSendQueueUpdate::SentEvent {
                        transaction_id: OwnedTransactionId::from("txnid1"),
                        event_id: owned_event_id!("$ev1"),
                    },
                })
                .unwrap();

            assert!(
                listen_to_updates(
                    &registered_rooms,
                    &mut room_event_cache_generic_update_receiver,
                    &mut send_queue_generic_update_receiver,
                    &mut room_info_update_receiver,
                    &latest_event_queue_sender,
                )
                .await
                .is_continue()
            );

            // A latest event computation has been triggered!
            assert!(latest_event_queue_receiver.is_empty().not());
        }
    }

    #[async_test]
    async fn test_inputs_task_can_listen_to_room_info() {
        let room_id = owned_room_id!("!r0");

        let server = MatrixMockServer::new().await;
        let client = server.client_builder().build().await;
        let weak_client = client.downgrade();
        let weak_room = WeakRoom::new(weak_client, room_id.clone());

        let event_cache = client.event_cache().unwrap();

        let registered_rooms = RwLock::new(HashMap::new());

        let (_room_event_cache_generic_update_sender, mut room_event_cache_generic_update_receiver) =
            broadcast::channel(1);
        let (_send_queue_generic_update_sender, mut send_queue_generic_update_receiver) =
            broadcast::channel(1);
        let (room_info_update_sender, mut room_info_update_receiver) = broadcast::channel(1);
        let (latest_event_queue_sender, latest_event_queue_receiver) = mpsc::unbounded_channel();

        // New room info update, but the `LatestEvents` isn't listening to it.
        {
            room_info_update_sender
                .send(RoomInfoNotableUpdate {
                    room_id: room_id.clone(),
                    reasons: RoomInfoNotableUpdateReasons::MEMBERSHIP,
                })
                .unwrap();

            // Run the task.
            assert!(
                listen_to_updates(
                    &registered_rooms,
                    &mut room_event_cache_generic_update_receiver,
                    &mut send_queue_generic_update_receiver,
                    &mut room_info_update_receiver,
                    &latest_event_queue_sender,
                )
                .await
                .is_continue()
            );

            // No latest event computation has been triggered.
            assert!(latest_event_queue_receiver.is_empty());
        }

        // New room info update, but this time, the `LatestEvents` is listening
        // to it.
        {
            registered_rooms.write().await.insert(
                room_id.clone(),
                With::inner(RoomLatestEvents::new(weak_room, event_cache)),
            );
            room_info_update_sender
                .send(RoomInfoNotableUpdate {
                    room_id: room_id.clone(),
                    reasons: RoomInfoNotableUpdateReasons::MEMBERSHIP,
                })
                .unwrap();

            assert!(
                listen_to_updates(
                    &registered_rooms,
                    &mut room_event_cache_generic_update_receiver,
                    &mut send_queue_generic_update_receiver,
                    &mut room_info_update_receiver,
                    &latest_event_queue_sender,
                )
                .await
                .is_continue()
            );

            // A latest event computation has been triggered!
            assert!(latest_event_queue_receiver.is_empty().not());
        }
    }

    #[async_test]
    async fn test_inputs_task_can_listen_to_specific_room_info_update_reasons() {
        let room_id = owned_room_id!("!r0");

        let server = MatrixMockServer::new().await;
        let client = server.client_builder().build().await;
        let weak_client = client.downgrade();
        let weak_room = WeakRoom::new(weak_client, room_id.clone());

        let event_cache = client.event_cache().unwrap();

        let registered_rooms = RwLock::new(HashMap::new());

        let (_room_event_cache_generic_update_sender, mut room_event_cache_generic_update_receiver) =
            broadcast::channel(1);
        let (_send_queue_generic_update_sender, mut send_queue_generic_update_receiver) =
            broadcast::channel(1);
        let (room_info_update_sender, mut room_info_update_receiver) = broadcast::channel(1);
        let (latest_event_queue_sender, latest_event_queue_receiver) = mpsc::unbounded_channel();

        registered_rooms
            .write()
            .await
            .insert(room_id.clone(), With::inner(RoomLatestEvents::new(weak_room, event_cache)));

        // - `RoomInfoNotableUpdateReasons::LATEST_EVENT` is forbidden,
        //   otherwise it could create loops.
        // - Other reasons are ignored, except
        //   `RoomInfoNotableUpdateReasons::MEMBERSHIP`.
        for reason in {
            let mut all = RoomInfoNotableUpdateReasons::all();
            all.remove(RoomInfoNotableUpdateReasons::MEMBERSHIP);

            all.iter()
        } {
            room_info_update_sender
                .send(RoomInfoNotableUpdate { room_id: room_id.clone(), reasons: reason })
                .unwrap();

            assert!(
                listen_to_updates(
                    &registered_rooms,
                    &mut room_event_cache_generic_update_receiver,
                    &mut send_queue_generic_update_receiver,
                    &mut room_info_update_receiver,
                    &latest_event_queue_sender,
                )
                .await
                .is_continue()
            );

            // No latest event computation has been triggered.
            assert!(latest_event_queue_receiver.is_empty());
        }

        // `RoomInfoNotableUpdateReason::MEMBERSHIP` is accepted.
        {
            room_info_update_sender
                .send(RoomInfoNotableUpdate {
                    room_id: room_id.clone(),
                    reasons: RoomInfoNotableUpdateReasons::MEMBERSHIP,
                })
                .unwrap();

            assert!(
                listen_to_updates(
                    &registered_rooms,
                    &mut room_event_cache_generic_update_receiver,
                    &mut send_queue_generic_update_receiver,
                    &mut room_info_update_receiver,
                    &latest_event_queue_sender,
                )
                .await
                .is_continue()
            );

            // A latest event computation has been triggered!
            assert!(latest_event_queue_receiver.is_empty().not());
        }
    }

    #[async_test]
    async fn test_inputs_task_stops_when_event_cache_channel_is_closed() {
        let registered_rooms = RwLock::new(HashMap::new());
        let (room_event_cache_generic_update_sender, mut room_event_cache_generic_update_receiver) =
            broadcast::channel(1);
        let (_send_queue_generic_update_sender, mut send_queue_generic_update_receiver) =
            broadcast::channel(1);
        let (_room_info_update_sender, mut room_info_update_receiver) = broadcast::channel(1);
        let (latest_event_queue_sender, latest_event_queue_receiver) = mpsc::unbounded_channel();

        // Drop the sender to close the channel.
        drop(room_event_cache_generic_update_sender);

        // Run the task.
        assert!(
            listen_to_updates(
                &registered_rooms,
                &mut room_event_cache_generic_update_receiver,
                &mut send_queue_generic_update_receiver,
                &mut room_info_update_receiver,
                &latest_event_queue_sender,
            )
            .await
            // It breaks!
            .is_break()
        );

        assert!(latest_event_queue_receiver.is_empty());
    }

    #[async_test]
    async fn test_inputs_task_stops_when_send_queue_channel_is_closed() {
        let registered_rooms = RwLock::new(HashMap::new());
        let (_room_event_cache_generic_update_sender, mut room_event_cache_generic_update_receiver) =
            broadcast::channel(1);
        let (send_queue_generic_update_sender, mut send_queue_generic_update_receiver) =
            broadcast::channel(1);
        let (_room_info_update_sender, mut room_info_update_receiver) = broadcast::channel(1);
        let (latest_event_queue_sender, latest_event_queue_receiver) = mpsc::unbounded_channel();

        // Drop the sender to close the channel.
        drop(send_queue_generic_update_sender);

        // Run the task.
        assert!(
            listen_to_updates(
                &registered_rooms,
                &mut room_event_cache_generic_update_receiver,
                &mut send_queue_generic_update_receiver,
                &mut room_info_update_receiver,
                &latest_event_queue_sender,
            )
            .await
            // It breaks!
            .is_break()
        );

        assert!(latest_event_queue_receiver.is_empty());
    }

    #[async_test]
    async fn test_inputs_task_stops_when_room_info_updates_are_closed() {
        let registered_rooms = RwLock::new(HashMap::new());
        let (_room_event_cache_generic_update_sender, mut room_event_cache_generic_update_receiver) =
            broadcast::channel(1);
        let (_send_queue_generic_update_sender, mut send_queue_generic_update_receiver) =
            broadcast::channel(1);
        let (room_info_update_sender, mut room_info_update_receiver) = broadcast::channel(1);
        let (latest_event_queue_sender, latest_event_queue_receiver) = mpsc::unbounded_channel();

        // Drop the sender to close the channel.
        drop(room_info_update_sender);

        // Run the task.
        assert!(
            listen_to_updates(
                &registered_rooms,
                &mut room_event_cache_generic_update_receiver,
                &mut send_queue_generic_update_receiver,
                &mut room_info_update_receiver,
                &latest_event_queue_sender,
            )
            .await
            // It breaks!
            .is_break()
        );

        assert!(latest_event_queue_receiver.is_empty());
    }

    #[async_test]
    async fn test_latest_event_value_is_updated_via_event_cache() {
        let room_id = owned_room_id!("!r0");
        let user_id = user_id!("@mnt_io:matrix.org");
        let event_factory = EventFactory::new().sender(user_id).room(&room_id);
        let event_id_0 = event_id!("$ev0");

        let server = MatrixMockServer::new().await;
        let client = server.client_builder().build().await;

        // Create the room.
        client.base_client().get_or_create_room(&room_id, RoomState::Joined);

        let latest_events = client.latest_events().await.unwrap();

        // Subscribe to the latest event values for this room.
        let mut latest_event_stream =
            latest_events.listen_and_subscribe_to_room(&room_id).await.unwrap().unwrap();

        // The stream is pending: no new latest event for the moment.
        assert_pending!(latest_event_stream);

        // Update the event cache with a sync.
        server
            .sync_room(
                &client,
                JoinedRoomBuilder::new(&room_id)
                    .add_timeline_event(event_factory.text_msg("raclette !").event_id(event_id_0)),
            )
            .await;

        // The event cache has received its update from the sync. It has emitted
        // a generic update, which has been received by `LatestEvents` tasks, up
        // to the `compute_latest_events` which has updated the latest event
        // value.
        assert_matches!(
            latest_event_stream.next().await,
            Some(LatestEventValue::Remote(RemoteLatestEventValue { kind: TimelineEventKind::PlainText { event }, .. })) => {
                assert_matches!(
                    event.deserialize().unwrap(),
                    AnySyncTimelineEvent::MessageLike(
                        AnySyncMessageLikeEvent::RoomMessage(
                            SyncMessageLikeEvent::Original(message_content)
                        )
                    ) => {
                        assert_eq!(message_content.content.body(), "raclette !");
                    }
                );
            }
        );

        assert_pending!(latest_event_stream);
    }

    #[async_test]
    async fn test_latest_event_value_is_initialized_by_the_event_cache_lazily() {
        let room_id = owned_room_id!("!r0");
        let user_id = user_id!("@mnt_io:matrix.org");
        let event_factory = EventFactory::new().sender(user_id).room(&room_id);
        let event_id_0 = event_id!("$ev0");

        // Initialise the event cache store.
        let event_cache_store = MemoryStore::new();
        event_cache_store
            .handle_linked_chunk_updates(
                LinkedChunkId::Room(&room_id),
                vec![
                    Update::NewItemsChunk {
                        previous: None,
                        new: ChunkIdentifier::new(0),
                        next: None,
                    },
                    Update::PushItems {
                        at: Position::new(ChunkIdentifier::new(0), 0),
                        items: vec![event_factory.text_msg("hello").event_id(event_id_0).into()],
                    },
                ],
            )
            .await
            .unwrap();

        let client = MockClientBuilder::new(None)
            .on_builder(move |builder| {
                builder.store_config(
                    StoreConfig::new(CrossProcessLockConfig::MultiProcess {
                        holder_name: "foo".to_owned(),
                    })
                    .event_cache_store(event_cache_store),
                )
            })
            .build()
            .await;

        // Create the room.
        client.base_client().get_or_create_room(&room_id, RoomState::Joined);

        let latest_events = client.latest_events().await.unwrap();

        let mut latest_event_stream =
            latest_events.listen_and_subscribe_to_room(&room_id).await.unwrap().unwrap();

        // We have a race if the system is busy. Initially, the latest event
        // value is `LatestEventValue::None`, then an Event Cache generic update
        // is broadcasted manually, computing a new `LatestEventValue`. So let's
        // wait on the system to finish this, and assert the final
        // `LatestEventValue`.
        yield_now().await;
        assert_matches!(latest_event_stream.next_now().await, LatestEventValue::Remote(_));

        assert_pending!(latest_event_stream);
    }

    /// This tests a part of
    /// [`test_latest_event_value_is_initialized_by_the_event_cache_lazily`].
    ///
    /// When `RegisteredRooms::room_latest_events` restores a
    /// `LatestEventValue::None` (via `RoomLatestEvents::new`), a
    /// `LatestEventQueueUpdate::EventCache` is broadcasted to compute a
    /// `LatestEventValue` from the Event Cache lazily.
    #[async_test]
    async fn test_latest_event_value_is_initialized_by_the_event_cache_lazily_inner() {
        let room_id_0 = owned_room_id!("!r0");
        let room_id_1 = owned_room_id!("!r1");

        let server = MatrixMockServer::new().await;
        let client = server.client_builder().build().await;

        // Create the rooms.
        let room_0 = client.base_client().get_or_create_room(&room_id_0, RoomState::Joined);
        let room_1 = client.base_client().get_or_create_room(&room_id_1, RoomState::Joined);

        // Set up the rooms. `room_0` always has a `LatestEventValue::None` as
        // its the default value.
        let mut room_info_1 = room_0.clone_info();
        room_info_1.set_latest_event(LatestEventValue::LocalIsSending(local_room_message("foo")));
        room_1.update_room_info(|_| (room_info_1, Default::default())).await;

        let weak_client = client.downgrade();

        let event_cache = client.event_cache().unwrap();

        let (latest_event_queue_sender, mut latest_event_queue_receiver) =
            mpsc::unbounded_channel();

        let registered_rooms =
            RegisteredRooms::new(weak_client, event_cache, &latest_event_queue_sender);

        // Room 0 has a `LatestEventValue::None`, a
        // `LatestEventQueueUpdate::EventCache` will be broadcasted.
        {
            let room_latest_events = registered_rooms.for_room(&room_id_0).await.unwrap().unwrap();
            assert_matches!(
                room_latest_events.read().await.for_room().get().await,
                LatestEventValue::None
            );
            assert_matches!(
                latest_event_queue_receiver.recv().await,
                Some(LatestEventQueueUpdate::EventCache { room_id }) => {
                    assert_eq!(room_id, room_id_0);
                }
            );
            assert!(latest_event_queue_receiver.is_empty());
        }

        // Room 1 has a `LatestEventValue::Local*`, a
        // `LatestEventQueueUpdate::EventCache` will NOT be broadcasted.
        {
            let room_latest_events = registered_rooms.for_room(&room_id_1).await.unwrap().unwrap();
            assert_matches!(
                room_latest_events.read().await.for_room().get().await,
                LatestEventValue::LocalIsSending(_)
            );
            assert!(latest_event_queue_receiver.is_empty());
        }
    }

    #[async_test]
    async fn test_latest_event_value_is_updated_via_room_infos_for_invites() {
        let room_id = owned_room_id!("!r0");
        let event_factory = EventFactory::new().room(&room_id);
        let event_id_0 = event_id!("$ev0");
        let event_id_1 = event_id!("$ev1");

        let server = MatrixMockServer::new().await;
        let client = server.client_builder().build().await;
        let own_user_id = client.user_id().unwrap();
        let other_user_id = user_id!("@other:servername");

        let latest_events = client.latest_events().await.unwrap();

        // Subscribe to the latest event values for this room.
        let mut latest_event_stream =
            latest_events.listen_and_subscribe_to_room(&room_id).await.unwrap().unwrap();

        // The stream is pending: no new latest event for the moment.
        assert_pending!(latest_event_stream);

        let now = MilliSecondsSinceUnixEpoch::now().get();

        // Update the room with a sync: the user is invited to a room.
        {
            server
                .sync_room(
                    &client,
                    InvitedRoomBuilder::new(&room_id).add_state_event(
                        event_factory
                            .member(other_user_id)
                            .invited(own_user_id)
                            .event_id(event_id_0),
                    ),
                )
                .await;

            // The room has received its update from the sync. It has emitted a
            // room info update, which has been received by `LatestEvents`
            // tasks, up to the `compute_latest_events` which has updated the
            // latest event value.
            assert_matches!(
                latest_event_stream.next().await,
                Some(LatestEventValue::RemoteInvite { event_id, timestamp, inviter }) => {
                    // It's a stripped state event: they don't have an event ID.
                    assert!(event_id.is_none());
                    // It's a stripped state event: they don't have a timestamp (`origin_server_ts`), but `now` is normally used as a fallback.
                    assert!(timestamp.get() >= now);
                    assert_eq!(inviter.as_deref(), Some(other_user_id));
                }
            );

            assert_pending!(latest_event_stream);
        };

        // Update the room with a sync: the user is invited to the same room.
        {
            server
                .sync_room(
                    &client,
                    InvitedRoomBuilder::new(&room_id).add_state_event(
                        event_factory
                            .member(other_user_id)
                            .invited(own_user_id)
                            .event_id(event_id_0),
                    ),
                )
                .await;

            // The room has received its update from the sync. It has emitted a
            // room info update, which has been received by `LatestEvents`
            // tasks, up to the `compute_latest_events` which has NOT updated
            // the latest event value because a previous `RemoteInvite` was
            // already computed.
            assert!(timeout(Duration::from_secs(1), latest_event_stream.next()).await.is_err());

            assert_pending!(latest_event_stream);
        }

        // Update the room with a sync: the user is joining the room.
        {
            let now = u64::from(now) + 10; // time flies
            server
                .sync_room(
                    &client,
                    JoinedRoomBuilder::new(&room_id).add_timeline_event(
                        event_factory
                            .member(own_user_id)
                            .membership(MembershipState::Join)
                            .event_id(event_id_1)
                            .server_ts(now),
                    ),
                )
                .await;

            // The event cache has received its update from the sync. It has
            // emitted a generic update, which has been received by
            // `LatestEvents` tasks, up to the `compute_latest_events` which has
            // updated the latest event value.
            assert_matches!(
                latest_event_stream.next().await,
                Some(LatestEventValue::Remote(RemoteLatestEventValue { kind: TimelineEventKind::PlainText { event }, .. })) => {
                    assert_matches!(
                        event.deserialize().unwrap(),
                        AnySyncTimelineEvent::State(
                            AnySyncStateEvent::RoomMember(
                                SyncRoomMemberEvent::Original(event)
                            )
                        ) => {
                            assert_eq!(event.event_id, event_id_1);
                            assert_eq!(event.content.membership, MembershipState::Join);
                            assert_eq!(event.sender, own_user_id);
                            assert_eq!(event.state_key, own_user_id);
                            assert_eq!(u64::from(event.origin_server_ts.get()), now);
                        }
                    );
                }
            );

            assert_pending!(latest_event_stream);
        }
    }

    /// Test that the `LatestEvents` tasks don't hold onto a `Client`, which
    /// would prevent it from being dropped.
    #[async_test]
    async fn test_client_can_drop_latest_events_tasks() {
        let room_id = room_id!("!r0");

        let server = MatrixMockServer::new().await;
        let client = server.client_builder().build().await;

        // Syncing initializes the `LatestEvents`, and thus spawns its tasks.
        let room = server.sync_joined_room(&client, room_id).await;
        assert!(client.latest_events().await.unwrap().is_listening_to_room(room_id).await);

        // A `Room` holds onto a `Client`.
        drop(room);

        // Wait for anything to start up.
        yield_now().await;

        // Get a weak reference to the client.
        let weak_client = client.downgrade();

        // Dropping the client will drop the tasks.
        drop(client);

        // Wait for anything to shutdown.
        sleep(Duration::from_secs(2)).await;

        // The client has been dropped correctly.
        assert_eq!(weak_client.strong_count(), 0);
    }
}