matrix_sdk/event_cache/caches/pinned_events/

mod.rs

// Copyright 2026 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.

mod updates;

use std::{cmp::Ordering, collections::BTreeSet, fmt, sync::Arc};

use eyeball_im::VectorDiff;
use futures_util::{StreamExt as _, stream};
use matrix_sdk_base::{
    apply_redaction,
    event_cache::{Event, Gap},
    linked_chunk::{LinkedChunkId, OwnedLinkedChunkId, Position, Update},
    serde_helpers::{extract_redaction_target, extract_relation},
    sync::Timeline,
    task_monitor::BackgroundTaskHandle,
};
use matrix_sdk_common::executor::spawn;
use ruma::{
    EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedRoomId, OwnedUserId, RoomId,
    events::{relation::RelationType, room::redaction::SyncRoomRedactionEvent},
    room_version_rules::RoomVersionRules,
};
use tokio::sync::broadcast::Receiver;
use tracing::{debug, instrument, trace, warn};

pub(super) use self::updates::PinnedEventsCacheUpdateSender;
#[cfg(feature = "e2e-encryption")]
use super::super::redecryptor::MaybeResolvedEvent;
use super::{
    super::{
        EventCacheError, EventsOrigin, Result,
        deduplicator::{DeduplicationOutcome, filter_duplicate_events},
        persistence::{find_event, send_updates_to_store},
        states::{
            CacheStateLock, ReloadPreprocessing, StateLock, StateLockWriteGuard,
            selectors::PinnedEventsStateSelector,
        },
    },
    EventLocation, TimelineVectorDiffs,
    event_linked_chunk::{EventLinkedChunk, sort_positions_descending},
};
use crate::{
    Room,
    config::RequestConfig,
    event_cache::{caches::room::LinkedChunkUpdateSender, deduplicator::is_valid_event},
    room::WeakRoom,
};

pub struct PinnedEventsCacheState {
    /// The ID of the room owning this list of pinned events.
    room_id: OwnedRoomId,

    /// The user's own user id.
    own_user_id: OwnedUserId,

    /// The rules for the version of this room.
    room_version_rules: RoomVersionRules,

    /// The linked chunk representing this room's pinned events.
    ///
    /// This linked chunk also contains related events. The events are sorted in
    /// the chronological order (oldest to newest), since it would be otherwise
    /// impossible to order them correctly, given that we fetch their relations
    /// over time.
    chunk: EventLinkedChunk,

    /// Update sender for this pinned events cache.
    pub update_sender: PinnedEventsCacheUpdateSender,

    /// A sender for the globally observable linked chunk updates that happened
    /// during a sync or a back-pagination.
    ///
    /// See also [`super::super::EventCacheInner::linked_chunk_update_sender`].
    linked_chunk_update_sender: LinkedChunkUpdateSender,
}

#[cfg(not(tarpaulin_include))]
impl fmt::Debug for PinnedEventsCacheState {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("PinnedEventsCacheState")
            .field("room_id", &self.room_id)
            .field("chunk", &self.chunk)
            .finish_non_exhaustive()
    }
}

impl<'a> StateLockWriteGuard<'a, PinnedEventsCacheState> {
    /// Reload the pinned-events: only the last events will be reloaded,
    /// shrinking the in-memory size of the cache.
    ///
    /// If `preprocessing` is set to [`ReloadPreprocessing::ForgetAll`], all
    /// events will be erased before reloaded.
    #[must_use = "Propagate `VectorDiff` updates via `TimelineVectorDiffs`"]
    pub async fn reload(
        &mut self,
        preprocessing: ReloadPreprocessing,
    ) -> Result<Vec<VectorDiff<Event>>> {
        match preprocessing {
            ReloadPreprocessing::ForgetAll => {
                // Clear the `LinkedChunk` and broadcast the updates to the
                // store.
                self.state.chunk.reset();
                self.propagate_changes().await?;
            }

            ReloadPreprocessing::None => {}
        }

        // The task will notice there is a desynchronisation and will reload
        // from network.
        self.reload_from_storage().await?;

        Ok(self.state.chunk.updates_as_vector_diffs())
    }

