hive-matrix-mcp: extract the unread decision into a pure function and test it

hive-matrix-mcp/src/handlers.rs

@@ -290,28 +290,53 @@ async fn room_unread_state(
         .map(|(id, _)| id);
 
     let is_encrypted = room.encryption_state().is_encrypted();
-    let events = recent_events(room).await;
-    for raw in &events {
-        let Ok(event) = raw.raw().deserialize() else {
+    let events: Vec<ParsedEvent> = recent_events(room)
+        .await
+        .iter()
+        .map(|raw| {
+            raw.raw()
+                .deserialize()
+                .ok()
+                .map(|event| (event, raw.kind.is_utd()))
+        })
+        .collect();
+    unread_verdict(own_user_id, read_event_id.as_deref(), is_encrypted, &events)
+}
+
+/// One timeline event as `unread_verdict` sees it: the parsed event and
+/// whether the SDK flagged it undecryptable, or `None` when it would not
+/// parse.
+type ParsedEvent = Option<(matrix_sdk::ruma::events::AnySyncTimelineEvent, bool)>;
+
+/// The decision half of [`room_unread_state`], on plain data so it needs no
+/// client or room. `events` is newest first.
+fn unread_verdict(
+    own_user_id: &matrix_sdk::ruma::UserId,
+    read_event_id: Option<&matrix_sdk::ruma::EventId>,
+    is_encrypted: bool,
+    events: &[ParsedEvent],
+) -> Option<(OwnedEventId, OwnedUserId, String)> {
+    for slot in events {
+        let Some((event, is_utd)) = slot else {
             // Unparseable — same as the old filter-then-scan behaviour:
             // skip it rather than let it match the read receipt or count
             // as the fallback's "newest".
             continue;
         };
-        if read_event_id.as_deref() == Some(event.event_id()) {
+        if read_event_id == Some(event.event_id()) {
             // Scanned back to the agent's own read receipt without
             // hitting anything intentional first — proven read, not
             // guessed: there cannot be an unread intentional event in a
             // range we've fully walked.
             return None;
         }
-        if !is_intentional(&event) {
+        if !is_intentional(event) {
             continue;
         }
         if event.sender().as_str() == own_user_id.as_str() {
             return None;
         }
-        let body = utd_gated_body(is_encrypted, raw.kind.is_utd(), &event);
+        let body = utd_gated_body(is_encrypted, *is_utd, event);
         return Some((event.event_id().to_owned(), event.sender().to_owned(), body));
     }
     // Exhausted the lookback window without hitting the receipt or an
@@ -322,12 +347,11 @@ async fn room_unread_state(
     // false "read" risks silently swallowing a real message. Expected
     // to be rare now that the state-churn source this fix targets is
     // itself bounded.
-    let newest_raw = events.first()?;
-    let newest = newest_raw.raw().deserialize().ok()?;
+    let (newest, is_utd) = events.first()?.as_ref()?;
     if newest.sender().as_str() == own_user_id.as_str() {
         return None;
     }
-    let body = utd_gated_body(is_encrypted, newest_raw.kind.is_utd(), &newest);
+    let body = utd_gated_body(is_encrypted, *is_utd, newest);
     Some((
         newest.event_id().to_owned(),
         newest.sender().to_owned(),
@@ -1110,3 +1134,300 @@ mod utd_body_tests {
         assert_eq!(got, "hello");
     }
 }
+
+#[cfg(test)]
+mod unread_tests {
+    use super::{
+        ParsedEvent, attachment_marker, extract_body_sync, extract_in_reply_to, is_intentional,
+        unread_verdict,
+    };
+    use matrix_sdk::ruma::events::AnySyncTimelineEvent;
+    use matrix_sdk::ruma::events::room::message::MessageType;
+    use matrix_sdk::ruma::{EventId, UserId, user_id};
+    use serde_json::json;
+
+    const ME: &str = "@me:example.org";
+
+    fn parse(value: serde_json::Value) -> AnySyncTimelineEvent {
+        serde_json::from_value(value).expect("valid sync event")
+    }
+
+    fn message(id: &str, sender: &str, body: &str) -> AnySyncTimelineEvent {
+        parse(json!({
+            "type": "m.room.message",
+            "event_id": id,
+            "sender": sender,
+            "origin_server_ts": 1,
+            "content": {"msgtype": "m.text", "body": body},
+        }))
+    }
+
+    fn space_child(id: &str, sender: &str) -> AnySyncTimelineEvent {
+        parse(json!({
+            "type": "m.space.child",
+            "state_key": "!child:example.org",
+            "event_id": id,
+            "sender": sender,
+            "origin_server_ts": 1,
+            "content": {"via": ["example.org"]},
+        }))
+    }
+
+    fn live(event: AnySyncTimelineEvent) -> (AnySyncTimelineEvent, bool) {
+        (event, false)
+    }
+
+    fn verdict(read: Option<&str>, events: &[ParsedEvent]) -> Option<(String, String, String)> {
+        let me: &UserId = user_id!("@me:example.org");
+        let read = read.map(|r| <&EventId>::try_from(r).expect("event id"));
+        unread_verdict(me, read, false, events)
+            .map(|(id, sender, body)| (id.to_string(), sender.to_string(), body))
+    }
+
+    #[test]
+    fn an_empty_room_is_caught_up() {
+        assert_eq!(verdict(None, &[]), None);
+    }
+
+    #[test]
+    fn a_message_from_someone_else_past_the_receipt_is_unread() {
+        let events = [
+            Some(live(message("$new:example.org", "@bob:example.org", "hi"))),
+            Some(live(message(
+                "$old:example.org",
+                "@bob:example.org",
+                "older",
+            ))),
+        ];
+        assert_eq!(
+            verdict(Some("$old:example.org"), &events),
+            Some((
+                "$new:example.org".into(),
+                "@bob:example.org".into(),
+                "hi".into()
+            ))
+        );
+    }
+
+    #[test]
+    fn a_receipt_on_the_newest_event_means_caught_up() {
+        let events = [Some(live(message(
+            "$a:example.org",
+            "@bob:example.org",
+            "hi",
+        )))];
+        assert_eq!(verdict(Some("$a:example.org"), &events), None);
+    }
+
+    #[test]
+    fn the_agents_own_latest_message_never_counts_as_unread() {
+        // A read receipt that lags behind the agent's own just-sent
+        // message must not wake the agent for it.
+        let events = [Some(live(message("$mine:example.org", ME, "sent")))];
+        assert_eq!(verdict(None, &events), None);
+    }
+
+    #[test]
+    fn a_space_child_re_emit_is_skipped_to_reach_the_real_message() {
+        let events = [
+            Some(live(space_child("$churn:example.org", "@bot:example.org"))),
+            Some(live(message(
+                "$real:example.org",
+                "@bob:example.org",
+                "hello",
+            ))),
+        ];
+        assert_eq!(
+            verdict(None, &events),
+            Some((
+                "$real:example.org".into(),
+                "@bob:example.org".into(),
+                "hello".into()
+            ))
+        );
+    }
+
+    #[test]
+    fn the_walk_stops_at_the_receipt_before_an_older_intentional_event() {
+        // Churn on top, then the receipt, then an older message: the
+        // receipt proves everything below it was read.
+        let events = [
+            Some(live(space_child("$churn:example.org", "@bot:example.org"))),
+            Some(live(message(
+                "$read:example.org",
+                "@bob:example.org",
+                "read",
+            ))),
+            Some(live(message(
+                "$older:example.org",
+                "@bob:example.org",
+                "older",
+            ))),
+        ];
+        assert_eq!(verdict(Some("$read:example.org"), &events), None);
+    }
+
+    #[test]
+    fn a_window_of_only_churn_falls_back_to_the_newest_event() {
+        // The receipt is further back than the lookback: unknown, so the
+        // newest event is surfaced (a spare wake beats a swallowed message).
+        let events = [Some(live(space_child(
+            "$churn:example.org",
+            "@bot:example.org",
+        )))];
+        assert_eq!(
+            verdict(None, &events),
+            Some((
+                "$churn:example.org".into(),
+                "@bot:example.org".into(),
+                String::new()
+            ))
+        );
+    }
+
+    #[test]
+    fn an_own_authored_newest_event_in_a_churn_window_is_not_unread() {
+        let events = [Some(live(space_child("$churn:example.org", ME)))];
+        assert_eq!(verdict(None, &events), None);
+    }
+
+    #[test]
+    fn an_unparseable_event_is_skipped_not_taken_for_the_newest() {
+        let events = [
+            None,
+            Some(live(message(
+                "$real:example.org",
+                "@bob:example.org",
+                "hello",
+            ))),
+        ];
+        assert_eq!(
+            verdict(None, &events),
+            Some((
+                "$real:example.org".into(),
+                "@bob:example.org".into(),
+                "hello".into()
+            ))
+        );
+    }
+
+    #[test]
+    fn an_undecryptable_event_in_an_encrypted_room_gets_a_per_event_sentinel() {
+        let me: &UserId = user_id!("@me:example.org");
+        let events: [ParsedEvent; 1] = [Some((
+            message("$utd:example.org", "@bob:example.org", "ciphertext"),
+            true,
+        ))];
+        let (_, _, body) = unread_verdict(me, None, true, &events).expect("unread");
+        assert_eq!(body, "[unable to decrypt] ($utd:example.org)");
+    }
+
+    #[test]
+    fn only_the_space_child_re_emit_is_unintentional() {
+        assert!(!is_intentional(&space_child(
+            "$c:example.org",
+            "@bot:example.org"
+        )));
+        assert!(is_intentional(&message(
+            "$m:example.org",
+            "@bob:example.org",
+            "x"
+        )));
+        let topic = parse(json!({
+            "type": "m.room.topic",
+            "state_key": "",
+            "event_id": "$t:example.org",
+            "sender": "@bob:example.org",
+            "origin_server_ts": 1,
+            "content": {"topic": "new"},
+        }));
+        assert!(is_intentional(&topic), "state changes are real activity");
+    }
+
+    fn msgtype(value: &serde_json::Value) -> MessageType {
+        MessageType::new(
+            value["msgtype"].as_str().expect("msgtype"),
+            value["body"].as_str().expect("body").to_owned(),
+            value.as_object().expect("object").clone(),
+        )
+        .expect("known msgtype")
+    }
+
+    #[test]
+    fn attachments_render_as_typed_markers_and_text_as_itself() {
+        let marker = |kind: &str, extra: serde_json::Value| {
+            let mut v = json!({"msgtype": kind, "body": "name.bin"});
+            v.as_object_mut()
+                .expect("object")
+                .extend(extra.as_object().expect("object").clone());
+            attachment_marker(&msgtype(&v))
+        };
+        assert_eq!(marker("m.text", json!({})), "name.bin");
+        assert_eq!(marker("m.notice", json!({})), "name.bin");
+        assert_eq!(marker("m.emote", json!({})), "* name.bin");
+        assert_eq!(
+            marker("m.file", json!({"url": "mxc://example.org/a"})),
+            "[file: name.bin]"
+        );
+        assert_eq!(
+            marker("m.image", json!({"url": "mxc://example.org/a"})),
+            "[image: name.bin]"
+        );
+        assert_eq!(
+            marker("m.audio", json!({"url": "mxc://example.org/a"})),
+            "[audio: name.bin]"
+        );
+        assert_eq!(
+            marker("m.video", json!({"url": "mxc://example.org/a"})),
+            "[video: name.bin]"
+        );
+    }
+
+    #[test]
+    fn a_non_message_event_has_no_body() {
+        assert_eq!(
+            extract_body_sync(&space_child("$c:example.org", "@bot:example.org")),
+            ""
+        );
+        assert_eq!(
+            extract_body_sync(&message("$m:example.org", "@bob:example.org", "hello")),
+            "hello"
+        );
+    }
+
+    #[test]
+    fn reply_ids_come_from_direct_replies_and_thread_fallbacks_only() {
+        let with_relation = |relation: serde_json::Value| {
+            parse(json!({
+                "type": "m.room.message",
+                "event_id": "$r:example.org",
+                "sender": "@bob:example.org",
+                "origin_server_ts": 1,
+                "content": {"msgtype": "m.text", "body": "x", "m.relates_to": relation},
+            }))
+        };
+        assert_eq!(
+            extract_in_reply_to(&with_relation(
+                json!({"m.in_reply_to": {"event_id": "$target:example.org"}})
+            )),
+            Some("$target:example.org".to_owned())
+        );
+        assert_eq!(
+            extract_in_reply_to(&with_relation(json!({
+                "rel_type": "m.thread",
+                "event_id": "$root:example.org",
+                "is_falling_back": true,
+                "m.in_reply_to": {"event_id": "$fallback:example.org"},
+            }))),
+            Some("$fallback:example.org".to_owned())
+        );
+        assert_eq!(
+            extract_in_reply_to(&message("$m:example.org", "@bob:example.org", "plain")),
+            None
+        );
+        assert_eq!(
+            extract_in_reply_to(&space_child("$c:example.org", "@bot:example.org")),
+            None
+        );
+    }
+}