macltok

public

Star0

Fork 0

Watch0

48 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `fa5b4e09943a`

tarball

atagen <boss@atagen.co>2026-10-24 09:36 UTC

6 files changed+213-75

timeline: show a room's pins as their own timeline and add set_pinned

Pinned events open as TimelineId::Pinned on the sdk's PinnedEvents focus instead of a one-shot list, because a MediaId only resolves while the timeline that registered it is open, and a list would hand back attachments nobody can load. Threads and pins now share the off-actor build and queue, keyed by TimelineId. set_pinned rewrites m.room.pinned_events from the server's copy so a pin made elsewhere a moment earlier survives, and pin and power level changes now refetch summaries since neither raises a notable room info update.

Modified file: src/engine/mutations.rs+52 additions−1 deletions@@ -15,6 +15,7 @@ use matrix_sdk::{
             },
             state::get_state_event_for_key,
         },
+        api::error::ErrorKind as ApiErrorKind,
         events::{
             AnyGlobalAccountDataEventContent, GlobalAccountDataEventType, MessageLikeEventType,
             StateEventType,
@@ -24,7 +25,7 @@ use matrix_sdk::{
         serde::{JsonObject, Raw},
     },
 };
-use ruma::{OwnedRoomId, OwnedUserId};
+use ruma::{EventId, OwnedRoomId, OwnedUserId};
 use serde_json::Value as JsonValue;
 use tokio::sync::mpsc;
 