    async fn handle_sync(&mut self, timeline: Timeline) -> Result<()> {
        let DeduplicationOutcome {
            all_events: events,
            in_memory_duplicated_event_ids,
            in_store_duplicated_event_ids,
            non_empty_all_duplicates: all_duplicates,
        } = filter_duplicate_events(
            &self.state.own_user_id,
            &self.store,
            LinkedChunkId::PinnedEvents(&self.state.room_id),
            &self.state.chunk,
            timeline.events,
        )
        .await?;

        if all_duplicates {
            // If all events are duplicates, we don't need to do anything;
            // ignore the new events.
            return Ok(());
        }

        // Remove the old duplicated events.
        //
        // We don't have to worry about the removals can change the position of
        // the existing events, because we are pushing all _new_ `events` at the
        // back.
        self.remove_events(in_memory_duplicated_event_ids, in_store_duplicated_event_ids).await?;

        // We've found new relations; append them to the linked chunk.
        self.state.chunk.push_live_events(None, &events);

        self.propagate_changes().await?;
        self.notify_subscribers(EventsOrigin::Sync);

        // Do stuff for each event.
        for event in &events {
            // Handle redaction.
            self.maybe_apply_new_redaction(event).await?;
        }

        Ok(())
    }

    /// Remove events by their position, in `EventLinkedChunk`.
    ///
    /// This method is purposely isolated because it must ensure that positions
    /// are sorted appropriately or it can be disastrous.
    #[instrument(skip_all)]
    pub async fn remove_events(
        &mut self,
        in_memory_events: Vec<(OwnedEventId, Position)>,
        in_store_events: Vec<(OwnedEventId, Position)>,
    ) -> Result<()> {
        // In-store events.
        if !in_store_events.is_empty() {
            let mut positions = in_store_events
                .into_iter()
                .map(|(_event_id, position)| position)
                .collect::<Vec<_>>();

            sort_positions_descending(&mut positions);

            let updates =
                positions.into_iter().map(|pos| Update::RemoveItem { at: pos }).collect::<Vec<_>>();

            self.apply_store_only_updates(updates).await?;
        }

        // In-memory events.
        if in_memory_events.is_empty() {
            // Nothing else to do, return early.
            return Ok(());
        }

        // `remove_events_by_position` is responsible of sorting positions.
        self.state
            .chunk
            .remove_events_by_position(
                in_memory_events.into_iter().map(|(_event_id, position)| position).collect(),
            )
            .expect("failed to remove an event");

        self.propagate_changes().await
    }

    /// Apply some updates that are effective only on the store itself.
    ///
    /// This method should be used only for updates that happen _outside_ the
    /// in-memory linked chunk. Such updates must be applied onto the persistent
    /// storage.
    async fn apply_store_only_updates(&mut self, updates: Vec<Update<Event, Gap>>) -> Result<()> {
        self.send_updates_to_store(updates).await
    }

    /// If the given event is a redaction, try to retrieve the to-be-redacted
    /// event in the chunk, and replace it by the redacted form.
    #[instrument(skip_all)]
    async fn maybe_apply_new_redaction(&mut self, event: &Event) -> Result<()> {
        let Some(event_id) =
            extract_redaction_target(event.raw(), &self.room_version_rules.redaction)
        else {
            return Ok(());
        };

        // Replace the redacted event by a redacted form, if we knew about it.
        let Some((location, mut target_event)) = self.find_event(&event_id).await? else {
            trace!("redacted event is missing from the linked chunk");
            return Ok(());
        };

        let target_event_raw = target_event.raw();

        // Don't redact already redacted events.
        if let Ok(deserialized) = target_event_raw.deserialize()
            && deserialized.is_redacted()
        {
            return Ok(());
        }

        if let Some(redacted_event) = apply_redaction(
            target_event_raw,
            event.raw().cast_ref_unchecked::<SyncRoomRedactionEvent>(),
            &self.room_version_rules.redaction,
        ) {
            // It's safe to cast `redacted_event` here:
            //
            // - either the event was an `AnyTimelineEvent` cast to
            //   `AnySyncTimelineEvent` when calling .raw(), so it's still one
            //   under the hood.
            // - or it wasn't, and it's a plain `AnySyncTimelineEvent` in this
            //   case.
            target_event.replace_raw(redacted_event.cast_unchecked());

            self.replace_event_at(location, target_event.clone()).await?;
        }

        Ok(())
    }

