atagen

macltok

public
Star0
Watch0

Commit b34c2df6d896

tarball

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

5 files changed+173-6
mentions: give inbox rows sender profiles and settled replies

Inbox rows came straight from normalise_observed, so they showed raw user ids and left every reply Pending for good. Observing now settles a row from what the client already holds, since the commit hook must not wait on the network, and a later capture's fetched profile and reply replace the offline ones.
Modified file: src/engine/context.rs+47 additions−1 deletions
@@ -18,7 +18,7 @@ use crate::{
1818 engine::MediaService,
1919 normalise::{
2020 MediaRegistration, MediaScope, apply_profiles, member_sender, normalise_context,
21- normalise_observed, resolve_replies,
21+ normalise_observed, resolve_replies, stored_member_sender,
2222 },
2323 transport::EventSender,
2424 };
@@ -540,6 +540,52 @@ async fn fetch_reply(
540540 }
541541 }
542542
543+/// settles an event built outside any page from what the client already holds, so it never
544+/// waits on the network. the sender gets its stored profile and the reply comes from `known`
545+/// or the event cache, or is unavailable
546+pub(crate) async fn settle_offline(
547+ room: &Room,
548+ current_user_id: Option<&ruma::UserId>,
549+ item: &mut TimelineItem,
550+ known: impl Fn(&ruma::EventId) -> Option<TimelineEvent>,
551+) {
552+ let TimelineItem::Event(event) = item else {
553+ return;
554+ };
555+ if let Some(sender) = stored_member_sender(room, &event.sender.user_id).await {
556+ event.sender = sender;
557+ }
558+ let Some(reply) = &event.reply_to else {
559+ return;
560+ };
561+ let target_id = reply.event_id.clone();
562+ let mut target = known(&target_id);
563+ if target.is_none()
564+ && let Ok((cache, _handles)) = room.event_cache().await
565+ && let Ok(Some(cached)) = cache.find_event(&target_id).await
566+ {
567+ target = serde_json::from_str::<Value>(cached.raw().json().get())
568+ .ok()
569+ .and_then(|value| {
570+ let identity = format!("{}\n{target_id}", room.room_id());
571+ match normalise_observed(&value, current_user_id, &identity)? {
572+ (TimelineItem::Event(event), _) => Some(*event),
573+ _ => None,
574+ }
575+ });
576+ }
577+ if let Some(target) = &mut target
578+ && let Some(sender) = stored_member_sender(room, &target.sender.user_id).await
579+ {
580+ target.sender = sender;
581+ }
582+ let targets: HashMap<_, _> = target
583+ .map(|target| (target_id, target))
584+ .into_iter()
585+ .collect();
586+ resolve_replies(std::iter::once(&mut **event), &targets);
587+}
588+
543589 /// under lazy-loaded members a sender may be unknown locally, and `get_member` fetches the
544590 /// member list once for the room when it is
545591 async fn profiles<'a>(
Modified file: src/mentions/service.rs+39 additions−1 deletions
@@ -893,9 +893,23 @@ impl Service {
893893 .or_insert_with(|| Candidate::new(room_id.clone(), value, 0));
894894 return;
895895 }
896- let Some((item, media)) = normalise_observed(&value, Some(&self.account), &identity) else {
896+ let Some((mut item, media)) = normalise_observed(&value, Some(&self.account), &identity)
897+ else {
897898 return;
898899 };
900+ if let Some(room) = self.client.get_room(room_id) {
901+ let known = |event_id: &ruma::EventId| {
902+ let key = id(&MentionKey {
903+ room_id: room_id.clone(),
904+ event_id: event_id.to_owned(),
905+ });
906+ match &data.events.get(&key)?.item {
907+ TimelineItem::Event(event) => Some((**event).clone()),
908+ _ => None,
909+ }
910+ };
911+ context::settle_offline(&room, Some(&self.account), &mut item, known).await;
912+ }
899913 self.supplement(data, &key, &item, &media, &value);
900914 let (reasons, suppressed, unresolved) = self.classify(data, room_id, &value).await;
901915 if unresolved {
@@ -1543,7 +1557,9 @@ impl Service {
15431557 raw: None,
15441558 };
15451559 if let Some(existing) = data.events.get(&event_key) {
1560+ let captured = saved.item;
15461561 saved = existing.clone();
1562+ improve(&mut saved.item, &captured);
15471563 }
15481564 if data.redacted.contains_key(&event_key) {
15491565 redact(&mut saved);
@@ -1556,6 +1572,7 @@ impl Service {
15561572 .into_iter()
15571573 .filter_map(&mut insert)
15581574 .collect();
1575+ let captured_target = context.event.clone();
15591576 let target = insert(context.event);
15601577 let mut after: Vec<_> = context
15611578 .events_after
@@ -1579,6 +1596,7 @@ impl Service {
15791596 after.dedup();
15801597 after.truncate(10);
15811598 let entry = data.entries.get_mut(&key).unwrap();
1599+ improve(&mut entry.event, &captured_target);
15821600 entry.capture = if target.is_none() {
15831601 MentionCaptureState::Unavailable
15841602 } else if after.len() < 10 || (before.len() < 10 && context.start_token.is_some()) {
@@ -1657,6 +1675,26 @@ fn retry_at(error: &Error, fallback_ms: u64) -> u64 {
16571675 now().saturating_add(fallback_ms)
16581676 }
16591677
1678+/// the capture fetched what observing at sync time couldn't, so its profile and reply win
1679+/// over the saved ones, while the saved content keeps its own edits and redactions
1680+fn improve(saved: &mut TimelineItem, captured: &TimelineItem) {
1681+ let (TimelineItem::Event(saved), TimelineItem::Event(captured)) = (saved, captured) else {
1682+ return;
1683+ };
1684+ if saved.event_id != captured.event_id {
1685+ return;
1686+ }
1687+ if captured.sender.display_name.is_some() || captured.sender.avatar.is_some() {
1688+ saved.sender = captured.sender.clone();
1689+ }
1690+ if let (Some(reply), Some(fetched)) = (&mut saved.reply_to, &captured.reply_to)
1691+ && !matches!(reply.details, crate::ReplyDetails::Ready { .. })
1692+ && matches!(fetched.details, crate::ReplyDetails::Ready { .. })
1693+ {
1694+ reply.details = fetched.details.clone();
1695+ }
1696+}
1697+
16601698 fn redact(event: &mut SavedEvent) {
16611699 if let TimelineItem::Event(item) = &mut event.item {
16621700 item.content = TimelineContent::Redacted;
Modified file: src/normalise/member.rs+16 additions−3 deletions
@@ -24,12 +24,25 @@ pub(crate) fn member_summary(member: RoomMember) -> MemberSummary {
2424 }
2525 }
2626
27+/// fetches the member list when it isn't synced yet, and settles for the store when that fails
2728 pub(crate) async fn member_sender(room: &Room, user_id: &OwnedUserId) -> Option<Sender> {
28- let member = room.get_member(user_id).await.ok().flatten()?;
29- Some(Sender {
29+ match room.get_member(user_id).await {
30+ Ok(Some(member)) => Some(sender(user_id, &member)),
31+ _ => stored_member_sender(room, user_id).await,
32+ }
33+}
34+
35+/// what the state store already knows, without asking the server
36+pub(crate) async fn stored_member_sender(room: &Room, user_id: &OwnedUserId) -> Option<Sender> {
37+ let member = room.get_member_no_sync(user_id).await.ok().flatten()?;
38+ Some(sender(user_id, &member))
39+}
40+
41+fn sender(user_id: &OwnedUserId, member: &RoomMember) -> Sender {
42+ Sender {
3043 user_id: user_id.clone(),
3144 display_name: member.display_name().map(str::to_owned),
3245 avatar: member.avatar_url().map(MediaId::plain),
3346 name_ambiguous: member.name_ambiguous(),
34- })
47+ }
3548 }
Modified file: src/normalise/mod.rs+1 additions−1 deletions
@@ -15,7 +15,7 @@ pub(crate) use image_pack::{
1515 };
1616 pub(crate) use link_preview::{ROOM_PREVIEW_URLS, server_preview, url_previews_allowed};
1717 pub(crate) use media::{MediaRegistration, MediaScope, MediaSourceInfo};
18-pub(crate) use member::{member_sender, member_summary};
18+pub(crate) use member::{member_sender, member_summary, stored_member_sender};
1919 pub(crate) use observed::{normalise_observed, poll_question};
2020 pub(crate) use room::notification_mode;
2121 pub(crate) use room::{room_summaries, space_child, stored_space_order};
Modified file: tests/mentions_inbox.rs+70 additions
@@ -1106,3 +1106,73 @@ fn mentions_command_and_page_round_trip_through_serde() {
11061106 let decoded_page: MentionsPage = serde_json::from_value(encoded_page).unwrap();
11071107 assert_eq!(decoded_page, page);
11081108 }
1109+
1110+#[tokio::test]
1111+async fn inbox_rows_carry_sender_profiles_and_never_leave_replies_pending() {
1112+ let server = MatrixMockServer::new().await;
1113+ let room_id = room_id!("!inbox-profiles:localhost");
1114+ let (client, own_user) = joined_client(&server, room_id).await;
1115+ let sender = user_id!("@alice:localhost");
1116+ let feed = SlidingFeed::new(
1117+ room_id.to_owned(),
1118+ vec![message("$parent", sender.as_str(), 1_000, "parent", None)],
1119+ );
1120+ feed.set_required_state(
1121+ room_id,
1122+ vec![json!({
1123+ "type": "m.room.member",
1124+ "state_key": sender,
1125+ "sender": sender,
1126+ "event_id": "$alice-member",
1127+ "origin_server_ts": 1,
1128+ "content": { "membership": "join", "displayname": "Alice" },
1129+ })],
1130+ );
1131+ feed.mount(&server).await;
1132+ let (engine, handle, mut events) = Engine::from_client(
1133+ client,
1134+ EngineConfig::default()
1135+ .with_initial_history_limit(0)
1136+ .with_search_backfill_pages_per_room(0),
1137+ )
1138+ .await
1139+ .unwrap();
1140+ let running = tokio::spawn(engine.run());
1141+ wait_for_requests(&feed, 2).await;
1142+
1143+ let reply = |event_id: &str, target: &str, timestamp| {
1144+ let mut event = message(
1145+ event_id,
1146+ sender.as_str(),
1147+ timestamp,
1148+ "reply",
1149+ Some(own_user.as_str()),
1150+ );
1151+ event["content"]["m.relates_to"] = json!({ "m.in_reply_to": { "event_id": target } });
1152+ event
1153+ };
1154+ feed.replace(
1155+ room_id,
1156+ vec![
1157+ reply("$answer", "$parent", 2_000),
1158+ reply("$orphan", "$missing", 2_100),
1159+ ],
1160+ );
1161+ let page = mentions_page(&mut events, room_id, |page| page.entries.len() == 2).await;
1162+ for entry in &page.entries {
1163+ let macltok::TimelineItem::Event(event) = &entry.event else {
1164+ panic!("a mention should be an event");
1165+ };
1166+ assert_eq!(event.sender.display_name.as_deref(), Some("Alice"));
1167+ let details = &event
1168+ .reply_to
1169+ .as_ref()
1170+ .expect("both mentions reply")
1171+ .details;
1172+ match entry.key.event_id.as_str() {
1173+ "$answer" => assert!(matches!(details, macltok::ReplyDetails::Ready { .. })),
1174+ _ => assert_eq!(details, &macltok::ReplyDetails::Unavailable),
1175+ }
1176+ }
1177+ shut_down(&handle, &mut events, running).await;
1178+}