Matrix: say a room has been read, and stop orphaning backups

Closes #71 and #67.

**Read markers.** `markBufferRead` was a no-op for Matrix, so a room read
here stayed bold on every other client, and one read elsewhere stayed bold
here. Both markers are sent, because Matrix has two and they answer
different questions: `m.read` is public and is what puts your avatar
against a message in somebody else's client, while `m.fully_read` is
private and is what your own other sessions use to stop showing the room
as unread.

Pointed at the newest message this client actually holds, by its real
event id - a marker is a claim about a message rather than about a time.
A room with nothing in it says nothing, since a receipt pointing at no
event is not a claim anybody can act on.

Best-effort: failing to say a room has been read must not make reading it
look like an error.

**Backups.** Making a new recovery key registered a new version and left
the old one on the server - still valid under the *old* key, still holding
keys the new one cannot read, and no longer receiving anything. Somebody
with the previous key written down would find it restored a partial
history and stopped, with nothing saying why.

The old version is deleted after the new one is registered, in that order
deliberately: deleting first would leave an account with no backup at all
if registering the replacement then failed, which is the one outcome worse
than having two. A failed delete is not a failed setup - refusing to hand
over a working recovery key over an untidy server would be the wrong
trade - so it warns and carries on.

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

src/backend/matrix/backup.rs