    /// See documentation of [`find_event`].
    pub(super) async fn find_event(
        &self,
        event_id: &EventId,
    ) -> Result<Option<(EventLocation, Event)>> {
        find_event(event_id, &self.room_id, &self.chunk, &self.store).await
    }

    /// Replaces a single event, be it saved in memory or in the store.
    ///
    /// If it was saved in memory, this will emit a notification to observers
    /// that a single item has been replaced. Otherwise, such a notification is
    /// not emitted, because observers are unlikely to observe the store updates
    /// directly.
    pub async fn replace_event_at(
        &mut self,
        location: EventLocation,
        new_event: Event,
    ) -> Result<()> {
        match location {
            EventLocation::Memory(position) => {
                self.state
                    .chunk
                    .replace_event_at(position, new_event)
                    .expect("should have been a valid position of an item");
                // We just changed the in-memory representation; synchronize
                // this with the store.
                self.propagate_changes().await?;
            }
            EventLocation::Store => {
                self.save_events([new_event]).await?;
            }
        }

        Ok(())
    }

    /// Save events into the database, without notifying observers.
    pub async fn save_events(&mut self, events: impl IntoIterator<Item = Event>) -> Result<()> {
        let store = self.store.clone();
        let room_id = self.state.room_id.clone();
        let events = events.into_iter().collect::<Vec<_>>();

        // Spawn a task so the save is uninterrupted by task cancellation.
        spawn(async move {
            for event in events {
                store.save_event(&room_id, event).await?;
            }

            Result::Ok(())
        })
        .await
        .expect("joining failed")?;

        Ok(())
    }

    /// Reload all the pinned events from storage, replacing the current linked
    /// chunk.
    async fn reload_from_storage(&mut self) -> Result<()> {
        let room_id = self.state.room_id.clone();
        let linked_chunk_id = LinkedChunkId::PinnedEvents(&room_id);

        let (last_chunk, chunk_id_gen) = self.store.load_last_chunk(linked_chunk_id).await?;

        let Some(last_chunk) = last_chunk else {
            // No pinned events stored, make sure the in-memory linked chunk is
            // sync'd (i.e. empty), and return.
            if self.state.chunk.events().next().is_some() {
                self.state.chunk.reset();
                self.notify_subscribers(EventsOrigin::Sync);
            }

            return Ok(());
        };

        {
            let mut current_chunk_identifier = last_chunk.identifier;
            self.state.chunk.shrink_to_last_reloaded_chunk(
                Some(last_chunk),
                chunk_id_gen,
                // This cache doesn't use the `OrderTracker`.
                None,
            )?;

            // Reload the entire chunk.
            while let Some(previous_chunk) =
                self.store.load_previous_chunk(linked_chunk_id, current_chunk_identifier).await?
            {
                current_chunk_identifier = previous_chunk.identifier;
                self.state.chunk.insert_new_chunk_as_first(previous_chunk)?;
            }
        }

        // Empty store updates, since we just reloaded from storage.
        self.state.chunk.store_updates().take();

        // Let observers know about it.
        self.notify_subscribers(EventsOrigin::Cache);

        Ok(())
    }

    async fn replace_all_events(&mut self, mut new_events: Vec<Event>) -> Result<()> {
        trace!("resetting all pinned events in linked chunk");

        // Invalid events - those with no ID or no event type - are never
        // persisted by the stores, so they must not reach the in-memory
        // linked chunk either: otherwise the positions in memory and in
        // the store would diverge.
        new_events.retain(is_valid_event);

        let previous_pinned_event_ids = self.state.current_event_ids();

        if new_events
            .iter()
            .filter_map(|e| e.event_id())
            .map(ToOwned::to_owned)
            .collect::<BTreeSet<_>>()
            == previous_pinned_event_ids.into_iter().collect()
        {
            // No change in the list of pinned events.
            return Ok(());
        }

        if self.state.chunk.events().next().is_some() {
            self.state.chunk.reset();
        }

        self.state.chunk.push_live_events(None, &new_events);
        self.propagate_changes().await?;
        self.notify_subscribers(EventsOrigin::Sync);

        Ok(())
    }

