macltok

public

Star0

Fork 0

Watch0

88 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `66db1b3cd895`

tarball

atagen <boss@atagen.co>2026-10-25 13:25 UTC

7 files changed+979-14

threads: list a room's threads and keep them current

Malice needs a thread list per room that stays current as threads gain replies, without opening the room. matrix-sdk-ui's ThreadListService never adds a thread it hasn't paged in, never reorders, and counts every event the cache replays or back paginates as a new reply. ThreadListCommand drives a worker that pages /threads through Room::list_threads, follows the room's event cache to count each newer reply once and pull in roots of newly active threads, and emits keyed upserts that ThreadListEntry::upsert orders so the mirror can't diverge.

Modified file: docs/design-decisions.md+25 additions@@ -124,6 +124,31 @@ a custom emote reaction keys on the image's mxc uri and carries its shortcode
 which image it is and the shortcode any sender gave, and `toggle_reaction`
 takes the shortcode to send
 
+## thread lists
+
+a room's thread list is its own thing, not a timeline and not the room actor's.
+it needs only the room, so a room needn't be open for its list. one worker holds
+every open list. `Open` sends an empty `ThreadListReset`, `Paginate` asks
+`/rooms/{id}/threads` through the sdk's `Room::list_threads`, and the list
+follows the room's event cache for as long as it's open. changes go out as
+upserts keyed by root, which `ThreadListEntry::upsert` puts in order, newest
+activity first and ties by root, so a list can't diverge the way an index diff
+can
+
+the sdk's `ThreadListService` was passed over. it never adds a thread that wasn't
+on a page already loaded, never reorders, and adds one to the count for every
+event the cache hands it, replays and back pagination included. here a reply
+counts once, only when it's newer than the thread's latest, and a reply sync
+brings to a thread we haven't loaded fetches its root and adds it. one a back
+pagination brings doesn't, since that says nothing about the thread being
+active. a redacted reply doesn't uncount, an edit doesn't change the latest
+reply's preview, and a preview's own reply target comes back `Unavailable`
+rather than fetched. a lagged listener misses whatever it skipped until the
+thread moves again
+
+a list's media belongs to its own scope, bounded like an event context's so a
+long open list sheds the previews its threads moved past, and released on close
+
 ## warming up to send
 
 the sdk does all its key work on the send path, in the room's send queue before

Modified file: src/engine/mod.rs+18 additions−2 deletions@@ -8,6 +8,7 @@ mod prepare;
 mod repair;
 mod room;
 mod session;
+mod thread_list;
 mod timeline;
 mod url_preview;
 mod watchers;
@@ -51,7 +52,7 @@ use crate::{
     AccountCommand, ClientHandle, Command, CommandOutput, ConnectionState, EncryptionCommand,
     EngineErrorKind, EngineFailure, Error, ErrorKind, Event, EventStream, ImagePackSource,
     LifecycleCommand, MediaCommand, NotificationCommand, PackCommand, RoomCommand, Termination,
-    TimelineCommand,
+    ThreadListCommand, TimelineCommand,
     normalise::{ROOM_PACKS, ROOM_PREVIEW_URLS},
     transport::{DriverPort, Request, Responder, transport_pair},
 };
@@ -289,6 +290,7 @@ struct Queues {
     mutations: mpsc::Receiver<MutationJob>,
     notifications: mpsc::Receiver<Job<NotificationCommand>>,
     encryption: mpsc::Receiver<Job<EncryptionCommand>>,
+    thread_lists: mpsc::Receiver<Job<ThreadListCommand>>,
     session_changes: broadcast::Receiver<SessionChange>,
     persistence: Option<Persistence>,
 }