@@ -184,6 +185,7 @@ fn key(mutation: &Mutation, server: Option<&ruma::ServerName>) -> Option<Key> {
             | RoomCommand::SetUserPowerLevel { room_id, .. }
             | RoomCommand::MuteUser { room_id, .. }
             | RoomCommand::SetFavourite { room_id, .. }
+            | RoomCommand::SetPinned { room_id, .. }
             | RoomCommand::SetOwnDisplayName { room_id, .. }
             | RoomCommand::SetOwnAvatar { room_id, .. }
             | RoomCommand::SetAccountData { room_id, .. },
@@ -334,6 +336,13 @@ async fn execute(
         } => mute_user(client, &room(&room_id)?, &user_id, muted)
             .await
             .map(done),
+        RoomCommand::SetPinned {
+            room_id,
+            event_id,
+            pinned,
+        } => set_pinned(client, &room(&room_id)?, &event_id, pinned)
+            .await
+            .map(done),
         RoomCommand::SetFavourite { room_id, favourite } => room(&room_id)?
             .set_is_favourite(favourite, None)
             .await
@@ -644,6 +653,48 @@ async fn set_own_member_field(
     .map_err(|source| Error::from_sdk("updating your member event failed", source))
 }
 
+/// against the server's list, since the store's lags a sync and another client may have just pinned
+async fn set_pinned(
+    client: &Client,
+    room: &Room,
+    event_id: &EventId,
+    pinned: bool,
+) -> Result<(), Error> {
+    let request = get_state_event_for_key::v3::Request::new(
+        room.room_id().to_owned(),
+        StateEventType::RoomPinnedEvents,
+        String::new(),
+    );
+    let mut content: JsonObject = match client.send(request).await {
+        Ok(response) => {
+            serde_json::from_str(response.event_or_content.get()).map_err(|source| {
+                Error::with_source(ErrorKind::Server, "the pinned list is malformed", source)
+            })?
+        }
+        Err(error) if error.client_api_error_kind() == Some(&ApiErrorKind::NotFound) => {
+            JsonObject::new()
+        }
+        Err(source) => return Err(Error::from_http("reading the pinned list failed", source)),
+    };
+    let list = content
+        .entry("pinned")
+        .or_insert_with(|| JsonValue::Array(Vec::new()));
+    if !list.is_array() {
+        *list = JsonValue::Array(Vec::new());
+    }
+    let list = list.as_array_mut().expect("just made it an array");
+    let present = list.iter().any(|id| id.as_str() == Some(event_id.as_str()));
+    match (pinned, present) {
+        (true, true) | (false, false) => return Ok(()),
+        (true, false) => list.push(event_id.as_str().into()),
+        (false, true) => list.retain(|id| id.as_str() != Some(event_id.as_str())),
+    }
+    room.send_state_event_raw("m.room.pinned_events", "", JsonValue::Object(content))
+        .await
+        .map(drop)
+        .map_err(|source| Error::from_sdk("updating the pinned list failed", source))
+}
+
 fn done<T>(_: T) -> CommandOutput {
     CommandOutput::Done
 }

Modified file: src/engine/room.rs+73 additions−56 deletions@@ -2,7 +2,7 @@ use std::collections::{HashMap, VecDeque};
 
 use matrix_sdk::{Room, RoomState};
 use matrix_sdk_ui::timeline::{RoomExt, Timeline, TimelineFocus, TimelineReadReceiptTracking};
-use ruma::{OwnedEventId, OwnedRoomId};
+use ruma::OwnedRoomId;
 use tokio::sync::mpsc;
 
 use super::executor::{AbortHandle, JoinError, TaskSet, is_panic, resume_unwind};
@@ -222,28 +222,52 @@ struct Timelines {
 impl Timelines {
     async fn execute(&mut self, command: TimelineCommand, reply: Responder, room: &Room) {
         match command {
-            TimelineCommand::OpenThread { root_event_id, .. } => {
-                self.threads.open(root_event_id, reply, room);
+            TimelineCommand::OpenThread {
+                room_id,
+                root_event_id,
+            } => self
+                .threads
+                .open(TimelineId::thread(room_id, root_event_id), reply, room),
+            TimelineCommand::CloseThread {
+                room_id,
+                root_event_id,
+            } => {
+                let id = TimelineId::thread(room_id, root_event_id);
+                reply.done(self.threads.close_one(&id).await);
             }
-            TimelineCommand::CloseThread { root_event_id, .. } => {
-                reply.done(self.threads.close_one(&root_event_id).await);
+            TimelineCommand::OpenPinned(room_id) => {
+                self.threads.open(TimelineId::Pinned(room_id), reply, room);
+            }
+            TimelineCommand::ClosePinned(room_id) => {
+                reply.done(self.threads.close_one(&TimelineId::Pinned(room_id)).await);
             }
             TimelineCommand::Resync {
                 timeline_id: TimelineId::Room(_),
             } => reply.done(self.room.resync().await),
             command => {
-                let root = command
+                let Some(id) = command
                     .timeline_id()
-                    .and_then(TimelineId::thread_root)
-                    .cloned();
-                let Some(root) = root else {
+                    .filter(|id| !matches!(id, TimelineId::Room(_)))
+                    .cloned()
+                else {
                     return self.room.execute(command, reply, &mut self.work).await;
                 };
+                if matches!(id, TimelineId::Pinned(_))
+                    && matches!(
+                        command,
+                        TimelineCommand::Send { .. } | TimelineCommand::Paginate { .. }
+                    )
+                {
+                    return reply.done(Err(Error::new(
+                        ErrorKind::Invalid,
+                        "the pinned timeline can't send or paginate",
+                    )));
+                }
                 if let TimelineCommand::MarkRead {
                     timeline_id,
                     event_id,
                 } = &command
-                    && !self.threads.open.contains_key(&root)
+                    && !self.threads.open.contains_key(&id)
                 {
                     let client = room.client();
                     let (timeline_id, event_id) = (timeline_id.clone(), event_id.clone());
@@ -254,7 +278,7 @@ impl Timelines {
                     });
                     return;
                 }
-                match self.threads.route(root, command, reply) {
+                match self.threads.route(id, command, reply) {
                     Routed::Open(session, TimelineCommand::Resync { .. }, reply) => {
                         reply.done(session.resync().await);
                     }
@@ -273,31 +297,25 @@ impl Timelines {
         room_id: &OwnedRoomId,
         context: &SessionContext,
     ) {
-        let Some((root, queued, timeline)) = self.threads.built(built) else {
+        let Some((id, queued, timeline)) = self.threads.built(built) else {
             return;
         };
         let timeline = match timeline {
             Ok(timeline) => timeline,
             Err(error) => {
-                tracing::warn!(%error, %room_id, %root, "thread timeline build failed");
-                let error =
-                    Error::with_source(ErrorKind::Timeline, "thread timeline build failed", error);
+                tracing::warn!(%error, %room_id, ?id, "timeline build failed");
+                let error = Error::with_source(ErrorKind::Timeline, "timeline build failed", error);
                 for (_, reply) in queued {
                     reply.done(Err(error.clone()));
                 }
                 return;
             }
         };
-        let id = TimelineId::thread(room_id.clone(), root.clone());
-        let Ok(session) = TimelineSession::start(id, timeline, context.clone()).await else {
+        let Ok(session) = TimelineSession::start(id.clone(), timeline, context.clone()).await
+        else {
             return;
         };
-        let session = self
-            .threads
-            .open
-            .entry(root)
-            .insert_entry(session)
-            .into_mut();
+        let session = self.threads.open.entry(id).insert_entry(session).into_mut();
         for (command, reply) in queued {
             match command {
                 None => reply.done(Ok(())),
@@ -309,7 +327,7 @@ impl Timelines {
 }
 
 type Built = (
-    OwnedEventId,
+    TimelineId,
     u64,
     Result<Timeline, matrix_sdk_ui::timeline::Error>,
 );
@@ -328,9 +346,10 @@ struct OpeningThread {
     queued: Queued,
 }
 
+/// threads and the pinned timeline, which build off the actor and queue what arrives meanwhile
 struct Threads {
-    open: HashMap<OwnedEventId, TimelineSession>,
-    opening: HashMap<OwnedEventId, OpeningThread>,
+    open: HashMap<TimelineId, TimelineSession>,
+    opening: HashMap<TimelineId, OpeningThread>,
     builds: TaskSet<Built>,
     next_generation: u64,
     queued: usize,
@@ -349,40 +368,43 @@ impl Threads {
         }
     }
 
-    fn enqueue(&mut self, root: &OwnedEventId, command: Option<TimelineCommand>, reply: Responder) {
+    fn enqueue(&mut self, id: &TimelineId, command: Option<TimelineCommand>, reply: Responder) {
         if self.queued >= self.capacity {
             return reply.done(Err(Error::new(
                 ErrorKind::QueueFull,
                 "thread command queue is full",
             )));
         }
-        let Some(opening) = self.opening.get_mut(root) else {
+        let Some(opening) = self.opening.get_mut(id) else {
             return reply.done(Err(Error::new(
                 ErrorKind::TimelineNotOpen,
-                "thread timeline is not open",
+                "timeline is not open",
             )));
         };
         opening.queued.push_back((command, reply));
         self.queued += 1;
     }
 
-    fn open(&mut self, root: OwnedEventId, reply: Responder, room: &Room) {
-        if self.open.contains_key(&root) {
+    fn open(&mut self, id: TimelineId, reply: Responder, room: &Room) {
+        if self.open.contains_key(&id) {
             return reply.done(Ok(()));
         }
-        if !self.opening.contains_key(&root) {
+        if !self.opening.contains_key(&id) {
             let generation = self.next_generation;
             self.next_generation = self.next_generation.wrapping_add(1);
             let room = room.clone();
-            let build_root = root.clone();
+            let build_id = id.clone();
             let build = self.builds.spawn(async move {
-                let focus = TimelineFocus::Thread {
-                    root_event_id: build_root.clone(),
+                let focus = match &build_id {
+                    TimelineId::Thread { root_event_id, .. } => TimelineFocus::Thread {
+                        root_event_id: root_event_id.clone(),
+                    },
+                    _ => TimelineFocus::PinnedEvents,
                 };
-                (build_root, generation, build_timeline(&room, focus).await)
+                (build_id, generation, build_timeline(&room, focus).await)
             });
             self.opening.insert(
-                root.clone(),
+                id.clone(),
                 OpeningThread {
                     generation,
                     build,
@@ -390,20 +412,15 @@ impl Threads {
                 },
             );
         }
-        self.enqueue(&root, None, reply);
+        self.enqueue(&id, None, reply);
     }
 
-    fn route(
-        &mut self,
-        root: OwnedEventId,
-        command: TimelineCommand,
-        reply: Responder,
-    ) -> Routed<'_> {
-        if !self.open.contains_key(&root) {
-            self.enqueue(&root, Some(command), reply);
+    fn route(&mut self, id: TimelineId, command: TimelineCommand, reply: Responder) -> Routed<'_> {
+        if !self.open.contains_key(&id) {
+            self.enqueue(&id, Some(command), reply);
             return Routed::Queued;
         }
-        let session = self.open.get_mut(&root).expect("thread was just checked");
+        let session = self.open.get_mut(&id).expect("timeline was just checked");
         Routed::Open(session, command, reply)
     }
 
@@ -412,36 +429,36 @@ impl Threads {
         &mut self,
         built: Result<Built, JoinError>,
     ) -> Option<(
-        OwnedEventId,
+        TimelineId,
         Queued,
         Result<Timeline, matrix_sdk_ui::timeline::Error>,
     )> {
-        let (root, generation, timeline) = match built {
+        let (id, generation, timeline) = match built {
             Ok(built) => built,
             Err(error) if error.is_cancelled() => return None,
             Err(error) => resume_unwind(error),
         };
         if self
             .opening
-            .get(&root)
+            .get(&id)
             .is_none_or(|opening| opening.generation != generation)
         {
             return None;
         }
-        let opening = self.opening.remove(&root)?;
+        let opening = self.opening.remove(&id)?;
         self.queued -= opening.queued.len();
-        Some((root, opening.queued, timeline))
+        Some((id, opening.queued, timeline))
     }
 
-    async fn close_one(&mut self, root: &OwnedEventId) -> Result<(), Error> {
-        if let Some(session) = self.open.remove(root) {
+    async fn close_one(&mut self, id: &TimelineId) -> Result<(), Error> {
+        if let Some(session) = self.open.remove(id) {
             session.close().await;
             return Ok(());
         }
-        let Some(opening) = self.opening.remove(root) else {
+        let Some(opening) = self.opening.remove(id) else {
             return Err(Error::new(
                 ErrorKind::TimelineNotOpen,
-                "thread timeline is not open",
+                "timeline is not open",
             ));
         };
         opening.build.abort();

Modified file: src/engine/watchers.rs+31 additions−13 deletions@@ -401,22 +401,38 @@ pub(super) async fn encryption(client: Client, events: EventSender) {
     }
 }
 
-/// space orders are room account data, which never touches room info
+/// space orders, pins and power levels all feed summaries without raising a notable room info update
 pub(super) async fn rooms(client: Client, events: EventSender) {
     let (_, mut rooms) = client.rooms_stream();
     let mut info = client.room_info_notable_update_receiver();
-    let order_changed = Arc::new(Notify::new());
-    let handler = client.add_event_handler({
-        let order_changed = order_changed.clone();
-        move |event: Raw<AnyRoomAccountDataEvent>| {
-            let order_changed = order_changed.clone();
-            async move {
-                if event_type(&event).as_deref() == Some(space_order::EVENT_TYPE) {
-                    order_changed.notify_one();
+    let changed = Arc::new(Notify::new());
+    let handlers = [
+        client.add_event_handler({
+            let changed = changed.clone();
+            move |event: Raw<AnyRoomAccountDataEvent>| {
+                let changed = changed.clone();
+                async move {
+                    if event_type(&event).as_deref() == Some(space_order::EVENT_TYPE) {
+                        changed.notify_one();
+                    }
                 }
             }
-        }
-    });
+        }),
+        client.add_event_handler({
+            let changed = changed.clone();
+            move |event: Raw<AnySyncStateEvent>| {
+                let changed = changed.clone();
+                async move {
+                    if matches!(
+                        event_type(&event).as_deref(),
+                        Some("m.room.pinned_events" | "m.room.power_levels")
+                    ) {
+                        changed.notify_one();
+                    }
+                }
+            }
+        }),
+    ];
     loop {
         if events.send(Event::RoomsChanged).await.is_err() {
             break;
@@ -424,14 +440,16 @@ pub(super) async fn rooms(client: Client, events: EventSender) {
         tokio::select! {
             update = rooms.next() => if update.is_none() { break },
             update = info.recv() => if matches!(update, Err(RecvError::Closed)) { break },
-            () = order_changed.notified() => {}
+            () = changed.notified() => {}
         }
         // a sync touching fifty rooms is one invalidation, not fifty
         while let Ok(_) | Err(tokio::sync::broadcast::error::TryRecvError::Lagged(_)) =
             info.try_recv()
         {}
     }
-    client.remove_event_handler(handler);
+    for handler in handlers {
+        client.remove_event_handler(handler);
+    }
 }
 
 /// packs live in room state and account data, and leaving a room takes its packs with it

Modified file: src/model.rs+24 additions−4 deletions@@ -115,6 +115,12 @@ pub enum RoomCommand {
         user_id: OwnedUserId,
         muted: bool,
     },
+    /// pinning appends, so the newest pin is last like everywhere else
+    SetPinned {
+        room_id: OwnedRoomId,
+        event_id: OwnedEventId,
+        pinned: bool,
+    },
     /// adding m.favourite drops m.lowpriority, the way Element does
     SetFavourite {
         room_id: OwnedRoomId,
@@ -240,6 +246,10 @@ pub enum TimelineCommand {
         room_id: OwnedRoomId,
         root_event_id: OwnedEventId,
     },
+    /// the room's pinned events as a timeline of their own, kept current as pins change.
+    /// it can't send or paginate, the pins are all it holds
+    OpenPinned(OwnedRoomId),
+    ClosePinned(OwnedRoomId),
     /// Re-emit an authoritative reset for an open room or thread without
     /// closing it. The reset arrives before this command's terminal result.
     Resync {
@@ -282,7 +292,10 @@ impl TimelineCommand {
     #[must_use]
     pub const fn room_id(&self) -> &OwnedRoomId {
         match self {
-            Self::OpenThread { room_id, .. } | Self::CloseThread { room_id, .. } => room_id,
+            Self::OpenThread { room_id, .. }
+            | Self::CloseThread { room_id, .. }
+            | Self::OpenPinned(room_id)
+            | Self::ClosePinned(room_id) => room_id,
             Self::Resync { timeline_id }
             | Self::Paginate { timeline_id, .. }
             | Self::Send { timeline_id, .. }
@@ -296,7 +309,10 @@ impl TimelineCommand {
     #[must_use]
     pub const fn timeline_id(&self) -> Option<&TimelineId> {
         match self {
-            Self::OpenThread { .. } | Self::CloseThread { .. } => None,
+            Self::OpenThread { .. }
+            | Self::CloseThread { .. }
+            | Self::OpenPinned(_)
+            | Self::ClosePinned(_) => None,
             Self::Resync { timeline_id }
             | Self::Paginate { timeline_id, .. }
             | Self::Send { timeline_id, .. }
@@ -317,13 +333,14 @@ pub enum TimelineId {
         room_id: OwnedRoomId,
         root_event_id: OwnedEventId,
     },
+    Pinned(OwnedRoomId),
 }
 
 impl TimelineId {
     #[must_use]
     pub const fn room_id(&self) -> &OwnedRoomId {
         match self {
-            Self::Room(room_id) | Self::Thread { room_id, .. } => room_id,
+            Self::Room(room_id) | Self::Thread { room_id, .. } | Self::Pinned(room_id) => room_id,
         }
     }
 
@@ -338,7 +355,7 @@ impl TimelineId {
     #[must_use]
     pub fn thread_root(&self) -> Option<&OwnedEventId> {
         match self {
-            Self::Room(_) => None,
+            Self::Room(_) | Self::Pinned(_) => None,
             Self::Thread { root_event_id, .. } => Some(root_event_id),
         }
     }
@@ -974,6 +991,9 @@ pub struct RoomSummary {
     pub space_order: Option<String>,
     /// you can send im.ponies.room_emotes here, so you can make and edit its packs
     pub can_edit_packs: bool,
+    /// m.room.pinned_events in order, the newest pin last
+    pub pinned_events: Vec<OwnedEventId>,
+    pub can_pin: bool,
 }
 
 #[derive(Clone, Debug, Eq, PartialEq)]

Modified file: src/normalise/room.rs+7 additions−1 deletions@@ -4,7 +4,8 @@ use matrix_sdk::{
     deserialized_responses::SyncOrStrippedState,
     latest_events::LatestEventValue,
     ruma::events::{
-        AnyMessageLikeEventContent, AnySyncMessageLikeEvent, AnySyncTimelineEvent, SyncStateEvent,
+        AnyMessageLikeEventContent, AnySyncMessageLikeEvent, AnySyncTimelineEvent, StateEventType,
+        SyncStateEvent,
         room::message::{Relation, RoomMessageEventContent},
         space::child::SpaceChildEventContent,
     },
@@ -108,6 +109,11 @@ async fn room_summary(room: &Room, spaces: &SpaceService) -> RoomSummary {
             None
         },
         can_edit_packs: super::can_edit_packs(room).await,
+        pinned_events: room.pinned_event_ids().unwrap_or_default(),
+        can_pin: room.state() == RoomState::Joined
+            && room.power_levels().await.is_ok_and(|levels| {
+                levels.user_can_send_state(room.own_user_id(), StateEventType::RoomPinnedEvents)
+            }),
     }
 }
 

Modified file: src/transport.rs+26 additions@@ -286,6 +286,32 @@ impl ClientHandle {
         .await
     }
 
+    /// the room's pins as `TimelineId::Pinned`, kept current as they change. the room has to be open
+    pub async fn open_pinned(&self, room_id: OwnedRoomId) -> Result<(), Error> {
+        self.run(Command::Timeline(TimelineCommand::OpenPinned(room_id)))
+            .await
+    }
+
+    pub async fn close_pinned(&self, room_id: OwnedRoomId) -> Result<(), Error> {
+        self.run(Command::Timeline(TimelineCommand::ClosePinned(room_id)))
+            .await
+    }
+
+    /// reads the server's list so a pin made elsewhere just before isn't dropped
+    pub async fn set_pinned(
+        &self,
+        room_id: OwnedRoomId,
+        event_id: OwnedEventId,
+        pinned: bool,
+    ) -> Result<(), Error> {
+        self.run(Command::Room(RoomCommand::SetPinned {
+            room_id,
+            event_id,
+            pinned,
+        }))
+        .await
+    }
+
     pub async fn resync(&self, timeline_id: TimelineId) -> Result<(), Error> {
         self.run(Command::Timeline(TimelineCommand::Resync { timeline_id }))
             .await

macltok

Commit fa5b4e09943a

Commit `fa5b4e09943a`