    /// Propagate the changes in this linked chunk to observers, and save the
    /// changes on disk.
    pub async fn propagate_changes(&mut self) -> Result<()> {
        let updates = self.state.chunk.store_updates().take();

        self.send_updates_to_store(updates).await
    }

    async fn send_updates_to_store(&mut self, updates: Vec<Update<Event, Gap>>) -> Result<()> {
        let linked_chunk_id = OwnedLinkedChunkId::PinnedEvents(self.room_id.clone());

        send_updates_to_store(
            &self.store,
            linked_chunk_id,
            &self.state.linked_chunk_update_sender,
            updates,
        )
        .await
    }

    /// Notify subscribers of timeline updates.
    fn notify_subscribers(&mut self, origin: EventsOrigin) {
        let diffs = self.state.chunk.updates_as_vector_diffs();

        if !diffs.is_empty() {
            self.update_sender.send(TimelineVectorDiffs { diffs, origin });
        }
    }
}

impl PinnedEventsCacheState {
    /// Return a list of the current event IDs in this linked chunk.
    pub(super) fn current_event_ids(&self) -> Vec<OwnedEventId> {
        self.chunk
            .events()
            .filter_map(|(_position, event)| event.event_id().map(ToOwned::to_owned))
            .collect()
    }

    /// Returns whether this contains exactly the given pinned events,
    /// along with the events related to them (reactions, edits, redactions,
    /// etc).
    ///
    /// Related events are ignored here when comparing the pinned event IDs
    /// to all of the [`Self::current_event_ids`], as that would cause
    /// differences if one pinned event has a related event, and then reload
    /// them all over again.
    fn has_exactly_pinned_events(&self, pinned_event_ids: &[OwnedEventId]) -> bool {
        let pinned_event_ids: BTreeSet<&EventId> =
            pinned_event_ids.iter().map(|event_id| &**event_id).collect();
        let event_ids: BTreeSet<&EventId> =
            self.chunk.events().filter_map(|(_position, event)| event.event_id()).collect();

        if !pinned_event_ids.is_subset(&event_ids) {
            return false;
        }

        // Every other event must relate to an event in this linked chunk,
        // just like `aggregate_timeline_for_pinned_events` does for a sync.
        // If not, then that event isn't pinned anymore.
        self.chunk.events().all(|(_position, event)| {
            event.event_id().is_some_and(|event_id| pinned_event_ids.contains(event_id))
                || extract_relation(event.raw()).is_some_and(|(relation_type, related_event_id)| {
                    relation_type != RelationType::Thread && event_ids.contains(&*related_event_id)
                })
                || extract_redaction_target(event.raw(), &self.room_version_rules.redaction)
                    .is_some_and(|redacted_event_id| event_ids.contains(&*redacted_event_id))
        })
    }
}

/// All the information related to room's pinned events..
///
/// Cloning is shallow, and thus is cheap to do.
#[derive(Clone)]
pub struct PinnedEventsCache {
    inner: Arc<PinnedEventsCacheInner>,

    /// The task handling the refreshing of pinned events for this specific
    /// room.
    _task: Arc<BackgroundTaskHandle>,
}

/// The (non-cloneable) details of the `PinnedEventsCache`.
struct PinnedEventsCacheInner {
    /// The ID of the room owning this list of pinned events.
    room_id: OwnedRoomId,

    /// State of this `PinnedEventsCache`.
    ///
    /// It is behind an `Arc` because it is shared with the task.
    state: CacheStateLock<PinnedEventsStateSelector>,
}

