Read and write what a room has pinned

`m.room.pinned_events` was neither read nor written, so the thing a room
tells everyone to read first was unreachable here while every other client
showed it.

Sync now keeps the ids, and `listMatrixPinned` turns them into messages:
scrollback first, since most pins point at something this window already
has, then the server, decrypted the same way any other message is. A pin
whose message is gone is still listed - the room is saying something is
pinned, and dropping it silently would be this client disagreeing with
every other one - and it says which kind of gone: no longer on the server,
or encrypted with a key this session never had.

`setMatrixPinned` reads the list back from the server before writing it.
Pinning replaces the whole list, so writing this client's copy would unpin
whatever somebody else pinned while the window was not looking.

Measured against real rooms: 21 pins in one, 7 in another, and a pin and
unpin round trip in an admin room that left it exactly as found.

Co-Authored-By: Claude Opus 5 <[email protected]>

src/backend/matrix/mod.rs

@@ -2276,6 +2276,108 @@ pub async fn create_room(
 ///
 /// Both are the same shape of call with a different type, and both are
 /// refused by the server if this account's power level is too low, which is
+/// The messages a room has pinned, as messages rather than as ids.
+///
+/// Looked up locally first, because most pins point at something this window
+/// already has; anything else is fetched from the server and decrypted the
+/// same way a scrollback message is. A pin whose message is gone is still
+/// listed - the room is saying something is pinned, and silently dropping it
+/// would be this client disagreeing with every other one.
+pub async fn list_pinned(state: &AppState, account_id: &str, buffer_id: &str) -> Result<Value> {
+    let account = state.accounts.get_matrix(account_id).context("account not connected")?;
+    let room_id = state.runtime.get_matrix_room(buffer_id).context("no known Matrix room for this buffer")?;
+    let base = account.homeserver_url.trim_end_matches('/');
+    let encoded_room = url::form_urlencoded::byte_serialize(room_id.as_bytes()).collect::<String>();
+    let session = state.runtime.get_matrix_machine(account_id);
+    let room = ruma_common::RoomId::parse(&room_id).ok();
+
+    let mut out = Vec::new();
+    for event_id in state.runtime.get_matrix_pinned(account_id, &room_id) {
+        if let Ok(Some(message)) = state.store.get_message(buffer_id, &event_id) {
+            out.push(serde_json::to_value(message)?);
+            continue;
+        }
+        let url = format!(
+            "{base}/_matrix/client/v3/rooms/{encoded_room}/event/{}",
+            url::form_urlencoded::byte_serialize(event_id.as_bytes()).collect::<String>()
+        );
+        let mut event = match http::get_json(&url, &account.access_token).await {
+            Ok(event) => event,
+            Err(e) => {
+                tracing::debug!("matrix: pinned event {event_id} could not be read: {e:#}");
+                out.push(serde_json::json!({
+                    "id": event_id,
+                    "bufferId": buffer_id,
+                    "from": "",
+                    "body": "This pinned message is no longer available.",
+                    "ts": 0,
+                    "kind": "system",
+                }));
+                continue;
+            }
+        };
+        if event["type"].as_str() == Some("m.room.encrypted") {
+            if let (Some(session), Some(room)) = (&session, &room) {
+                match crypto::decrypt_room_event(session, &event, room).await {
+                    Ok(plain) => event = plain,
+                    Err(e) => tracing::debug!("matrix: pinned event {event_id} would not decrypt: {e:#}"),
+                }
+            }
+        }
+        let sender = event["sender"].as_str().unwrap_or_default();
+        out.push(serde_json::json!({
+            "id": event_id,
+            "bufferId": buffer_id,
+            "from": state
+                .runtime
+                .get_matrix_room_members(account_id, &room_id)
+                .get(sender)
+                .cloned()
+                .unwrap_or_else(|| protocol::mxid_localpart(sender)),
+            "body": event["content"]["body"].as_str().unwrap_or("This pinned message cannot be read here."),
+            "ts": event["origin_server_ts"].as_i64().unwrap_or_default() / 1000,
+            "kind": "chat",
+        }));
+    }
+    Ok(serde_json::json!({ "bufferId": buffer_id, "pinned": out }))
+}
+
+/// Pins a message, or takes the pin off.
+///
+/// The list is read back from the server before it is written rather than
+/// taken from this client's copy: pinning is a whole-list replacement, and
+/// writing a stale list would unpin whatever somebody else pinned while this
+/// window was not looking. Whether this account may do it at all is the
+/// server's decision, and its refusal is reported in its own words.
+pub async fn set_pinned(state: &AppState, account_id: &str, buffer_id: &str, event_id: &str, pinned: bool) -> Result<()> {
+    let account = state.accounts.get_matrix(account_id).context("account not connected")?;
+    let room_id = state.runtime.get_matrix_room(buffer_id).context("no known Matrix room for this buffer")?;
+    let base = account.homeserver_url.trim_end_matches('/');
+    let encoded_room = url::form_urlencoded::byte_serialize(room_id.as_bytes()).collect::<String>();
+
+    let url = format!("{base}/_matrix/client/v3/rooms/{encoded_room}/state/m.room.pinned_events/");
+    // A room that has never pinned anything has no such state event at all,
+    // which is a 404 and an empty list rather than a failure.
+    let current: Vec<String> = match http::get_json(&url, &account.access_token).await {
+        Ok(content) => content["pinned"]
+            .as_array()
+            .map(|ids| ids.iter().filter_map(|id| id.as_str().map(|s| s.to_string())).collect())
+            .unwrap_or_default(),
+        Err(_) => Vec::new(),
+    };
+
+    let mut next: Vec<String> = current.into_iter().filter(|id| id != event_id).collect();
+    if pinned {
+        next.push(event_id.to_string());
+    }
+    set_room_state(state, account_id, buffer_id, "m.room.pinned_events", serde_json::json!({ "pinned": next })).await?;
+    // Locally too, so the list is right before the next sync arrives.
+    state.runtime.set_matrix_pinned(account_id, &room_id, next.clone());
+    state.events.emit("matrixPinned", serde_json::json!({ "bufferId": buffer_id, "pinned": next }));
+    Ok(())
+}
+
+
 /// where that question belongs.
 pub async fn set_room_state(
     state: &AppState,

src/backend/matrix/roomstate.rs

@@ -38,6 +38,23 @@ pub async fn process_state_events(state: &AppState, account_id: &str, room_id: &
             "m.room.power_levels" => {
                 state.runtime.set_matrix_power_levels(account_id, room_id, event["content"].clone());
             }
+            // What the room has told everyone to read first. Ids only: the
+            // messages themselves are looked up when somebody asks to see
+            // them, because a pin can point at something said years before
+            // this client ever joined.
+            "m.room.pinned_events" => {
+                let pinned: Vec<String> = event["content"]["pinned"]
+                    .as_array()
+                    .map(|ids| ids.iter().filter_map(|id| id.as_str().map(|s| s.to_string())).collect())
+                    .unwrap_or_default();
+                state.runtime.set_matrix_pinned(account_id, room_id, pinned.clone());
+                if let Some(buffer_id) = state.runtime.matrix_buffer_for_room(account_id, room_id) {
+                    state.events.emit(
+                        "matrixPinned",
+                        serde_json::json!({ "bufferId": buffer_id, "pinned": pinned }),
+                    );
+                }
+            }
             "m.room.avatar" => {
                 if let Some(mxc) = event["content"]["url"].as_str() {
                     if let Some(path) = cached_media_path(homeserver_url, access_token, mxc, "").await {

src/rpc/methods.rs

@@ -1013,6 +1013,37 @@ pub async fn dispatch(
             }
         }
 
+        // What a room has told everyone to read first.
+        "listMatrixPinned" => {
+            let Some(buffer_id) = p_str_opt(params, "bufferId") else {
+                return (None, Some("listMatrixPinned requires \"bufferId\"".to_string()));
+            };
+            let Some(buffer) = state.runtime.get_buffer(buffer_id) else {
+                return (None, Some("no such conversation".to_string()));
+            };
+            match backend::matrix::list_pinned(state, &buffer.account_id, buffer_id).await {
+                Ok(answer) => (Some(answer), None),
+                Err(e) => (None, Some(format!("{e:#}"))),
+            }
+        }
+
+        // Pinning one, or taking the pin off. Whether this account may is the
+        // server's decision, and its refusal is passed through in its words.
+        "setMatrixPinned" => {
+            let (buffer_id, event_id) = match (p_str_opt(params, "bufferId"), p_str_opt(params, "eventId")) {
+                (Some(b), Some(e)) => (b, e),
+                _ => return (None, Some("setMatrixPinned requires \"bufferId\" and \"eventId\"".to_string())),
+            };
+            let Some(buffer) = state.runtime.get_buffer(buffer_id) else {
+                return (None, Some("no such conversation".to_string()));
+            };
+            let pinned = params.get("pinned").and_then(|v| v.as_bool()).unwrap_or(true);
+            match backend::matrix::set_pinned(state, &buffer.account_id, buffer_id, event_id, pinned).await {
+                Ok(()) => (Some(ok_node()), None),
+                Err(e) => (None, Some(format!("{e:#}"))),
+            }
+        }
+
         // The same question asked of the homeserver rather than of this
         // window's own copy.
         //

src/runtime.rs

@@ -546,6 +546,12 @@ pub struct Runtime {
     /// this is advisory/UI-gating only, an unauthorized action would be
     /// rejected server-side regardless of what's cached here.
     matrix_power_levels: Mutex<HashMap<(String, String), Value>>,
+    /// The event ids a room has pinned, by account and room.
+    ///
+    /// The ids only: what those events *say* lives in scrollback or on the
+    /// server, and a pin outliving the message it points at is a normal state
+    /// rather than a broken one.
+    matrix_pinned: Mutex<HashMap<(String, String), Vec<String>>>,
     /// Matrix-specific: (account id, room id) -> {user id -> display name}
     /// for every member currently *joined* to that room (see roomstate.rs's
     /// m.room.member handling - a leave/ban removes the entry entirely,
@@ -766,6 +772,7 @@ impl Runtime {
             matrix_member_avatars: Mutex::new(HashMap::new()),
             matrix_room_avatars: Mutex::new(HashMap::new()),
             matrix_power_levels: Mutex::new(HashMap::new()),
+            matrix_pinned: Mutex::new(HashMap::new()),
             matrix_room_members: Mutex::new(HashMap::new()),
             matrix_read_receipts: Mutex::new(HashMap::new()),
             matrix_upgrades: Mutex::new(Vec::new()),
@@ -2186,6 +2193,19 @@ impl Runtime {
         self.matrix_power_levels.lock().unwrap().insert((account_id.to_string(), room_id.to_string()), content);
     }
 
+    pub fn set_matrix_pinned(&self, account_id: &str, room_id: &str, events: Vec<String>) {
+        self.matrix_pinned.lock().unwrap().insert((account_id.to_string(), room_id.to_string()), events);
+    }
+
+    pub fn get_matrix_pinned(&self, account_id: &str, room_id: &str) -> Vec<String> {
+        self.matrix_pinned
+            .lock()
+            .unwrap()
+            .get(&(account_id.to_string(), room_id.to_string()))
+            .cloned()
+            .unwrap_or_default()
+    }
+
     pub fn get_matrix_power_levels(&self, account_id: &str, room_id: &str) -> Option<Value> {
         self.matrix_power_levels.lock().unwrap().get(&(account_id.to_string(), room_id.to_string())).cloned()
     }