@@ -79,15 +79,45 @@ pub async fn reactivate_on_connect(state: &AppState, account_id: &str, session:
 /// piggybacks on the sync loop's existing per-cycle dispatch - see
 /// crypto.rs's run_pending_backup, called from mod.rs's run_sync.
 ///
-/// v1 limitation: always creates a brand new backup version rather than
-/// checking for/reusing an existing one. Calling this twice for the same
-/// account leaves the older backup version on the server, still valid
-/// under the *old* recovery key but no longer receiving this device's
-/// future keys - deleting old versions explicitly is a follow-up, not
-/// handled here.
+/// Making a new key replaces the old backup rather than leaving it behind.
+///
+/// The previous version is deleted after the new one is registered, and in
+/// that order deliberately: a delete that ran first would leave an account
+/// with no backup at all if registering the replacement then failed, which is
+/// the one outcome worse than having two.
+///
+/// It used to leave the old version in place - still valid under the *old*
+/// recovery key, still holding keys the new one cannot read, and no longer
+/// receiving anything. Somebody with the previous key written down would find
+/// it restored a partial history and stopped, with nothing saying why.
+///
+/// Failing to delete is not failure. The new backup works; a stale version is
+/// untidy, and refusing to hand over a working recovery key over untidiness
+/// would be the wrong trade.
+/// The backup version the server currently holds, if it holds one.
+///
+/// Absent is the ordinary answer for an account that has never had a backup,
+/// so it is not an error - and neither is a server that will not say, since
+/// the only thing this is used for is tidying up afterwards.
+async fn current_version(base: &str, access_token: &str) -> Option<String> {
+    let resp = http::get_json(&format!("{base}/_matrix/client/v3/room_keys/version"), access_token).await.ok()?;
+    resp["version"].as_str().map(str::to_string)
+}
+
+async fn delete_version(base: &str, access_token: &str, version: &str) -> Result<()> {
+    let url = format!(
+        "{base}/_matrix/client/v3/room_keys/version/{}",
+        url::form_urlencoded::byte_serialize(version.as_bytes()).collect::<String>()
+    );
+    http::delete_json(&url, access_token).await.map(|_| ()).context("deleting the previous backup version")
+}
+
 pub async fn setup_recovery_key(state: &AppState, account_id: &str) -> Result<String> {
     let (config, session) = account_session(state, account_id).await?;
     let base = config.homeserver_url.trim_end_matches('/');
+    // Read before anything is changed, so what is replaced is the version that
+    // existed when the request was made.
+    let previous = current_version(base, &config.access_token).await;
 
     let key = BackupDecryptionKey::new();
     let recovery_key = key.to_base58();
@@ -104,9 +134,18 @@ pub async fn setup_recovery_key(state: &AppState, account_id: &str) -> Result<St
     let backup_key: MegolmV1BackupKey = key.megolm_v1_public_key();
     backup_key.set_version(version.clone());
     backup_machine.enable_backup_v1(backup_key).await.context("enable_backup_v1")?;
-    backup_machine.save_decryption_key(Some(key), Some(version)).await.context("save_decryption_key")?;
+    backup_machine.save_decryption_key(Some(key), Some(version.clone())).await.context("save_decryption_key")?;
     state.runtime.mark_matrix_backup_enabled(account_id);
 
+    if let Some(previous) = previous.filter(|p| *p != version) {
+        match delete_version(base, &config.access_token, &previous).await {
+            Ok(()) => tracing::info!("matrix[{account_id}]: replaced backup version {previous} with {version}"),
+            // Said out loud rather than swallowed: the new key works, and
+            // somebody may still find an orphan on the server later.
+            Err(e) => tracing::warn!("matrix[{account_id}]: new backup is {version}, but the old {previous} could not be removed: {e:#}"),
+        }
+    }
+
     Ok(recovery_key)
 }
 

src/backend/matrix/http.rs

@@ -116,6 +116,11 @@ pub async fn get_json(url: &str, token: &str) -> Result<Value> {
     handle_response(resp).await
 }
 
+pub async fn delete_json(url: &str, token: &str) -> Result<Value> {
+    let resp = http_client().delete(url).bearer_auth(token).send().await.context("request failed")?;
+    handle_response(resp).await
+}
+
 pub async fn put_json(url: &str, token: &str, body: Value) -> Result<Value> {
     let resp = http_client().put(url).bearer_auth(token).json(&body).send().await.context("request failed")?;
     handle_response(resp).await

src/backend/matrix/mod.rs

@@ -1135,6 +1135,42 @@ pub async fn backfill(state: &AppState, account_id: &str, buffer_id: &str, limit
 /// Matrix wants a timeout with the notice and cancels with `typing: false`,
 /// unlike Discord's single fire-and-expire - so a caller that stops typing
 /// can actually say so rather than waiting the notice out.
+/// Says this room has been read, as far as its newest message.
+///
+/// Two markers, because Matrix has two and they answer different questions.
+/// `m.read` is public - it is what puts your avatar against a message in
+/// somebody else's client - and `m.fully_read` is private, and is what your
+/// own other clients use to stop showing the room as unread.
+///
+/// Sent together through the one endpoint that takes both, so reading a room
+/// here stops it being bold on your phone. Before this, `markBufferRead` was
+/// a no-op for Matrix and the two never agreed.
+pub async fn mark_read(state: &AppState, account_id: &str, buffer_id: &str) -> Result<()> {
+    let Some(config) = state.accounts.get_matrix(account_id) else { anyhow::bail!("no such account") };
+    let Some(room_id) = state.runtime.get_matrix_room(buffer_id) else {
+        anyhow::bail!("no known Matrix room for this buffer")
+    };
+    // The newest message this client actually holds. Nothing to say if the
+    // room has never had one - and claiming to have read a room that is empty
+    // would be a receipt pointing at nothing.
+    let Some(event_id) = state.store.newest_message_id(buffer_id).ok().flatten().filter(|id| id.starts_with('$')) else {
+        return Ok(());
+    };
+    let base = config.homeserver_url.trim_end_matches('/');
+    let url = format!(
+        "{base}/_matrix/client/v3/rooms/{}/read_markers",
+        url::form_urlencoded::byte_serialize(room_id.as_bytes()).collect::<String>()
+    );
+    http::post_json(
+        &url,
+        Some(&config.access_token),
+        serde_json::json!({ "m.fully_read": event_id, "m.read": event_id }),
+    )
+    .await
+    .context("sending read markers")?;
+    Ok(())
+}
+
 pub async fn send_typing(state: &AppState, account_id: &str, room_id: &str, typing: bool) -> Result<()> {
     let account = state.accounts.get_matrix(account_id).context("account not connected")?;
     let base = account.homeserver_url.trim_end_matches('/');

src/rpc/methods.rs

@@ -503,6 +503,16 @@ pub async fn dispatch(
                         Err(e) => (None, Some(e.to_string())),
                     },
                 },
+                // Matrix has read markers and they were never sent, so a room
+                // read here stayed bold everywhere else. Best-effort: failing
+                // to say a room has been read must not make reading it look
+                // like an error.
+                Some(buffer) if buffer.account_id.starts_with("matrix:") => {
+                    if let Err(e) = backend::matrix::mark_read(state, &buffer.account_id, buffer_id).await {
+                        tracing::debug!("matrix read markers: {e:#}");
+                    }
+                    (Some(ok_node()), None)
+                }
                 _ => (Some(ok_node()), None),
             }
         }

src/store.rs

@@ -610,6 +610,20 @@ impl Store {
     /// both when the buffer/id genuinely doesn't exist and when it's
     /// aged out of local scrollback (prune_old_messages) - either way,
     /// callers fall back to a minimal reply preview with no body/from.
+    /// The id of the newest message stored for a buffer.
+    ///
+    /// Used to say how far a room has been read, which is a claim about a
+    /// specific message rather than about a time - so it has to be the id the
+    /// service itself gave, not one generated here.
+    pub fn newest_message_id(&self, buffer_id: &str) -> Result<Option<String>> {
+        let conn = self.conn.lock().unwrap();
+        let mut stmt = conn.prepare(
+            "SELECT msg_id FROM messages WHERE buffer_id = ?1 ORDER BY ts DESC, rowid DESC LIMIT 1",
+        )?;
+        let mut rows = stmt.query_map(params![buffer_id], |row| row.get::<_, Option<String>>(0))?;
+        Ok(rows.next().transpose()?.flatten())
+    }
+
     pub fn get_message(&self, buffer_id: &str, msg_id: &str) -> Result<Option<Message>> {
         let conn = self.conn.lock().unwrap();
         let mut stmt = conn.prepare(
@@ -730,6 +744,28 @@ mod tests {
             .expect("appending");
     }
 
+    /// Like `append`, but at a time - so "newest" can be checked.
+    fn append_at(s: &Store, buffer: &str, id: &str, ts: i64) {
+        s.append_message(buffer, id, "someone", "hi", ts, false, false, "chat", None, &[], false, None, &[], &[], None, None, "channel", None, &[])
+            .expect("appending");
+    }
+
+    #[test]
+    fn finds_the_newest_message_to_mark_a_room_read_up_to() {
+        let (s, _dir) = store();
+        assert_eq!(s.newest_message_id("b").unwrap(), None, "an empty room has nothing to have read");
+
+        append_at(&s, "b", "$one", 100);
+        append_at(&s, "b", "$two", 300);
+        append_at(&s, "b", "$mid", 200);
+        // Newest by time, not by arrival: history arriving after a live
+        // message must not move the marker backwards.
+        assert_eq!(s.newest_message_id("b").unwrap().as_deref(), Some("$two"));
+        // And it is that room's newest, not the store's.
+        append_at(&s, "other", "$elsewhere", 999);
+        assert_eq!(s.newest_message_id("b").unwrap().as_deref(), Some("$two"));
+    }
+
     #[test]
     fn a_mention_is_found_without_the_conversation_being_open() {
         // The whole point: buffers live only as long as the daemon, so a