impl PinnedEventsCache {
    /// Creates a new [`PinnedEventsCache`] for the given room.
    pub(in super::super) async fn new(
        weak_room: &WeakRoom,
        own_user_id: OwnedUserId,
        room_version_rules: RoomVersionRules,
        linked_chunk_update_sender: LinkedChunkUpdateSender,
        state: &StateLock,
    ) -> Result<Self> {
        let room = weak_room.get().ok_or(EventCacheError::ClientDropped)?;
        let room_id = room.room_id().to_owned();

        let cache_state = state
            .try_insert_once_with(
                PinnedEventsStateSelector::new(room_id.clone()),
                |_store_guard| async {
                    Ok(PinnedEventsCacheState {
                        room_id: room_id.clone(),
                        own_user_id,
                        room_version_rules,
                        chunk: EventLinkedChunk::new(),
                        update_sender: PinnedEventsCacheUpdateSender::new(),
                        linked_chunk_update_sender,
                    })
                },
            )
            .await?;

        let inner = Arc::new(PinnedEventsCacheInner { room_id, state: cache_state });

        let task = room
            .client()
            .task_monitor()
            .spawn_infinite_task(
                "pinned_event_listener_task",
                Self::pinned_event_listener_task(room, inner.clone()),
            )
            .abort_on_drop();

        Ok(Self { inner, _task: Arc::new(task) })
    }

    /// Get the room ID of this cache.
    pub fn room_id(&self) -> &RoomId {
        &self.inner.room_id
    }

    /// Return a reference to the state.
    pub(super) fn state(&self) -> &CacheStateLock<PinnedEventsStateSelector> {
        &self.inner.state
    }

    /// Subscribe to live events from this room's pinned events cache.
    pub async fn subscribe(&self) -> Result<(Vec<Event>, Receiver<TimelineVectorDiffs>)> {
        let guard = self.inner.state.read().await?;
        let events = guard.state.chunk.events().map(|(_position, item)| item.clone()).collect();

        let recv = guard.state.update_sender.new_pinned_events_receiver();

        Ok((events, recv))
    }

    /// Try to locate the events in the linked chunk corresponding to the given
    /// list of resolved events, and replace them, while alerting observers
    /// about the update.
    #[cfg(feature = "e2e-encryption")]
    pub(in super::super) async fn replace_in_memory_utds(
        &self,
        resolved_events: &[MaybeResolvedEvent],
    ) -> Result<()> {
        let mut state = self.inner.state.write().await?;

        // Drain the updates to the store, events have already been updated
        // before calling this method.
        let _ = state.state.chunk.store_updates().take();

        if state.state.chunk.replace_utds(resolved_events) {
            state.propagate_changes().await?;
            state.notify_subscribers(EventsOrigin::Cache);
        }

        Ok(())
    }

    /// Handle an update from a joined room.
    #[instrument(skip_all, fields(room_id = %self.inner.room_id))]
    pub(super) async fn handle_joined_room_update(&self, timeline: Timeline) -> Result<()> {
        self.handle_timeline(timeline).await
    }

    /// Handle an update from a left room.
    #[instrument(skip_all, fields(room_id = %self.inner.room_id))]
    pub(super) async fn handle_left_room_update(&self, timeline: Timeline) -> Result<()> {
        self.handle_timeline(timeline).await
    }

    /// Handle a [`Timeline`], i.e. new events received by a sync for this
    /// thread.
    async fn handle_timeline(&self, timeline: Timeline) -> Result<()> {
        if timeline.events.is_empty() {
            return Ok(());
        }

        trace!("adding new {} events", timeline.events.len());

        self.inner.state.write().await?.handle_sync(timeline).await
    }