@@ -314,6 +316,7 @@ pub struct Engine {
     mutations: mpsc::Sender<MutationJob>,
     notifications: mpsc::Sender<Job<NotificationCommand>>,
     encryption: mpsc::Sender<Job<EncryptionCommand>>,
+    thread_lists: mpsc::Sender<Job<ThreadListCommand>>,
     queues: Option<Queues>,
     /// the client can outlive us, so the handler has to come off it when we stop
     verification_handler: Option<matrix_sdk::event_handler::EventHandlerHandle>,
@@ -403,6 +406,7 @@ impl Engine {
         let (mutations, mutation_queue) = mpsc::channel(config.command_capacity);
         let (notifications, notification_queue) = mpsc::channel(config.command_capacity);
         let (encryption, encryption_queue) = mpsc::channel(config.command_capacity);
+        let (thread_lists, thread_list_queue) = mpsc::channel(config.command_capacity);
         let engine = Self {
             contexts: Arc::new(context::Sessions::new(port.event_sender())),
             client,
@@ -421,10 +425,12 @@ impl Engine {
             mutations,
             notifications,
             encryption,
+            thread_lists,
             queues: Some(Queues {
                 mutations: mutation_queue,
                 notifications: notification_queue,
                 encryption: encryption_queue,
+                thread_lists: thread_list_queue,
                 session_changes,
                 persistence,
             }),
@@ -538,7 +544,7 @@ impl Engine {
         let client = &self.client;
         let events = || self.port.event_sender();
         let (notices, notice_queue) = mpsc::unbounded_channel();
-        let watchers: [(&'static str, EngineBoxFuture<'static, ()>); 14] = [
+        let watchers: [(&'static str, EngineBoxFuture<'static, ()>); 15] = [
             (
                 "upload tracker",
                 Box::pin(watchers::uploads(
@@ -642,6 +648,15 @@ impl Engine {
                     incoming_verifications,
                 )),
             ),
+            (
+                "thread list worker",
+                Box::pin(thread_list::run(
+                    client.clone(),
+                    events(),
+                    self.media.clone(),
+                    queues.thread_lists,
+                )),
+            ),
         ];
         for (label, task) in watchers {
             self.tasks.spawn(TaskContext::Watcher(label), task);
@@ -847,6 +862,7 @@ impl Engine {
                 enqueue(&self.notifications, Job { command, reply });
             }
             Command::Pack(command) => self.pack(command, reply),
+            Command::ThreadList(command) => enqueue(&self.thread_lists, Job { command, reply }),
         }
         None
     }

Added file: src/engine/thread_list.rs+740 additions@@ -0,0 +1,740 @@
+use std::{
+    collections::{HashMap, hash_map::Entry},
+    sync::Arc,
+};
+
+use matrix_sdk::{
+    Client, Room,
+    deserialized_responses::TimelineEvent as SdkTimelineEvent,
+    event_cache::{EventsOrigin, RoomEventCacheSubscriber, RoomEventCacheUpdate},
+    room::ListThreadsOptions,
+    serde_helpers::extract_thread_root,
+};
+use matrix_sdk_ui::eyeball_im::VectorDiff;
+use ruma::{OwnedEventId, OwnedRoomId, OwnedUserId, UInt, push::Action};
+use tokio::sync::{broadcast::error::RecvError, mpsc};
+
+use super::{
+    MediaService,
+    commands::Job,
+    executor::{TaskSet, is_panic, resume_unwind},
+};
+use crate::{
+    Error, ErrorKind, Event, HistoryState, MediaId, Sender, ThreadListCommand, ThreadListEntry,
+    ThreadSummary, TimelineItem,
+    normalise::{MediaRegistration, MediaScope, normalise_observed, resolve_replies},
+    transport::{EventSender, Responder},
+};
+
+const MAX_PAGE: u16 = 50;
+
+struct Normalised {
+    event_id: OwnedEventId,
+    timestamp_ms: u64,
+    item: TimelineItem,
+    media: Vec<MediaRegistration>,
+}
+
+enum Arrival {
+    Page {
+        room_id: OwnedRoomId,
+        generation: u64,
+        page: Result<Page, Error>,
+    },
+    /// something landed in a thread, or a root changed, as the room's event cache saw it
+    Live {
+        room_id: OwnedRoomId,
+        generation: u64,
+        live: Live,
+    },
+    /// a root we heard a reply for before we knew of its thread
+    Root {
+        room_id: OwnedRoomId,
+        generation: u64,
+        root_event_id: OwnedEventId,
+        built: Option<Built>,
+    },
+}
+
+enum Live {
+    Reply {
+        root_event_id: OwnedEventId,
+        reply: Normalised,
+        /// back pagination, which says nothing about a thread being active
+        paginated: bool,
+    },
+    Root(Normalised),
+}
+
+struct Page {
+    entries: Vec<Built>,
+    token: Option<String>,
+}
+
+struct Built {
+    entry: ThreadListEntry,
+    media: Vec<MediaRegistration>,
+}
+
+struct List {
+    room: Room,
+    generation: u64,
+    scope: MediaScope,
+    entries: Vec<ThreadListEntry>,
+    token: Option<String>,
+    end_reached: bool,
+    /// non-empty while a page is in flight, everyone waiting on it
+    paginating: Vec<Responder>,
+    /// roots being fetched, with the replies that arrived meanwhile
+    fetching: HashMap<OwnedEventId, Vec<Normalised>>,
+    listener: super::executor::AbortHandle,
+}
+
+pub(super) async fn run(
+    client: Client,
+    events: EventSender,
+    media: Arc<MediaService>,
+    mut jobs: mpsc::Receiver<Job<ThreadListCommand>>,
+) {
+    let (arrivals, mut arrived) = mpsc::unbounded_channel();
+    let mut lists = ThreadLists {
+        current_user_id: client.user_id().map(ToOwned::to_owned),
+        client,
+        events,
+        media,
+        lists: HashMap::new(),
+        next_generation: 0,
+        arrivals,
+        tasks: TaskSet::new(),
+    };
+    loop {
+        tokio::select! {
+            job = jobs.recv() => {
+                let Some(Job { command, reply }) = job else { break };
+                lists.execute(command, reply).await;
+            }
+            Some(arrival) = arrived.recv() => lists.arrive(arrival).await,
+            Some(done) = lists.tasks.join_next(), if !lists.tasks.is_empty() => {
+                if let Err(error) = done && is_panic(&error) {
+                    resume_unwind(error);
+                }
+            }
+        }
+    }
+    lists.tasks.shutdown().await;
+}
+
+struct ThreadLists {
+    client: Client,
+    current_user_id: Option<OwnedUserId>,
+    events: EventSender,
+    media: Arc<MediaService>,
+    lists: HashMap<OwnedRoomId, List>,
+    next_generation: u64,
+    arrivals: mpsc::UnboundedSender<Arrival>,
+    tasks: TaskSet<()>,
+}
+
+impl ThreadLists {
+    async fn execute(&mut self, command: ThreadListCommand, reply: Responder) {
+        match command {
+            ThreadListCommand::Open(room_id) => self.open(room_id, reply).await,
+            ThreadListCommand::Paginate { room_id, limit } => {
+                self.paginate(room_id, limit, reply).await;
+            }
+            ThreadListCommand::Close(room_id) => self.close(room_id, reply).await,
+        }
+    }
+
+    async fn open(&mut self, room_id: OwnedRoomId, reply: Responder) {
+        if let Some(list) = self.lists.get(&room_id) {
+            let reset = Event::ThreadListReset {
+                room_id,
+                entries: list.entries.iter().cloned().collect(),
+            };
+            return reply.done(self.emit(reset).await);
+        }
+        let Some(room) = self.client.get_room(&room_id) else {
+            return reply.done(Err(Error::room_unknown()));
+        };
+        self.next_generation = self.next_generation.wrapping_add(1);
+        let generation = self.next_generation;
+        let listener = self.tasks.spawn(listen(
+            room.clone(),
+            generation,
+            self.current_user_id.clone(),
+            self.arrivals.clone(),
+        ));
+        self.lists.insert(
+            room_id.clone(),
+            List {
+                room,
+                generation,
+                scope: self.media.context_scope(),
+                entries: Vec::new(),
+                token: None,
+                end_reached: false,
+                paginating: Vec::new(),
+                fetching: HashMap::new(),
+                listener,
+            },
+        );
+        let reset = Event::ThreadListReset {
+            room_id,
+            entries: imbl::Vector::new(),
+        };
+        reply.done(self.emit(reset).await);
+    }
+
+    async fn paginate(&mut self, room_id: OwnedRoomId, limit: u16, reply: Responder) {
+        if limit == 0 || limit > MAX_PAGE {
+            return reply.done(Err(Error::new(
+                ErrorKind::Invalid,
+                "thread list page limit must be between 1 and 50",
+            )));
+        }
+        let Some(list) = self.lists.get_mut(&room_id) else {
+            return reply.done(Err(not_open()));
+        };
+        if list.end_reached {
+            return reply.done(Ok(()));
+        }
+        list.paginating.push(reply);
+        if list.paginating.len() > 1 {
+            return;
+        }
+        let fetch = fetch_page(
+            list.room.clone(),
+            list.token.clone(),
+            limit,
+            self.current_user_id.clone(),
+        );
+        let generation = list.generation;
+        let arrivals = self.arrivals.clone();
+        let page_room_id = room_id.clone();
+        self.tasks.spawn(async move {
+            let page = fetch.await;
+            let _ = arrivals.send(Arrival::Page {
+                room_id: page_room_id,
+                generation,
+                page,
+            });
+        });
+        let _ = self
+            .emit(Event::ThreadListHistoryChanged {
+                room_id,
+                state: HistoryState::Loading,
+            })
+            .await;
+    }
+
+    async fn close(&mut self, room_id: OwnedRoomId, reply: Responder) {
+        let Some(list) = self.lists.remove(&room_id) else {
+            return reply.done(Err(not_open()));
+        };
+        list.listener.abort();
+        for waiting in list.paginating {
+            waiting.done(Err(Error::interrupted()));
+        }
+        self.media.release(&list.scope).await;
+        reply.done(self.emit(Event::ThreadListClosed { room_id }).await);
+    }
+
+    async fn arrive(&mut self, arrival: Arrival) {
+        match arrival {
+            Arrival::Page {
+                room_id,
+                generation,
+                page,
+            } => self.page_arrived(room_id, generation, page).await,
+            Arrival::Live {
+                room_id,
+                generation,
+                live,
+            } => self.live_arrived(room_id, generation, live).await,
+            Arrival::Root {
+                room_id,
+                generation,
+                root_event_id,
+                built,
+            } => {
+                let Some(list) = current(&mut self.lists, &room_id, generation) else {
+                    return;
+                };
+                let replies = list.fetching.remove(&root_event_id).unwrap_or_default();
+                let Some(mut built) = built else {
+                    return;
+                };
+                if list.position(&root_event_id).is_some() {
+                    return;
+                }
+                for reply in replies {
+                    built.media.extend(add_reply(&mut built.entry, reply));
+                }
+                self.publish(&room_id, vec![built]).await;
+            }
+        }
+    }
+
+    async fn page_arrived(
+        &mut self,
+        room_id: OwnedRoomId,
+        generation: u64,
+        page: Result<Page, Error>,
+    ) {
+        let Some(list) = current(&mut self.lists, &room_id, generation) else {
+            return;
+        };
+        let waiting = std::mem::take(&mut list.paginating);
+        let (result, state) = match page {
+            Ok(page) => {
+                list.end_reached = page.token.is_none();
+                list.token = page.token;
+                let fresh = page
+                    .entries
+                    .into_iter()
+                    .filter(|built| {
+                        list.position(&built.entry.root_event_id)
+                            .is_none_or(|index| {
+                                list.entries[index].latest_activity_ms
+                                    <= built.entry.latest_activity_ms
+                            })
+                    })
+                    .collect();
+                let state = if list.end_reached {
+                    HistoryState::StartReached
+                } else {
+                    HistoryState::Idle
+                };
+                self.publish(&room_id, fresh).await;
+                (Ok(()), state)
+            }
+            Err(error) => {
+                let state = HistoryState::Failed(error.message.clone());
+                (Err(error), state)
+            }
+        };
+        let _ = self
+            .emit(Event::ThreadListHistoryChanged { room_id, state })
+            .await;
+        for reply in waiting {
+            reply.done(result.clone());
+        }
+    }
+
+    async fn live_arrived(&mut self, room_id: OwnedRoomId, generation: u64, live: Live) {
+        let Some(list) = current(&mut self.lists, &room_id, generation) else {
+            return;
+        };
+        match live {
+            Live::Reply {
+                root_event_id,
+                reply,
+                paginated,
+            } => {
+                if let Some(index) = list.position(&root_event_id) {
+                    let mut entry = list.entries[index].clone();
+                    let media = add_reply(&mut entry, reply);
+                    if entry != list.entries[index] {
+                        self.publish(&room_id, vec![Built { entry, media }]).await;
+                    }
+                    return;
+                }
+                if paginated {
+                    return;
+                }
+                match list.fetching.entry(root_event_id.clone()) {
+                    Entry::Occupied(mut waiting) => waiting.get_mut().push(reply),
+                    Entry::Vacant(vacant) => {
+                        vacant.insert(vec![reply]);
+                        let room = list.room.clone();
+                        let current_user_id = self.current_user_id.clone();
+                        let arrivals = self.arrivals.clone();
+                        self.tasks.spawn(async move {
+                            let built =
+                                fetch_root(&room, &root_event_id, current_user_id).await;
+                            let _ = arrivals.send(Arrival::Root {
+                                room_id: room.room_id().to_owned(),
+                                generation,
+                                root_event_id,
+                                built,
+                            });
+                        });
+                    }
+                }
+            }
+            Live::Root(root) => {
+                let Some(index) = list.position(&root.event_id) else {
+                    return;
+                };
+                let mut entry = list.entries[index].clone();
+                entry.root = with_summary(root.item, &entry.summary);
+                if entry != list.entries[index] {
+                    let media = root.media;
+                    self.publish(&room_id, vec![Built { entry, media }]).await;
+                }
+            }
+        }
+    }
+
+    /// the one place a thread list's media is registered. its scope is bounded like an event
+    /// context's, so previews a long open list replaced age out, and it goes on close
+    async fn publish(&mut self, room_id: &OwnedRoomId, built: Vec<Built>) {
+        if built.is_empty() {
+            return;
+        }
+        let Some(list) = self.lists.get_mut(room_id) else {
+            return;
+        };
+        let mut media = Vec::new();
+        let scope = list.scope.clone();
+        let mut upserts = Vec::with_capacity(built.len());
+        for Built {
+            entry,
+            media: registrations,
+        } in built
+        {
+            media.extend(registrations);
+            list.upsert(entry.clone());
+            upserts.push(entry);
+        }
+        self.media.register(&scope, media).await;
+        let _ = self
+            .emit(Event::ThreadListChanged {
+                room_id: room_id.clone(),
+                upserts,
+            })
+            .await;
+    }
+
+    async fn emit(&self, event: Event) -> Result<(), Error> {
+        self.events
+            .send(event)
+            .await
+            .map_err(|_| Error::interrupted())
+    }
+}
+
+impl List {
+    fn position(&self, root_event_id: &OwnedEventId) -> Option<usize> {
+        self.entries
+            .iter()
+            .position(|entry| &entry.root_event_id == root_event_id)
+    }
+
+    fn upsert(&mut self, entry: ThreadListEntry) {
+        if let Some(index) = self.position(&entry.root_event_id) {
+            self.entries.remove(index);
+        }
+        let index = self
+            .entries
+            .partition_point(|known| known.sorts_before(&entry));
+        self.entries.insert(index, entry);
+    }
+}
+
+fn current<'a>(
+    lists: &'a mut HashMap<OwnedRoomId, List>,
+    room_id: &OwnedRoomId,
+    generation: u64,
+) -> Option<&'a mut List> {
+    lists
+        .get_mut(room_id)
+        .filter(|list| list.generation == generation)
+}
+
+fn not_open() -> Error {
+    Error::new(ErrorKind::TimelineNotOpen, "thread list is not open")
+}
+
+/// a reply newer than the thread's latest counts once. the event cache replays what it
+/// already had on a reset or a reload, and back pagination brings older ones, neither of
+/// which the server's count left out
+fn add_reply(entry: &mut ThreadListEntry, reply: Normalised) -> Vec<MediaRegistration> {
+    if entry.summary.latest_event_id.as_ref() == Some(&reply.event_id) {
+        entry.latest_reply = Some(reply.item);
+        return reply.media;
+    }
+    if reply.timestamp_ms < entry.summary.latest_event_timestamp_ms.unwrap_or(0) {
+        return Vec::new();
+    }
+    entry.summary.reply_count = entry.summary.reply_count.saturating_add(1);
+    entry.summary.latest_event_id = Some(reply.event_id);
+    entry.summary.latest_event_timestamp_ms = Some(reply.timestamp_ms);
+    entry.latest_activity_ms = entry.latest_activity_ms.max(reply.timestamp_ms);
+    entry.latest_reply = Some(reply.item);
+    entry.root = with_summary(entry.root.clone(), &entry.summary);
+    reply.media
+}
+
+fn with_summary(mut root: TimelineItem, summary: &ThreadSummary) -> TimelineItem {
+    if let TimelineItem::Event(event) = &mut root {
+        event.thread_summary = Some(summary.clone());
+    }
+    root
+}
+
+async fn fetch_page(
+    room: Room,
+    from: Option<String>,
+    limit: u16,
+    current_user_id: Option<OwnedUserId>,
+) -> Result<Page, Error> {
+    let roots = room
+        .list_threads(ListThreadsOptions {
+            from,
+            limit: Some(UInt::from(limit)),
+            ..ListThreadsOptions::default()
+        })
+        .await
+        .map_err(|source| Error::from_sdk("listing threads failed", source))?;
+    let mut entries = Vec::with_capacity(roots.chunk.len());
+    for root in roots.chunk {
+        if let Some(built) = build_entry(&room, root, current_user_id.as_deref()).await {
+            entries.push(built);
+        }
+    }
+    Ok(Page {
+        entries,
+        token: roots.prev_batch_token,
+    })
+}
+
+async fn fetch_root(
+    room: &Room,
+    root_event_id: &OwnedEventId,
+    current_user_id: Option<OwnedUserId>,
+) -> Option<Built> {
+    let root = room
+        .load_or_fetch_event(root_event_id, None)
+        .await
+        .inspect_err(
+            |error| tracing::debug!(%root_event_id, %error, "fetching a thread root failed"),
+        )
+        .ok()?;
+    build_entry(room, root, current_user_id.as_deref()).await
+}
+
+async fn build_entry(
+    room: &Room,
+    root: SdkTimelineEvent,
+    current_user_id: Option<&ruma::UserId>,
+) -> Option<Built> {
+    let summary = root.thread_summary.summary().cloned();
+    let latest = root.bundled_latest_thread_event.clone();
+    let root = normalise(room, &root, current_user_id).await?;
+    let latest = match latest {
+        Some(latest) => normalise(room, &latest, current_user_id).await,
+        None => None,
+    };
+    let summary = ThreadSummary {
+        reply_count: summary.as_ref().map_or(0, |summary| summary.num_replies),
+        latest_event_id: summary
+            .and_then(|summary| summary.latest_reply)
+            .or_else(|| latest.as_ref().map(|latest| latest.event_id.clone())),
+        latest_event_timestamp_ms: latest.as_ref().map(|latest| latest.timestamp_ms),
+    };
+    let mut media = root.media;
+    let latest_reply = latest.map(|latest| {
+        media.extend(latest.media);
+        latest.item
+    });
+    Some(Built {
+        entry: ThreadListEntry {
+            root_event_id: root.event_id,
+            latest_activity_ms: summary
+                .latest_event_timestamp_ms
+                .unwrap_or(root.timestamp_ms)
+                .max(root.timestamp_ms),
+            root: with_summary(root.item, &summary),
+            summary,
+            latest_reply,
+        },
+        media,
+    })
+}
+
+async fn normalise(
+    room: &Room,
+    event: &SdkTimelineEvent,
+    current_user_id: Option<&ruma::UserId>,
+) -> Option<Normalised> {
+    let event_id = event.event_id()?;
+    let value: serde_json::Value = serde_json::from_str(event.raw().json().get()).ok()?;
+    let identity = format!("{}\n{event_id}", room.room_id());
+    let (mut item, media) = normalise_observed(&value, current_user_id, &identity)?;
+    let TimelineItem::Event(normalised) = &mut item else {
+        return None;
+    };
+    normalised.highlight = event
+        .push_actions()
+        .is_some_and(|actions| actions.iter().any(Action::is_highlight));
+    if let Some(sender) = sender(room, &normalised.sender.user_id).await {
+        normalised.sender = sender;
+    }
+    // a preview doesn't chase what a root or a reply was replying to
+    resolve_replies(std::iter::once(&mut **normalised), &HashMap::new());
+    Some(Normalised {
+        event_id,
+        timestamp_ms: normalised.timestamp_ms,
+        item,
+        media,
+    })
+}
+
+async fn sender(room: &Room, user_id: &OwnedUserId) -> Option<Sender> {
+    let member = room.get_member(user_id).await.ok().flatten()?;
+    Some(Sender {
+        user_id: user_id.clone(),
+        display_name: member.display_name().map(str::to_owned),
+        avatar: member.avatar_url().map(MediaId::plain),
+        name_ambiguous: member.name_ambiguous(),
+    })
+}
+
+/// the room's event cache, followed for as long as the list is open. a lag loses whatever
+/// it skipped until the thread moves again
+async fn listen(
+    room: Room,
+    generation: u64,
+    current_user_id: Option<OwnedUserId>,
+    arrivals: mpsc::UnboundedSender<Arrival>,
+) {
+    let subscribed = async {
+        let (cache, drop_handles) = room.event_cache().await?;
+        let (_, subscriber) = cache.subscribe().await?;
+        matrix_sdk::event_cache::Result::Ok((drop_handles, subscriber))
+    }
+    .await;
+    let (_drop_handles, mut subscriber): (_, RoomEventCacheSubscriber) = match subscribed {
+        Ok(subscribed) => subscribed,
+        Err(error) => {
+            tracing::warn!(room_id = %room.room_id(), %error, "thread list can't follow the room");
+            return;
+        }
+    };
+    let room_id = room.room_id().to_owned();
+    loop {
+        let update = match subscriber.recv().await {
+            Ok(update) => update,
+            Err(RecvError::Lagged(skipped)) => {
+                tracing::warn!(%room_id, skipped, "thread list listener lagged");
+                continue;
+            }
+            Err(RecvError::Closed) => return,
+        };
+        let RoomEventCacheUpdate::UpdateTimelineEvents(update) = update else {
+            continue;
+        };
+        let paginated = matches!(update.origin, EventsOrigin::Pagination);
+        for event in added(update.diffs) {
+            let thread_root = extract_thread_root(event.raw());
+            if thread_root.is_none() && event.thread_summary.summary().is_none() {
+                continue;
+            }
+            let Some(normalised) =
+                normalise(&room, &event, current_user_id.as_deref()).await
+            else {
+                continue;
+            };
+            let live = match thread_root {
+                Some(root_event_id) => Live::Reply {
+                    root_event_id,
+                    reply: normalised,
+                    paginated,
+                },
+                None => Live::Root(normalised),
+            };
+            let arrival = Arrival::Live {
+                room_id: room_id.clone(),
+                generation,
+                live,
+            };
+            if arrivals.send(arrival).is_err() {
+                return;
+            }
+        }
+    }
+}
+
+fn added(diffs: Vec<VectorDiff<SdkTimelineEvent>>) -> Vec<SdkTimelineEvent> {
+    let mut events = Vec::new();
+    for diff in diffs {
+        match diff {
+            VectorDiff::Append { values } | VectorDiff::Reset { values } => events.extend(values),
+            VectorDiff::PushBack { value }
+            | VectorDiff::PushFront { value }
+            | VectorDiff::Insert { value, .. }
+            | VectorDiff::Set { value, .. } => events.push(value),
+            VectorDiff::Clear
+            | VectorDiff::PopBack
+            | VectorDiff::PopFront
+            | VectorDiff::Remove { .. }
+            | VectorDiff::Truncate { .. } => {}
+        }
+    }
+    events
+}
+
+#[cfg(test)]
+mod tests {
+    use ruma::{event_id, owned_event_id};
+
+    use super::*;
+
+    fn reply(event_id: &str, timestamp_ms: u64) -> Normalised {
+        Normalised {
+            event_id: event_id.try_into().unwrap(),
+            timestamp_ms,
+            item: TimelineItem::DayDivider { timestamp_ms },
+            media: Vec::new(),
+        }
+    }
+
+    fn entry(root: &str, latest_activity_ms: u64) -> ThreadListEntry {
+        ThreadListEntry {
+            root_event_id: root.try_into().unwrap(),
+            root: TimelineItem::DayDivider { timestamp_ms: 0 },
+            summary: ThreadSummary {
+                reply_count: 2,
+                latest_event_id: Some(owned_event_id!("$latest")),
+                latest_event_timestamp_ms: Some(latest_activity_ms),
+            },
+            latest_reply: None,
+            latest_activity_ms,
+        }
+    }
+
+    #[test]
+    fn a_reply_counts_once_and_only_when_newer() {
+        let mut thread = entry("$root", 20);
+        add_reply(&mut thread, reply("$latest", 20));
+        add_reply(&mut thread, reply("$older", 10));
+        assert_eq!(
+            thread.summary.reply_count, 2,
+            "a replayed or paginated reply was counted"
+        );
+        add_reply(&mut thread, reply("$newer", 30));
+        add_reply(&mut thread, reply("$newer", 30));
+        assert_eq!(thread.summary.reply_count, 3);
+        assert_eq!(thread.latest_activity_ms, 30);
+        assert_eq!(
+            thread.summary.latest_event_id.as_deref(),
+            Some(event_id!("$newer"))
+        );
+    }
+
+    #[test]
+    fn upserts_keep_newest_activity_first() {
+        let mut entries = imbl::Vector::new();
+        for (root, activity) in [("$a", 10), ("$b", 30), ("$c", 20), ("$d", 20)] {
+            ThreadListEntry::upsert(&mut entries, entry(root, activity));
+        }
+        ThreadListEntry::upsert(&mut entries, entry("$a", 40));
+        let order: Vec<_> = entries
+            .iter()
+            .map(|entry| entry.root_event_id.as_str())
+            .collect();
+        assert_eq!(order, ["$a", "$b", "$c", "$d"]);
+    }
+}

Modified file: src/lib.rs+7 additions−7 deletions@@ -51,8 +51,8 @@ pub use mentions_model::{
 };
 #[cfg(feature = "mirror")]
 pub use mirror::{
-    Mirror, MirrorHandle, SessionStatus, State, StateChange, Subscription, TimelineChange,
-    TimelineState, Update, UploadFailure, UploadState,
+    Mirror, MirrorHandle, SessionStatus, State, StateChange, Subscription, ThreadListChange,
+    ThreadListState, TimelineChange, TimelineState, Update, UploadFailure, UploadState,
 };
 pub use model::{
     AccountCommand, AttachmentEncryption, AttachmentUpload, Change, Command, CommandOutput,
@@ -68,11 +68,11 @@ pub use model::{
     RecentHistoryState, RecoveryState, ReplyDetails, ReplyReference, Retain, RichText,
     RoomCategory, RoomCommand, RoomHero, RoomListFilter, RoomMatcher, RoomMembership,
     RoomNotificationMode, RoomNotificationSettings, RoomPowerLevels, RoomPreset, RoomSummary,
-    RoomVisibility, SendState, Sender, SpaceChild, Termination, ThreadSummary, ThreadUnread,
-    TimelineCommand,
-    TimelineContent, TimelineDiff, TimelineDiffError, TimelineEvent, TimelineId, TimelineItem,
-    TimelineItemId, TypingState, UploadDetails, UploadOutcome, UploadThumbnail, UtdCause,
-    VerificationAction, VerificationFailure, VerificationState, VerificationTarget,
+    RoomVisibility, SendState, Sender, SpaceChild, Termination, ThreadListCommand, ThreadListEntry,
+    ThreadSummary, ThreadUnread, TimelineCommand, TimelineContent, TimelineDiff, TimelineDiffError,
+    TimelineEvent, TimelineId, TimelineItem, TimelineItemId, TypingState, UploadDetails,
+    UploadOutcome, UploadThumbnail, UtdCause, VerificationAction, VerificationFailure,
+    VerificationState, VerificationTarget,
 };
 pub use repair_model::{
     HistoryContinuity, HistoryGap, HistoryGapLifecycle, HistoryRepairCommand, HistoryRepairPhase,

Modified file: src/mirror.rs+78 additions−3 deletions@@ -20,9 +20,9 @@ use tokio::sync::broadcast;
 
 use crate::{
     ClientHandle, Command, ConnectionState, EncryptionState, EngineFailure, ErrorKind, Event,
-    EventContextCloseReason, HistoryRepairReport, HistoryState, ImagePacks, RecentHistoryState, RoomSummary,
-    TimelineCommand, TimelineDiff, TimelineId, TimelineItem, TypingState, UploadOutcome,
-    VerificationState, VerificationTarget, engine::executor::EngineBoxFuture,
+    EventContextCloseReason, HistoryRepairReport, HistoryState, ImagePacks, RecentHistoryState,
+    RoomSummary, ThreadListEntry, TimelineCommand, TimelineDiff, TimelineId, TimelineItem, TypingState,
+    UploadOutcome, VerificationState, VerificationTarget, engine::executor::EngineBoxFuture,
     transport::PublicationRegistration,
 };
 
@@ -50,6 +50,7 @@ pub struct State {
     pub rooms: Vector<RoomSummary>,
     pub image_packs: ImagePacks,
     pub timelines: imbl::HashMap<TimelineId, TimelineState>,
+    pub thread_lists: imbl::HashMap<OwnedRoomId, ThreadListState>,
     /// a room without an entry is idle. closing a room drops its entries here and in `repairs`
     pub recent_history: imbl::HashMap<OwnedRoomId, RecentHistoryState>,
     pub repairs: imbl::HashMap<OwnedRoomId, HistoryRepairReport>,
@@ -71,6 +72,14 @@ pub struct TimelineState {
     pub diverged: bool,
 }
 
+#[derive(Clone, Debug, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[non_exhaustive]
+pub struct ThreadListState {
+    pub entries: Vector<ThreadListEntry>,
+    pub history: HistoryState,
+}
+
 #[derive(Clone, Debug, Eq, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 #[non_exhaustive]
@@ -139,6 +148,10 @@ pub enum StateChange {
         timeline_id: TimelineId,
         state: HistoryState,
     },
+    ThreadList {
+        room_id: OwnedRoomId,
+        change: ThreadListChange,
+    },
     RecentHistory {
         room_id: OwnedRoomId,
         state: RecentHistoryState,
@@ -176,6 +189,20 @@ pub enum TimelineChange {
     Closed,
 }
 
+#[derive(Clone, Debug)]
+pub enum ThreadListChange {
+    Reset {
+        entries: Vector<ThreadListEntry>,
+    },
+    /// `entries` is what applying `upserts` left
+    Changed {
+        upserts: Arc<[ThreadListEntry]>,
+        entries: Vector<ThreadListEntry>,
+    },
+    History(HistoryState),
+    Closed,
+}
+
 /// drains the event stream into a `State` that any number of `MirrorHandle`s watch
 pub struct Mirror {
     handle: ClientHandle,
@@ -220,6 +247,7 @@ impl Mirror {
             rooms: Vector::new(),
             image_packs: ImagePacks::default(),
             timelines: imbl::HashMap::new(),
+            thread_lists: imbl::HashMap::new(),
             recent_history: imbl::HashMap::new(),
             repairs: imbl::HashMap::new(),
             typing: imbl::HashMap::new(),
@@ -414,6 +442,53 @@ impl Mirror {
             Event::EventContextClosed { context_id, reason } => {
                 self.publish(|_| Some(StateChange::EventContextClosed { context_id, reason }));
             }
+            Event::ThreadListReset { room_id, entries } => self.publish(|state| {
+                let history = state
+                    .thread_lists
+                    .get(&room_id)
+                    .map_or(HistoryState::Idle, |list| list.history.clone());
+                state.thread_lists.insert(
+                    room_id.clone(),
+                    ThreadListState {
+                        entries: entries.clone(),
+                        history,
+                    },
+                );
+                Some(StateChange::ThreadList {
+                    room_id,
+                    change: ThreadListChange::Reset { entries },
+                })
+            }),
+            Event::ThreadListChanged { room_id, upserts } => self.publish(|state| {
+                let list = state.thread_lists.get_mut(&room_id)?;
+                for entry in &upserts {
+                    ThreadListEntry::upsert(&mut list.entries, entry.clone());
+                }
+                Some(StateChange::ThreadList {
+                    room_id,
+                    change: ThreadListChange::Changed {
+                        upserts: upserts.into(),
+                        entries: list.entries.clone(),
+                    },
+                })
+            }),
+            Event::ThreadListHistoryChanged {
+                room_id,
+                state: history,
+            } => self.publish(|state| {
+                state.thread_lists.get_mut(&room_id)?.history = history.clone();
+                Some(StateChange::ThreadList {
+                    room_id,
+                    change: ThreadListChange::History(history),
+                })
+            }),
+            Event::ThreadListClosed { room_id } => self.publish(|state| {
+                state.thread_lists.remove(&room_id)?;
+                Some(StateChange::ThreadList {
+                    room_id,
+                    change: ThreadListChange::Closed,
+                })
+            }),
             Event::TypingChanged(typing) => self.publish(|state| {
                 if typing.users.is_empty() {
                     state.typing.remove(&typing.room_id);

Modified file: src/model.rs+85 additions@@ -31,6 +31,7 @@ pub enum Command {
     Lifecycle(LifecycleCommand),
     Pack(PackCommand),
     HistoryRepair(crate::HistoryRepairCommand),
+    ThreadList(ThreadListCommand),
 }
 
 #[derive(Clone, Debug, Eq, PartialEq)]
@@ -404,6 +405,24 @@ pub enum MediaCommand {
     Upload(AttachmentUpload),
 }
 
+/// a room's threads, newest activity first, kept current as they gain replies. the room
+/// needn't be open
+#[derive(Clone, Debug, Eq, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[non_exhaustive]
+pub enum ThreadListCommand {
+    /// answers once `ThreadListReset` has gone out, empty until a page arrives. opening an
+    /// open list sends a fresh reset
+    Open(OwnedRoomId),
+    /// the next `limit` threads from the server, at most 50. one page runs at a time and a
+    /// second ask waits for it
+    Paginate {
+        room_id: OwnedRoomId,
+        limit: u16,
+    },
+    Close(OwnedRoomId),
+}
+
 /// msc2545 pack editing. edits read the server's copy and keep keys macltok doesn't model
 #[derive(Clone, Debug, Eq, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
@@ -945,6 +964,25 @@ pub enum Event {
         context_id: String,
         reason: EventContextCloseReason,
     },
+    /// replaces whatever you had for this room's thread list, newest activity first
+    ThreadListReset {
+        room_id: OwnedRoomId,
+        entries: Vector<ThreadListEntry>,
+    },
+    /// apply each with `ThreadListEntry::upsert`
+    ThreadListChanged {
+        room_id: OwnedRoomId,
+        upserts: Vec<ThreadListEntry>,
+    },
+    /// `StartReached` once the oldest thread is in
+    ThreadListHistoryChanged {
+        room_id: OwnedRoomId,
+        state: HistoryState,
+    },
+    /// nothing more comes for it until another `ThreadListReset`
+    ThreadListClosed {
+        room_id: OwnedRoomId,
+    },
     /// a room that stops being watched says so with an empty `users`
     TypingChanged(TypingState),
     MediaUploadProgress {
@@ -1876,6 +1914,53 @@ pub struct ThreadSummary {
     pub latest_event_timestamp_ms: Option<u64>,
 }
 
+#[derive(Clone, Debug, Eq, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[non_exhaustive]
+pub struct ThreadListEntry {
+    pub root_event_id: OwnedEventId,
+    /// always an event, carrying `summary` as its own
+    pub root: TimelineItem,
+    pub summary: ThreadSummary,
+    pub latest_reply: Option<TimelineItem>,
+    /// the latest reply's timestamp, or the root's before any reply
+    pub latest_activity_ms: u64,
+}
+
+impl ThreadListEntry {
+    /// replaces the entry with the same root, or adds it, where its activity puts it.
+    /// ties go by root event id
+    pub fn upsert(entries: &mut Vector<Self>, entry: Self) {
+        if let Some(index) = entries
+            .iter()
+            .position(|known| known.root_event_id == entry.root_event_id)
+        {
+            entries.remove(index);
+        }
+        let index = entries
+            .binary_search_by(|known| {
+                if known.sorts_before(&entry) {
+                    std::cmp::Ordering::Less
+                } else {
+                    std::cmp::Ordering::Greater
+                }
+            })
+            .unwrap_or_else(|index| index);
+        entries.insert(index, entry);
+    }
+
+    #[must_use]
+    pub fn sorts_before(&self, other: &Self) -> bool {
+        (
+            std::cmp::Reverse(self.latest_activity_ms),
+            &self.root_event_id,
+        ) < (
+            std::cmp::Reverse(other.latest_activity_ms),
+            &other.root_event_id,
+        )
+    }
+}
+
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 pub enum SendState {

Modified file: src/transport.rs+26 additions−2 deletions@@ -36,8 +36,8 @@ use crate::{
     NotificationCommand, OwnDevice, PackCommand, PackImageInfo, PackUsage, Patch,
     PowerLevelChanges, Profile, PushNotification, PushNotificationEvent, PushNotificationStatus,
     RoomCommand, RoomListFilter, RoomNotificationMode, RoomNotificationSettings, RoomPowerLevels,
-    RoomSummary, SavedMentionContext, SpaceChild, TimelineCommand, TimelineId, VerificationAction,
-    VerificationTarget,
+    RoomSummary, SavedMentionContext, SpaceChild, ThreadListCommand, TimelineCommand, TimelineId,
+    VerificationAction, VerificationTarget,
     encryption::encryption_state,
     engine::{MediaService, StoreLock},
     model::{PushNotificationRoom, PushNotificationSender},
@@ -304,6 +304,30 @@ impl ClientHandle {
             .await
     }
 
+    /// the room's threads, newest activity first, kept current as they gain replies.
+    /// the room needn't be open. it starts empty, `paginate_thread_list` fills it
+    pub async fn open_thread_list(&self, room_id: OwnedRoomId) -> Result<(), Error> {
+        self.run(Command::ThreadList(ThreadListCommand::Open(room_id)))
+            .await
+    }
+
+    pub async fn paginate_thread_list(
+        &self,
+        room_id: OwnedRoomId,
+        limit: u16,
+    ) -> Result<(), Error> {
+        self.run(Command::ThreadList(ThreadListCommand::Paginate {
+            room_id,
+            limit,
+        }))
+        .await
+    }
+
+    pub async fn close_thread_list(&self, room_id: OwnedRoomId) -> Result<(), Error> {
+        self.run(Command::ThreadList(ThreadListCommand::Close(room_id)))
+            .await
+    }
+
     /// reads the server's list so a pin made elsewhere just before isn't dropped
     pub async fn set_pinned(
         &self,

macltok

Commit 66db1b3cd895

Commit `66db1b3cd895`