macltok

public

Star0

Fork 0

Watch0

87 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `e5c6427adb35`

tarball

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

8 files changed+374-20

rooms: count unread thread replies per root

With threading support on, the sdk leaves thread events out of RoomReadReceipts entirely, so thread mentions vanished from unread_mentions and nothing said a thread had unread replies. RoomSummary::thread_unreads counts them per root against threaded (MSC3771) and unthreaded receipts, and the room list watcher recounts on event cache updates and receipts since neither touches room info.

Modified file: docs/design-decisions.md+13 additions@@ -334,6 +334,19 @@ membership until it's forgotten. push rule changes never touch room info, so
 `NotificationSettingsChanged` brings a `RoomsChanged` along for the modes in
 the summaries
 
+with threads on, the sdk leaves thread events out of a room's counts and keeps
+none of its own for them. `thread_unreads` works them out per root the way the
+sdk does a room's, over the events its cache holds in memory, newest first until
+a receipt for that thread (msc3771), an unthreaded one, or the user's own reply.
+a main-timeline receipt doesn't reach into threads. a thread's mentions also count
+in the room's `unread_mentions`, so nobody misses one, while its messages and
+notifications stay apart, since they don't move the main timeline's read marker.
+thread events and threaded receipts leave room info alone, so the room list
+watcher also recounts a room on each event cache update and each receipt, and
+only says `RoomsChanged` when the counts moved. counting in the sdk would be
+cheaper and survive memory being trimmed, but that's a behaviour change in the
+fork
+
 ## link previews
 
 a text message can carry its sender's own previews (msc4095), and those come

Modified file: src/engine/watchers.rs+62 additions−13 deletions@@ -1,31 +1,34 @@
 use std::{
-    collections::{HashSet, VecDeque},
+    collections::{BTreeMap, HashMap, HashSet, VecDeque},
     sync::Arc,
     time::Duration,
 };
 
 use futures_util::StreamExt;
 use matrix_sdk::{
-    Client,
+    Client, Room,
     ruma::events::{AnyMessageLikeEventContent, room::message::MessageType},
     send_queue::{LocalEchoContent, RoomSendQueueUpdate, SendQueueUpdate},
 };
 use matrix_sdk_base::RoomInfoNotableUpdateReasons;
 use matrix_sdk_ui::sync_service::{State as SyncState, SyncService};
 use ruma::{
-    OwnedRoomId, OwnedTransactionId,
-    events::{AnyGlobalAccountDataEvent, AnyRoomAccountDataEvent, AnySyncStateEvent},
+    OwnedEventId, OwnedRoomId, OwnedTransactionId,
+    events::{
+        AnyGlobalAccountDataEvent, AnyRoomAccountDataEvent, AnySyncStateEvent,
+        receipt::SyncReceiptEvent,
+    },
     serde::Raw,
     time::Instant,
 };
-use tokio::sync::{Notify, broadcast::error::RecvError, watch};
+use tokio::sync::{Notify, broadcast::error::RecvError, mpsc, watch};
 
 use super::executor::TaskSet;
 use crate::{
-    ConnectionState, Event, UploadOutcome,
+    ConnectionState, Event, ThreadUnread, UploadOutcome,
     encryption::encryption_state,
     error::classify_sdk_error,
-    normalise::{ENABLED_ROOMS, ROOM_PACKS, USER_PACK},
+    normalise::{ENABLED_ROOMS, ROOM_PACKS, USER_PACK, thread_unreads},
     space_order,
     transport::EventSender,
 };
@@ -433,25 +436,71 @@ pub(super) async fn rooms(client: Client, events: EventSender) {
             }
         }),
     ];