    #[instrument(fields(%room_id = room.room_id()), skip(room, inner))]
    async fn pinned_event_listener_task(room: Room, inner: Arc<PinnedEventsCacheInner>) {
        debug!("pinned events listener task started");

        let max_pinned_events_to_load = room
            .client()
            .event_cache()
            .expect("Pinned-events are running inside the Event Cache")
            .config()
            .max_pinned_events_to_load;

        let reload_from_network = async |room: Room| {
            let events = match Self::reload_pinned_events(room).await {
                Ok(Some(events)) => events,
                Ok(None) => Vec::new(),
                Err(err) => {
                    warn!("error when loading pinned events: {err}");
                    return;
                }
            };

            // Replace the whole linked chunk with those new events, and
            // propagate updates to the observers.
            match inner.state.write().await {
                Ok(mut guard) => {
                    guard.replace_all_events(events).await.unwrap_or_else(|err| {
                        warn!("error when replacing pinned events: {err}");
                    });
                }

                Err(err) => {
                    warn!("error when acquiring write lock to replace pinned events: {err}");
                }
            }
        };

        // Reload the pinned events from the storage first.
        match inner.state.write().await {
            Ok(mut guard) => {
                // On startup, reload the pinned events from storage.
                guard.reload_from_storage().await.unwrap_or_else(|err| {
                    warn!("error when reloading pinned events from storage, at start: {err}");
                });

                // Compare the initial list of pinned events to the one in the
                // linked chunk.
                let actual_pinned_events = pinned_event_ids_to_load(
                    max_pinned_events_to_load,
                    room.pinned_event_ids().unwrap_or_default(),
                );

                if !guard.state.has_exactly_pinned_events(&actual_pinned_events) {
                    // Reload the list of pinned events from network.
                    drop(guard);
                    reload_from_network(room.clone()).await;
                }
            }

            Err(err) => {
                warn!("error when acquiring write lock to initialize pinned events: {err}");
            }
        }

        let weak_room = WeakRoom::new(room.client().downgrade(), room.room_id().to_owned());

        let mut stream = room.pinned_event_ids_stream();

        drop(room);

        // Whenever the list of pinned events changes, reload it.
        while let Some(new_list) = stream.next().await {
            trace!("handling update");

            let Some(room) = weak_room.get() else {
                debug!("room has been dropped, ending pinned events listener task");
                break;
            };

            let new_list = pinned_event_ids_to_load(max_pinned_events_to_load, new_list);

            let guard = match inner.state.read().await {
                Ok(guard) => guard,
                Err(err) => {
                    warn!("error when acquiring read lock to handle pinned events update: {err}");
                    break;
                }
            };

            // Compare to the current linked chunk.
            if guard.state.has_exactly_pinned_events(&new_list) {
                // All the events in the pinned list are the same, don't reload.
                continue;
            }

            drop(guard);

            // Event IDs differ, so reload all the pinned events.
            reload_from_network(room).await;
        }

        debug!("pinned events listener task ended");
    }

    /// Loads the pinned events in this room, using the cache first and then
    /// requesting the event from the homeserver if it couldn't be found. This
    /// method will perform as many concurrent requests for events as
    /// `max_concurrent_requests` allows, to avoid overwhelming the server.
    ///
    /// Returns `None` if the list of pinned events hasn't changed since the
    /// previous time we loaded them. May return an error if there was an issue
    /// fetching the full events.
    async fn reload_pinned_events(room: Room) -> Result<Option<Vec<Event>>> {
        let (max_concurrent_requests, max_pinned_events_to_load) = {
            let client = room.client();
            let config = client
                .event_cache()
                .expect("Pinned-events are running inside the Event Cache")
                .config();

            (config.max_pinned_events_concurrent_requests, config.max_pinned_events_to_load)
        };

        let pinned_event_ids = pinned_event_ids_to_load(
            max_pinned_events_to_load,
            room.pinned_event_ids().unwrap_or_default(),
        );

        if pinned_event_ids.is_empty() {
            return Ok(Some(Vec::new()));
        }

        let mut num_successful_loads = 0;

        let mut loaded_events: Vec<Event> =
            stream::iter(pinned_event_ids.clone().into_iter().map(|event_id| {
                let room = room.clone();
                let filter = vec![RelationType::Annotation, RelationType::Replacement];
                let request_config = RequestConfig::default().retry_limit(3);

                async move {
                    let (target, mut relations) = room
                        .load_or_fetch_event_with_relations(
                            &event_id,
                            Some(filter),
                            Some(request_config),
                        )
                        .await?;

                    relations.insert(0, target);
                    Ok::<_, crate::Error>(relations)
                }
            }))
            .buffer_unordered(max_concurrent_requests)
            // Count successful queries.
            .inspect(|result| {
                if result.is_ok() {
                    num_successful_loads += 1;
                }
            })
            // Get rid of error results.
            .flat_map(stream::iter)
            // Flatten the list of `Vec<Event>` into a list of `Event`.
            .flat_map(stream::iter)
            .collect()
            .await;

        if num_successful_loads != pinned_event_ids.len() {
            warn!(
                "only successfully loaded {} out of {} pinned events",
                num_successful_loads,
                pinned_event_ids.len()
            );
        }

        if loaded_events.is_empty() {
            // If the list of loaded events is empty, we ran into an error to
            // load _all_ the pinned events, which needs to be reported to the
            // caller.
            return Err(EventCacheError::UnableToLoadPinnedEvents);
        }

        // Since we have all the events and their related events, we can't
        // nicely sort them, since we've lost all ordering information from
        // using /event or /relations. Resort to sorting using chronological
        // ordering (oldest -> newest).
        loaded_events.sort_by(compare_pinned_items);

        Ok(Some(loaded_events))
    }
}