-    loop {
+    let (receipts, mut receipt_rooms) = mpsc::unbounded_channel();
+    let receipt_handler = client.add_event_handler(move |_: SyncReceiptEvent, room: Room| {
+        let _ = receipts.send(room.room_id().to_owned());
+        async {}
+    });
+    let mut timelines = client.event_cache().subscribe_to_room_generic_updates();
+    let mut thread_unreads = ThreadUnreads::default();
+    'invalidate: loop {
         if events.send(Event::RoomsChanged).await.is_err() {
             break;
         }
-        tokio::select! {
-            update = rooms.next() => if update.is_none() { break },
-            update = info.recv() => if matches!(update, Err(RecvError::Closed)) { break },
-            () = changed.notified() => {}
+        // thread events and threaded receipts leave room info alone, so neither reaches `info`
+        loop {
+            let room_id = tokio::select! {
+                update = rooms.next() => if update.is_none() { break 'invalidate } else { break },
+                update = info.recv() => {
+                    if matches!(update, Err(RecvError::Closed)) { break 'invalidate }
+                    break;
+                }
+                () = changed.notified() => break,
+                Some(room_id) = receipt_rooms.recv() => room_id,
+                update = timelines.recv() => match update {
+                    Ok(update) => update.room_id,
+                    Err(RecvError::Lagged(_)) => {
+                        thread_unreads.0.clear();
+                        break;
+                    }
+                    Err(RecvError::Closed) => break 'invalidate,
+                },
+            };
+            if thread_unreads.changed(&client, room_id).await {
+                break;
+            }
         }
         // a sync touching fifty rooms is one invalidation, not fifty
         while let Ok(_) | Err(tokio::sync::broadcast::error::TryRecvError::Lagged(_)) =
             info.try_recv()
         {}
     }
-    for handler in handlers {
+    for handler in handlers.into_iter().chain([receipt_handler]) {
         client.remove_event_handler(handler);
     }
 }
 
+#[derive(Default)]
+struct ThreadUnreads(HashMap<OwnedRoomId, BTreeMap<OwnedEventId, ThreadUnread>>);
+
+impl ThreadUnreads {
+    async fn changed(&mut self, client: &Client, room_id: OwnedRoomId) -> bool {
+        let Some(room) = client.get_room(&room_id) else {
+            return self.0.remove(&room_id).is_some();
+        };
+        let unreads = thread_unreads(&room).await;
+        if self
+            .0
+            .get(&room_id)
+            .map_or(unreads.is_empty(), |known| *known == unreads)
+        {
+            return false;
+        }
+        self.0.insert(room_id, unreads);
+        true
+    }
+}
+
 /// packs live in room state and account data, and leaving a room takes its packs with it
 pub(super) async fn image_packs(client: Client, events: EventSender) {
     let changed = Arc::new(Notify::new());

Modified file: src/lib.rs+2 additions−1 deletions@@ -68,7 +68,8 @@ 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, TimelineCommand,
+    RoomVisibility, SendState, Sender, SpaceChild, Termination, ThreadSummary, ThreadUnread,
+    TimelineCommand,
     TimelineContent, TimelineDiff, TimelineDiffError, TimelineEvent, TimelineId, TimelineItem,
     TimelineItemId, TypingState, UploadDetails, UploadOutcome, UploadThumbnail, UtdCause,
     VerificationAction, VerificationFailure, VerificationState, VerificationTarget,

Modified file: src/model.rs+16 additions−2 deletions@@ -1,4 +1,4 @@
-use std::{fmt, sync::Arc};
+use std::{collections::BTreeMap, fmt, sync::Arc};
 
 use imbl::Vector;
 use ruma::{
@@ -1047,10 +1047,15 @@ pub struct RoomSummary {
     pub room_id: OwnedRoomId,
     pub name: String,
     pub membership: RoomMembership,
+    /// the main timeline's, threads have theirs in `thread_unreads`
     pub unread_messages: u64,
+    /// thread mentions included
     pub unread_mentions: u64,
-    /// unread events the push rules said to notify about, which is neither of the above
+    /// unread events the push rules said to notify about, which is neither of the above.
+    /// the main timeline's
     pub unread_notifications: u64,
+    /// threads with anything unread, by root
+    pub thread_unreads: BTreeMap<OwnedEventId, ThreadUnread>,
     pub encrypted: bool,
     pub space: bool,
     pub direct: bool,
@@ -1083,6 +1088,15 @@ pub struct RoomSummary {
     pub can_pin: bool,
 }
 
+#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[non_exhaustive]
+pub struct ThreadUnread {
+    pub messages: u64,
+    pub mentions: u64,
+    pub notifications: u64,
+}
+
 #[derive(Clone, Debug, Eq, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 #[non_exhaustive]

Modified file: src/normalise/mod.rs+2

@@ -5,6 +5,7 @@ mod media;

55 mod member;

66 mod observed;

77 mod room;

8+mod thread;

89 mod timeline;

910

1011 pub(crate) use context::{apply_profiles, normalise_context, resolve_replies};

@@ -18,4 +19,5 @@ pub(crate) use member::member_summary;

1819 pub(crate) use observed::normalise_observed;

1920 pub(crate) use room::notification_mode;

2021 pub(crate) use room::{room_summaries, space_child, stored_space_order};

22+pub(crate) use thread::thread_unreads;

2123 pub(crate) use timeline::{message_kind, normalise_timeline, normalise_timeline_diffs};

Modified file: src/normalise/room.rs+7 additions−1 deletions@@ -62,13 +62,19 @@ async fn room_summary(room: &Room, spaces: &SpaceService) -> RoomSummary {
     } else {
         Vec::new()
     };
+    let thread_unreads = super::thread_unreads(room).await;
     RoomSummary {
         room_id: room.room_id().to_owned(),
         name,
         membership: membership(room.state()),
         unread_messages: room.num_unread_messages(),
-        unread_mentions: room.num_unread_mentions(),
+        unread_mentions: room.num_unread_mentions()
+            + thread_unreads
+                .values()
+                .map(|unread| unread.mentions)
+                .sum::<u64>(),
         unread_notifications: room.num_unread_notifications(),
+        thread_unreads,
         encrypted: room.encryption_state().is_encrypted(),
         space: room.is_space(),
         direct: room.is_direct().await.unwrap_or(false),

Added file: src/normalise/thread.rs+133 additions@@ -0,0 +1,133 @@
+use std::collections::{BTreeMap, HashMap};
+
+use matrix_sdk::{
+    Room, RoomState, deserialized_responses::TimelineEvent, serde_helpers::extract_thread_root,
+};
+use ruma::{
+    OwnedEventId, UserId,
+    events::receipt::{ReceiptThread, ReceiptType},
+};
+use serde_json::Value;
+
+use crate::ThreadUnread;
+
+struct Reply {
+    event_id: Option<OwnedEventId>,
+    own: bool,
+    counts: bool,
+    notify: bool,
+    highlight: bool,
+}
+
+impl Reply {
+    fn of(event: &TimelineEvent, own_user_id: &UserId) -> Self {
+        let own = event.sender().as_deref() == Some(own_user_id);
+        let counts = !own && counts_as_unread(event);
+        let actions = event.push_actions().unwrap_or_default();
+        Self {
+            event_id: event.event_id(),
+            own,
+            counts,
+            notify: counts && actions.iter().any(|action| action.should_notify()),
+            highlight: counts && actions.iter().any(|action| action.is_highlight()),
+        }
+    }
+}
+
+/// the sdk leaves thread events out of the room's counts once threads are on and keeps none
+/// of its own for them. so these are worked out the same way over the events it has in
+/// memory, newest first until a receipt for that thread (msc3771), an unthreaded receipt or
+/// the user's own reply. a receipt older than memory counts everything in memory, as the
+/// sdk's does
+pub(crate) async fn thread_unreads(room: &Room) -> BTreeMap<OwnedEventId, ThreadUnread> {
+    if room.state() != RoomState::Joined {
+        return BTreeMap::new();
+    }
+    let own_user_id = room.own_user_id();
+    let unthreaded = own_receipts(room, ReceiptThread::Unthreaded).await;
+    let Ok((cache, _drop_handles)) = room.event_cache().await else {
+        return BTreeMap::new();
+    };
+    let mut threads: HashMap<OwnedEventId, Vec<Reply>> = HashMap::new();
+    let scanned = cache
+        .rfind_map_event_in_memory_by(|event| {
+            let event_id = event.event_id();
+            if event_id.is_some_and(|event_id| unthreaded.contains(&event_id)) {
+                return Some(());
+            }
+            if let Some(root) = extract_thread_root(event.raw()) {
+                threads
+                    .entry(root)
+                    .or_default()
+                    .push(Reply::of(event, own_user_id));
+            }
+            None
+        })
+        .await;
+    if let Err(error) = scanned {
+        tracing::debug!(room_id = %room.room_id(), %error, "reading events for thread unreads failed");
+        return BTreeMap::new();
+    }
+    let mut unreads = BTreeMap::new();
+    for (root, replies) in threads {
+        let read = own_receipts(room, ReceiptThread::Thread(root.clone())).await;
+        let mut unread = ThreadUnread::default();
+        for reply in replies {
+            if reply.own
+                || reply
+                    .event_id
+                    .as_ref()
+                    .is_some_and(|event_id| read.contains(event_id))
+            {
+                break;
+            }
+            unread.messages += u64::from(reply.counts);
+            unread.notifications += u64::from(reply.notify);
+            unread.mentions += u64::from(reply.highlight);
+        }
+        if unread != ThreadUnread::default() {
+            unreads.insert(root, unread);
+        }
+    }
+    unreads
+}
+
+async fn own_receipts(room: &Room, thread: ReceiptThread) -> Vec<OwnedEventId> {
+    let mut found = Vec::new();
+    for receipt_type in [ReceiptType::Read, ReceiptType::ReadPrivate] {
+        match room
+            .load_user_receipt(receipt_type, thread.clone(), room.own_user_id())
+            .await
+        {
+            Ok(Some((event_id, _))) => found.push(event_id),
+            Ok(None) => {}
+            Err(error) => {
+                tracing::debug!(room_id = %room.room_id(), %error, "reading a receipt failed");
+            }
+        }
+    }
+    found
+}
+
+/// the sdk's rule for its room counts, which is private to it
+fn counts_as_unread(event: &TimelineEvent) -> bool {
+    let raw = event.raw();
+    let redacted = raw
+        .get_field::<Value>("unsigned")
+        .ok()
+        .flatten()
+        .is_some_and(|unsigned| unsigned.get("redacted_because").is_some());
+    let message_like = matches!(
+        raw.get_field::<String>("type").ok().flatten().as_deref(),
+        Some(
+            "m.room.message"
+                | "m.room.encrypted"
+                | "m.sticker"
+                | "m.poll.start"
+                | "org.matrix.msc3381.poll.start"
+                | "m.poll.end"
+                | "org.matrix.msc3381.poll.end"
+        )
+    );
+    message_like && !redacted
+}

Modified file: tests/state.rs+139 additions−3 deletions@@ -4,13 +4,23 @@ use std::time::Duration;
 
 use macltok::{
     ConnectionState, Engine, EngineConfig, EngineError, Event, PackUsage, Retain, RoomMembership,
-    Termination,
+    Termination, ThreadUnread,
 };
 use matrix_sdk::{
-    ruma::{api::client::sync::sync_events::v5, events::AnySyncStateEvent, room_id, serde::Raw},
+    ruma::{
+        RoomId,
+        api::client::sync::sync_events::v5,
+        event_id,
+        events::{
+            AnySyncStateEvent, Mentions,
+            receipt::{ReceiptThread, ReceiptType},
+        },
+        room_id,
+        serde::Raw,
+    },
     test_utils::mocks::MatrixMockServer,
 };
-use matrix_sdk_test::JoinedRoomBuilder;
+use matrix_sdk_test::{ALICE, JoinedRoomBuilder, event_factory::EventFactory};
 use serde_json::json;
 use tokio::sync::watch;
 
@@ -233,3 +243,129 @@ async fn image_packs_follow_room_state_and_account_data() {
 
     support::shut_down(&handle, &mut events, running).await;
 }
+
+/// the sdk counts none of this once threads are on, and a threaded receipt never reaches
+/// room info, so only our own recount notices it
+#[tokio::test]
+async fn thread_unreads_follow_threaded_receipts() {
+    let server = MatrixMockServer::new().await;
+    server
+        .mock_sliding_sync()
+        .ok(v5::Response::new("1".to_owned()))
+        .mount()
+        .await;
+    let client = server.client_builder().build().await;
+    let room_id = room_id!("!threads:localhost");
+    server.mock_room_state_encryption().plain().mount().await;
+    server.sync_joined_room(&client, room_id).await;
+    client.event_cache().subscribe().unwrap();
+    let config = EngineConfig::default().with_initial_history_limit(0);
+    let (engine, handle, mut events) = Engine::from_client(client.clone(), config).await.unwrap();
+    let running = tokio::spawn(engine.run());
+    let own = client.user_id().unwrap().to_owned();
+
+    let (root, first, second) = (
+        event_id!("$root"),
+        event_id!("$first"),
+        event_id!("$second"),
+    );
+    let factory = EventFactory::new().room(room_id).sender(&ALICE);
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(factory.text_msg("root").event_id(root))
+                .add_timeline_event(
+                    factory
+                        .text_msg("you there?")
+                        .in_thread(root, root)
+                        .mentions(Mentions::with_user_ids([own.clone()]))
+                        .event_id(first),
+                )
+                .add_timeline_event(
+                    factory
+                        .text_msg("hello?")
+                        .in_thread(root, first)
+                        .event_id(second),
+                ),
+        )
+        .await;
+    let unread = thread_unread_once(&handle, &mut events, room_id, |unread| {
+        unread.is_some_and(|unread| unread.messages == 2)
+    })
+    .await;
+    assert_eq!(unread.mentions, 1);
+    let summary = room(&handle, room_id).await;
+    assert!(
+        summary.unread_mentions >= 1,
+        "the thread mention left the room's count"
+    );
+
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id).add_receipt(
+                factory
+                    .read_receipts()
+                    .add(
+                        first,
+                        &own,
+                        ReceiptType::Read,
+                        ReceiptThread::Thread(root.to_owned()),
+                    )
+                    .into_event(),
+            ),
+        )
+        .await;
+    let unread = thread_unread_once(&handle, &mut events, room_id, |unread| {
+        unread.is_some_and(|unread| unread.messages == 1)
+    })
+    .await;
+    assert_eq!(unread.mentions, 0);
+
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id).add_receipt(
+                factory
+                    .read_receipts()
+                    .add(second, &own, ReceiptType::Read, ReceiptThread::Unthreaded)
+                    .into_event(),
+            ),
+        )
+        .await;
+    thread_unread_once(&handle, &mut events, room_id, |unread| unread.is_none()).await;
+
+    support::shut_down(&handle, &mut events, running).await;
+}
+
+async fn room(handle: &macltok::ClientHandle, room_id: &RoomId) -> macltok::RoomSummary {
+    handle
+        .rooms()
+        .await
+        .into_iter()
+        .find(|summary| summary.room_id == room_id)
+        .expect("synced room was absent from the room summaries")
+}
+
+/// waits out `RoomsChanged` until the root's counts satisfy `done`
+async fn thread_unread_once(
+    handle: &macltok::ClientHandle,
+    events: &mut macltok::EventStream,
+    room_id: &RoomId,
+    done: impl Fn(Option<&ThreadUnread>) -> bool,
+) -> ThreadUnread {
+    let root = event_id!("$root");
+    tokio::time::timeout(Duration::from_secs(10), async {
+        loop {
+            let summary = room(handle, room_id).await;
+            let unread = summary.thread_unreads.get(root);
+            if done(unread) {
+                return unread.copied().unwrap_or_default();
+            }
+            support::receive_until(events, |event| matches!(event, Event::RoomsChanged)).await;
+        }
+    })
+    .await
+    .expect("thread unreads never settled")
+}

macltok

Commit e5c6427adb35

Commit `e5c6427adb35`