impl fmt::Debug for PinnedEventsCache {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("PinnedEventsCache").finish_non_exhaustive()
    }
}

/// Returns the IDs of the pinned events that this cache loads
/// from the given list of all the room's pinned events.
///
/// This will include only the most recently pinned ones, up to the limit of
/// the chosen `max_pinned_events_to_load` cfg value.
#[inline]
fn pinned_event_ids_to_load(
    max_events_to_load: usize,
    pinned_event_ids: Vec<OwnedEventId>,
) -> Vec<OwnedEventId> {
    pinned_event_ids.into_iter().rev().take(max_events_to_load).rev().collect()
}

fn compare_pinned_items(a: &Event, b: &Event) -> Ordering {
    let a_time: Option<MilliSecondsSinceUnixEpoch> = a.timestamp_raw();
    let b_time: Option<MilliSecondsSinceUnixEpoch> = b.timestamp_raw();

    compare_by_optional_timestamp(a_time, b_time)
}

fn compare_by_optional_timestamp(
    a: Option<MilliSecondsSinceUnixEpoch>,
    b: Option<MilliSecondsSinceUnixEpoch>,
) -> Ordering {
    match (a, b) {
        (None, None) => Ordering::Equal,
        (None, Some(_)) => Ordering::Greater,
        (Some(_), None) => Ordering::Less,
        (Some(a), Some(b)) => a.cmp(&b),
    }
}

#[cfg(not(target_family = "wasm"))]
#[cfg(test)]
mod tests {
    use proptest::prelude::*;
    use ruma::UInt;

    use super::*;

    fn any_timestamp() -> impl Strategy<Value = Option<MilliSecondsSinceUnixEpoch>> {
        prop::option::of(
            any::<u32>().prop_map(|value| MilliSecondsSinceUnixEpoch(UInt::from(value))),
        )
    }

    #[test]
    fn sort_pinned_events_never_panics_only_nones() {
        let mut vec = vec![None; 100_000];
        vec.sort_by(|a, b| compare_by_optional_timestamp(*a, *b))
    }

    proptest! {
    #[test]
    fn sort_pinned_events_never_panics(mut v in prop::collection::vec(any_timestamp(), 0..1000)) {
        v.sort_by(
            |a, b| compare_by_optional_timestamp(*a, *b))
    }

    #[test]
    fn compare_pinned_events_reflexive(a in any_timestamp()) {
        prop_assert_eq!(compare_by_optional_timestamp(a, a), Ordering::Equal);
    }

    #[test]
    fn compare_pinned_events_antisymmetric(a in any_timestamp(), b in any_timestamp()) {
        let ab = compare_by_optional_timestamp(a, b);
        let ba = compare_by_optional_timestamp(b, a);

        prop_assert_eq!(ab, ba.reverse());
    }

    #[test]
    fn compare_pinned_events_transitive(
        a in any_timestamp(),
        b in any_timestamp(),
        c in any_timestamp()
    ) {
        let ab = compare_by_optional_timestamp(a, b);
        let bc = compare_by_optional_timestamp(b, c);
        let ac = compare_by_optional_timestamp(a, c);

        if ab == Ordering::Less && bc == Ordering::Less {
            prop_assert_eq!(ac, Ordering::Less);
        }

        if ab == Ordering::Equal && bc == Ordering::Equal {
            prop_assert_eq!(ac, Ordering::Equal);
        }

        if ab == Ordering::Greater && bc == Ordering::Greater {
            prop_assert_eq!(ac, Ordering::Greater);
        }
    }
    }
}