atagen

macltok

public
Star0
Watch0

Commit cb7b3362e1ce

tarball

atagen <boss@atagen.co>2026-10-25 09:30 UTC

7 files changed+479-187
context: restore what the focused timeline gave event context

Raw /messages pages lost sender profiles, showed edits, reactions and redactions as rows of their own and left replies Pending forever. Pages now fold in-page edits and reactions into their targets, hide thread replies as the live timeline does, take profiles from room members and settle every reply from the page or a fetch. Truncating to the limit while handing back the server's end token skipped events, so pages are no longer cut.
Modified file: src/engine/context.rs+208 additions−159 deletions
@@ -1,15 +1,17 @@
1-use std::sync::Arc;
1+use std::{collections::HashMap, sync::Arc};
22
3-use matrix_sdk::{
4- Room, deserialized_responses::TimelineEvent as SdkTimelineEvent, room::MessagesOptions,
5-};
6-use ruma::{OwnedEventId, OwnedUserId, RoomId, UInt};
3+use futures_util::future::{join_all, try_join};
4+use matrix_sdk::{Room, room::MessagesOptions};
5+use ruma::{OwnedEventId, OwnedUserId, UInt};
6+use serde_json::Value;
77
88 use crate::{
9- Error, ErrorKind, EventContext, EventContextCommand, EventContextDirection, EventContextPage,
10- TimelineItem,
9+ Error, ErrorKind, EventContext, EventContextCommand, EventContextPage, MediaId, Sender,
10+ TimelineEvent, TimelineItem,
1111 engine::MediaService,
12- normalise::{MediaRegistration, normalise_observed},
12+ normalise::{
13+ MediaRegistration, apply_profiles, normalise_context, normalise_observed, resolve_replies,
14+ },
1315 };
1416
1517 const MAX_CONTEXT_LIMIT: u16 = 50;
@@ -20,44 +22,42 @@ pub(crate) async fn execute(
2022 media: Arc<MediaService>,
2123 command: EventContextCommand,
2224 ) -> Result<crate::CommandOutput, Error> {
23- if let EventContextCommand::Before { token, limit, .. }
24- | EventContextCommand::After { token, limit, .. } = &command
25- {
26- validate_limit(*limit)?;
27- if token.is_empty() {
28- return Err(Error::new(
29- ErrorKind::Invalid,
30- "context pagination token is empty",
31- ));
32- }
33- return execute_page(
34- room,
35- current_user_id,
36- media,
37- token,
38- *limit,
39- matches!(command, EventContextCommand::Before { .. }),
40- )
41- .await;
42- }
43-
44- let (event_id, before_limit, after_limit) = match &command {
25+ let (token, limit, backwards) = match command {
4526 EventContextCommand::Fetch {
4627 event_id,
4728 before_limit,
4829 after_limit,
4930 ..
5031 } => {
51- validate_limit(*before_limit)?;
52- validate_limit(*after_limit)?;
53- (event_id.clone(), *before_limit, *after_limit)
32+ let (context, registrations, _) =
33+ capture(room, current_user_id, event_id, before_limit, after_limit).await?;
34+ media.register(registrations).await;
35+ return Ok(crate::CommandOutput::EventContext(context));
5436 }
55- EventContextCommand::Before { .. } | EventContextCommand::After { .. } => unreachable!(),
37+ EventContextCommand::Before { token, limit, .. } => (token, limit, true),
38+ EventContextCommand::After { token, limit, .. } => (token, limit, false),
5639 };
57- let (context, registrations, _) =
58- capture(room, current_user_id, event_id, before_limit, after_limit).await?;
59- media.register(registrations).await;
60- Ok(crate::CommandOutput::EventContext(context))
40+ if limit == 0 {
41+ return Err(Error::new(
42+ ErrorKind::Invalid,
43+ "context page limit must be at least 1",
44+ ));
45+ }
46+ validate_limit(limit)?;
47+ if token.is_empty() {
48+ return Err(Error::new(
49+ ErrorKind::Invalid,
50+ "context pagination token is empty",
51+ ));
52+ }
53+ let side = fetch_side(&room, Some(&token), limit, backwards).await?;
54+ let mut page = Assembly::new(&room, current_user_id.as_deref(), &side.values, None).await;
55+ let events = page.take(0..side.values.len());
56+ media.register(page.media).await;
57+ Ok(crate::CommandOutput::EventContextPage(EventContextPage {
58+ events,
59+ next_token: side.token,
60+ }))
6161 }
6262
6363 pub(crate) async fn capture(
@@ -66,7 +66,7 @@ pub(crate) async fn capture(
6666 event_id: OwnedEventId,
6767 before_limit: u16,
6868 after_limit: u16,
69-) -> Result<(EventContext, Vec<MediaRegistration>, Vec<serde_json::Value>), Error> {
69+) -> Result<(EventContext, Vec<MediaRegistration>, Vec<Value>), Error> {
7070 validate_limit(before_limit)?;
7171 validate_limit(after_limit)?;
7272 let raw = room
@@ -84,76 +84,70 @@ pub(crate) async fn capture(
8484 error
8585 }
8686 })?;
87- if raw.event.is_none() {
88- return Err(Error::new(ErrorKind::EventNotFound, "event was not found"));
89- }
90- let before = fetch_side(
91- &room,
92- raw.prev_batch_token.as_deref(),
93- before_limit,
94- true,
95- current_user_id.as_deref(),
96- );
97- let after = fetch_side(
98- &room,
99- raw.next_batch_token.as_deref(),
100- after_limit,
101- false,
102- current_user_id.as_deref(),
103- );
104- let (events_before, mut before_media, start_token, mut observed) = before.await?;
105- let (events_after, after_media, end_token, following) = after.await?;
106- observed.extend(following);
107- if let Some(value) = raw
87+ let target = raw
10888 .event
10989 .as_ref()
110- .and_then(|event| serde_json::from_str(event.raw().json().get()).ok())
111- {
112- observed.push(value);
113- }
114- let event = raw.event.as_ref().and_then(|event| {
115- normalise_sdk_event(event, current_user_id.as_deref(), room.room_id()).map(
116- |(item, media)| {
117- before_media.extend(media);
118- item
119- },
120- )
121- });
122- before_media.extend(after_media);
90+ .and_then(|event| serde_json::from_str::<Value>(event.raw().json().get()).ok())
91+ .ok_or_else(|| Error::new(ErrorKind::EventNotFound, "event was not found"))?;
92+ let (before, after) = try_join(
93+ fetch_side(&room, raw.prev_batch_token.as_deref(), before_limit, true),
94+ fetch_side(&room, raw.next_batch_token.as_deref(), after_limit, false),
95+ )
96+ .await?;
97+ let target_index = before.values.len();
98+ let mut values = before.values;
99+ values.push(target);
100+ values.extend(after.values);
101+ let mut page = Assembly::new(
102+ &room,
103+ current_user_id.as_deref(),
104+ &values,
105+ Some(target_index),
106+ )
107+ .await;
108+ let events_before = page.take(0..target_index);
109+ let event = page
110+ .take(target_index..target_index + 1)
111+ .pop()
112+ .ok_or_else(|| Error::new(ErrorKind::EventNotFound, "event could not be read"))?;
113+ let events_after = page.take(target_index + 1..values.len());
123114 Ok((
124115 EventContext {
125116 events_before,
126117 event,
127118 events_after,
128- start_token,
129- end_token,
119+ start_token: before.token,
120+ end_token: after.token,
130121 },
131- before_media,
132- observed,
122+ page.media,
123+ values,
133124 ))
134125 }
135126
127+struct Side {
128+ /// in timeline order
129+ values: Vec<Value>,
130+ token: Option<String>,
131+}
132+
136133 async fn fetch_side(
137134 room: &Room,
138135 token: Option<&str>,
139136 limit: u16,
140137 backwards: bool,
141- current_user_id: Option<&ruma::UserId>,
142-) -> Result<
143- (
144- Vec<TimelineItem>,
145- Vec<MediaRegistration>,
146- Option<String>,
147- Vec<serde_json::Value>,
148- ),
149- Error,
150-> {
151- if limit == 0 {
152- return Ok((Vec::new(), Vec::new(), token.map(str::to_owned), Vec::new()));
153- }
138+) -> Result<Side, Error> {
154139 let Some(token) = token else {
155- return Ok((Vec::new(), Vec::new(), None, Vec::new()));
140+ return Ok(Side {
141+ values: Vec::new(),
142+ token: None,
143+ });
156144 };
145+ if limit == 0 {
146+ return Ok(Side {
147+ values: Vec::new(),
148+ token: Some(token.to_owned()),
149+ });
150+ }
157151 let mut options = if backwards {
158152 MessagesOptions::backward()
159153 } else {
@@ -165,87 +159,142 @@ async fn fetch_side(
165159 .messages(options)
166160 .await
167161 .map_err(|source| Error::from_sdk("fetching event context events failed", source))?;
168- let mut events = messages.chunk;
169- events.truncate(usize::from(limit));
170- if backwards {
171- events.reverse();
172- }
173- let next_token = messages.end;
174- let observed = events
162+ let mut values: Vec<Value> = messages
163+ .chunk
175164 .iter()
176165 .filter_map(|event| serde_json::from_str(event.raw().json().get()).ok())
177166 .collect();
178- let (items, media) = normalise_events(events, current_user_id, room.room_id());
179- Ok((items, media, next_token, observed))
180-}
181-
182-async fn execute_page(
183- room: Room,
184- current_user_id: Option<OwnedUserId>,
185- media: Arc<MediaService>,
186- token: &str,
187- limit: u16,
188- backwards: bool,
189-) -> Result<crate::CommandOutput, Error> {
190- let mut options = if backwards {
191- MessagesOptions::backward()
192- } else {
193- MessagesOptions::forward()
194- };
195- options = options.from(token);
196- options.limit = UInt::from(limit);
197- let messages = room
198- .messages(options)
199- .await
200- .map_err(|source| Error::from_sdk("fetching event context page failed", source))?;
201- let next_token = messages.end.clone();
202- let mut events = messages.chunk;
203- events.truncate(usize::from(limit));
204167 if backwards {
205- events.reverse();
168+ values.reverse();
206169 }
207- let (events, registrations) =
208- normalise_events(events, current_user_id.as_deref(), room.room_id());
209- media.register(registrations).await;
210- let end_reached = next_token.is_none();
211- let direction = if backwards {
212- EventContextDirection::Before
213- } else {
214- EventContextDirection::After
215- };
216- Ok(crate::CommandOutput::EventContextPage(EventContextPage {
217- direction,
218- events,
219- next_token,
220- end_reached,
221- }))
170+ Ok(Side {
171+ values,
172+ token: messages.end,
173+ })
222174 }
223175
224-fn normalise_events(
225- events: Vec<SdkTimelineEvent>,
226- current_user_id: Option<&ruma::UserId>,
227- room_id: &RoomId,
228-) -> (Vec<TimelineItem>, Vec<MediaRegistration>) {
229- let mut items = Vec::with_capacity(events.len());
230- let mut media = Vec::new();
231- for event in events {
232- if let Some((item, registrations)) = normalise_sdk_event(&event, current_user_id, room_id) {
233- items.push(item);
176+/// a window of raw events normalised together, so relations inside it land on their targets
177+struct Assembly {
178+ items: Vec<(usize, TimelineItem)>,
179+ media: Vec<MediaRegistration>,
180+}
181+
182+impl Assembly {
183+ async fn new(
184+ room: &Room,
185+ current_user_id: Option<&ruma::UserId>,
186+ values: &[Value],
187+ keep: Option<usize>,
188+ ) -> Self {
189+ let page = normalise_context(values, keep, current_user_id, room.room_id());
190+ let mut items = page.items;
191+ let mut media = page.media;
192+ let mut targets: HashMap<OwnedEventId, TimelineEvent> = HashMap::new();
193+ let mut missing = Vec::new();
194+ for (_, item) in &items {
195+ if let TimelineItem::Event(event) = item
196+ && let Some(event_id) = &event.event_id
197+ {
198+ targets.insert(event_id.clone(), (**event).clone());
199+ }
200+ }
201+ for event in events(&mut items) {
202+ if let Some(reply) = &event.reply_to
203+ && !targets.contains_key(&reply.event_id)
204+ && !missing.contains(&reply.event_id)
205+ {
206+ missing.push(reply.event_id.clone());
207+ }
208+ }
209+ let fetched = join_all(
210+ missing
211+ .into_iter()
212+ .map(|event_id| fetch_reply(room, current_user_id, event_id)),
213+ )
214+ .await;
215+ for (event, registrations) in fetched.into_iter().flatten() {
234216 media.extend(registrations);
217+ if let Some(event_id) = event.event_id.clone() {
218+ targets.insert(event_id, event);
219+ }
235220 }
221+ let profiles = profiles(
222+ room,
223+ items
224+ .iter()
225+ .filter_map(|(_, item)| match item {
226+ TimelineItem::Event(event) => Some(&event.sender.user_id),
227+ _ => None,
228+ })
229+ .chain(targets.values().map(|event| &event.sender.user_id)),
230+ )
231+ .await;
232+ apply_profiles(targets.values_mut(), &profiles);
233+ apply_profiles(events(&mut items), &profiles);
234+ resolve_replies(events(&mut items), &targets);
235+ Self { items, media }
236236 }
237- (items, media)
237+
238+ /// the items that came from raw events in `range`
239+ fn take(&mut self, range: std::ops::Range<usize>) -> Vec<TimelineItem> {
240+ let (taken, kept) = std::mem::take(&mut self.items)
241+ .into_iter()
242+ .partition(|(index, _)| range.contains(index));
243+ self.items = kept;
244+ taken.into_iter().map(|(_, item)| item).collect()
245+ }
246+}
247+
248+fn events(items: &mut [(usize, TimelineItem)]) -> impl Iterator<Item = &mut TimelineEvent> {
249+ items.iter_mut().filter_map(|(_, item)| match item {
250+ TimelineItem::Event(event) => Some(&mut **event),
251+ _ => None,
252+ })
238253 }
239254
240-fn normalise_sdk_event(
241- event: &SdkTimelineEvent,
255+async fn fetch_reply(
256+ room: &Room,
242257 current_user_id: Option<&ruma::UserId>,
243- room_id: &RoomId,
244-) -> Option<(TimelineItem, Vec<MediaRegistration>)> {
245- let value: serde_json::Value = serde_json::from_str(event.raw().json().get()).ok()?;
246- let event_id = value.get("event_id")?.as_str()?;
247- let identity = format!("{room_id}\n{event_id}");
248- normalise_observed(&value, current_user_id, &identity)
258+ event_id: OwnedEventId,
259+) -> Option<(TimelineEvent, Vec<MediaRegistration>)> {
260+ let event = room
261+ .load_or_fetch_event(&event_id, None)
262+ .await
263+ .inspect_err(|error| tracing::debug!(%event_id, %error, "fetching a reply target failed"))
264+ .ok()?;
265+ let value: Value = serde_json::from_str(event.raw().json().get()).ok()?;
266+ let identity = format!("{}\n{event_id}", room.room_id());
267+ match normalise_observed(&value, current_user_id, &identity)? {
268+ (TimelineItem::Event(event), media) => Some((*event, media)),
269+ _ => None,
270+ }
271+}
272+
273+/// under lazy-loaded members a sender may be unknown locally, and `get_member` fetches the
274+/// member list once for the room when it is
275+async fn profiles<'a>(
276+ room: &Room,
277+ user_ids: impl Iterator<Item = &'a OwnedUserId>,
278+) -> HashMap<OwnedUserId, Sender> {
279+ let mut unique: Vec<&OwnedUserId> = user_ids.collect();
280+ unique.sort();
281+ unique.dedup();
282+ join_all(unique.into_iter().map(|user_id| async move {
283+ let member = room.get_member(user_id).await.ok().flatten()?;
284+ Some((
285+ user_id.clone(),
286+ Sender {
287+ user_id: user_id.clone(),
288+ display_name: member.display_name().map(str::to_owned),
289+ avatar: member.avatar_url().map(MediaId::plain),
290+ name_ambiguous: member.name_ambiguous(),
291+ },
292+ ))
293+ }))
294+ .await
295+ .into_iter()
296+ .flatten()
297+ .collect()
249298 }
250299
251300 fn validate_limit(limit: u16) -> Result<(), Error> {
Modified file: src/lib.rs+9 additions−9 deletions
@@ -57,15 +57,15 @@ pub use mirror::{
5757 pub use model::{
5858 AccountCommand, AttachmentEncryption, AttachmentUpload, Change, Command, CommandOutput,
5959 ConnectionState, CreateRoomOptions, EncryptionCommand, EncryptionState, EngineErrorKind,
60- EngineFailure, Event, EventContext, EventContextCommand, EventContextDirection,
61- EventContextPage, EventRef, HistoryState, ImagePack, ImagePackSource, ImagePacks,
62- KeyBackupState, LatestEvent, LifecycleCommand, LinkPreview, LinkPreviewImage, Media,
63- MediaCommand, MediaDescriptor, MediaFile, MediaId, MediaKind, MediaLocation, MediaSize,
64- MemberFilter, MemberMembership, MemberSummary, MembershipChange, Message, MessageContent,
65- MessageEditContent, MessageKind, MessageMentions, NotificationCommand, OwnDevice, PackCommand,
66- PackImage, PackImageInfo, PackUsage, Patch, PowerLevelChanges, Profile, PushNotification,
67- PushNotificationEvent, PushNotificationRoom, PushNotificationSender, PushNotificationStatus,
68- Reaction, RecentHistoryState, RecoveryState, ReplyDetails, ReplyReference, Retain, RichText,
60+ EngineFailure, Event, EventContext, EventContextCommand, EventContextPage, EventRef,
61+ HistoryState, ImagePack, ImagePackSource, ImagePacks, KeyBackupState, LatestEvent,
62+ LifecycleCommand, LinkPreview, LinkPreviewImage, Media, MediaCommand, MediaDescriptor,
63+ MediaFile, MediaId, MediaKind, MediaLocation, MediaSize, MemberFilter, MemberMembership,
64+ MemberSummary, MembershipChange, Message, MessageContent, MessageEditContent, MessageKind,
65+ MessageMentions, NotificationCommand, OwnDevice, PackCommand, PackImage, PackImageInfo,
66+ PackUsage, Patch, PowerLevelChanges, Profile, PushNotification, PushNotificationEvent,
67+ PushNotificationRoom, PushNotificationSender, PushNotificationStatus, Reaction,
68+ RecentHistoryState, RecoveryState, ReplyDetails, ReplyReference, Retain, RichText,
6969 RoomCategory, RoomCommand, RoomHero, RoomListFilter, RoomMatcher, RoomMembership,
7070 RoomNotificationMode, RoomNotificationSettings, RoomPowerLevels, RoomPreset, RoomSummary,
7171 RoomVisibility, SendState, Sender, SpaceChild, Termination, ThreadSummary, TimelineCommand,
Modified file: src/mentions/state_tests.rs+1 additions−1 deletions
@@ -257,7 +257,7 @@ async fn context_uses_the_redacted_saved_event_instead_of_the_entry_preview() {
257257 );
258258
259259 let (context, media) = data.context(&mention).unwrap();
260- let saved = context.context.unwrap().event.unwrap();
260+ let saved = context.context.unwrap().event;
261261 let TimelineItem::Event(saved) = saved else {
262262 panic!("the saved target should remain a timeline event");
263263 };
Modified file: src/model.rs+2 additions−10 deletions
@@ -679,27 +679,19 @@ pub enum CommandOutput {
679679 #[non_exhaustive]
680680 pub struct EventContext {
681681 pub events_before: Vec<TimelineItem>,
682- pub event: Option<TimelineItem>,
682+ pub event: TimelineItem,
683683 pub events_after: Vec<TimelineItem>,
684684 pub start_token: Option<String>,
685685 pub end_token: Option<String>,
686686 }
687687
688-#[derive(Clone, Copy, Debug, Eq, PartialEq)]
689-#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
690-pub enum EventContextDirection {
691- Before,
692- After,
693-}
694-
695688 #[derive(Clone, Debug, Eq, PartialEq)]
696689 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
697690 #[non_exhaustive]
698691 pub struct EventContextPage {
699- pub direction: EventContextDirection,
700692 pub events: Vec<TimelineItem>,
693+ /// `None` once that end of the room is reached
701694 pub next_token: Option<String>,
702- pub end_reached: bool,
703695 }
704696
705697 #[derive(Clone, Debug, Eq, PartialEq)]
Added file: src/normalise/context.rs+249 additions
@@ -0,0 +1,249 @@
1+use std::collections::HashMap;
2+
3+use ruma::{MxcUri, OwnedEventId, OwnedUserId, RoomId, UserId};
4+use serde_json::Value;
5+
6+use super::{media::MediaRegistration, observed::normalise_observed};
7+use crate::{MediaId, Reaction, ReplyDetails, Sender, TimelineEvent, TimelineItem};
8+
9+pub(crate) struct ContextPage {
10+ /// each with the index of the raw event it came from
11+ pub(crate) items: Vec<(usize, TimelineItem)>,
12+ pub(crate) media: Vec<MediaRegistration>,
13+}
14+
15+struct Annotation<'a> {
16+ key: &'a str,
17+ sender: &'a str,
18+ shortcode: Option<&'a str>,
19+}
20+
21+/// raw events in timeline order, shown the way the live room timeline shows them. edits and
22+/// reactions fold into their targets within the page, and redactions, unmatched relations
23+/// and thread replies go. `keep` is never dropped
24+pub(crate) fn normalise_context(
25+ values: &[Value],
26+ keep: Option<usize>,
27+ current_user_id: Option<&UserId>,
28+ room_id: &RoomId,
29+) -> ContextPage {
30+ let mut edits: HashMap<&str, Vec<&Value>> = HashMap::new();
31+ let mut annotations: HashMap<&str, Vec<Annotation<'_>>> = HashMap::new();
32+ let mut shown = Vec::new();
33+ for (index, value) in values.iter().enumerate() {
34+ let str_at = |pointer: &str| value.pointer(pointer).and_then(Value::as_str);
35+ let target = str_at("/content/m.relates_to/event_id");
36+ if Some(index) == keep {
37+ shown.push(index);
38+ continue;
39+ }
40+ match (
41+ str_at("/type"),
42+ str_at("/content/m.relates_to/rel_type"),
43+ target,
44+ ) {
45+ (Some("m.room.redaction"), ..) => {}
46+ (Some("m.reaction"), Some("m.annotation"), Some(target)) => {
47+ if let (Some(key), Some(sender)) =
48+ (str_at("/content/m.relates_to/key"), str_at("/sender"))
49+ {
50+ annotations.entry(target).or_default().push(Annotation {
51+ key,
52+ sender,
53+ shortcode: str_at("/content/com.beeper.reaction.shortcode")
54+ .or_else(|| str_at("/content/shortcode")),
55+ });
56+ }
57+ }
58+ (Some("m.reaction"), ..) => {}
59+ (_, Some("m.replace"), Some(target)) => edits.entry(target).or_default().push(value),
60+ (_, Some("m.thread"), _) => {}
61+ _ => shown.push(index),
62+ }
63+ }
64+ let mut media = Vec::new();
65+ let items = shown
66+ .into_iter()
67+ .filter_map(|index| {
68+ let mut value = std::borrow::Cow::Borrowed(&values[index]);
69+ let event_id = value.get("event_id")?.as_str()?.to_owned();
70+ let latest_edit = edits.get(event_id.as_str()).and_then(|edits| {
71+ edits
72+ .iter()
73+ .rev()
74+ .find(|edit| edit.get("sender") == value.get("sender"))
75+ });
76+ if let Some(edit) = latest_edit.copied()
77+ && value.pointer("/unsigned/m.relations/m.replace").is_none()
78+ {
79+ let unsigned = value
80+ .to_mut()
81+ .as_object_mut()?
82+ .entry("unsigned")
83+ .or_insert_with(|| Value::Object(Default::default()));
84+ let relations = unsigned
85+ .as_object_mut()?
86+ .entry("m.relations")
87+ .or_insert_with(|| Value::Object(Default::default()));
88+ relations
89+ .as_object_mut()?
90+ .insert("m.replace".to_owned(), edit.clone());
91+ }
92+ let identity = format!("{room_id}\n{event_id}");
93+ let (mut item, registrations) = normalise_observed(&value, current_user_id, &identity)?;
94+ media.extend(registrations);
95+ if let (TimelineItem::Event(event), Some(annotations)) =
96+ (&mut item, annotations.get(event_id.as_str()))
97+ {
98+ event.reactions = reactions(annotations, current_user_id);
99+ }
100+ Some((index, item))
101+ })
102+ .collect();
103+ ContextPage { items, media }
104+}
105+
106+fn reactions(annotations: &[Annotation<'_>], current_user_id: Option<&UserId>) -> Vec<Reaction> {
107+ let mut reactions: Vec<Reaction> = Vec::new();
108+ for annotation in annotations {
109+ let Ok(sender) = OwnedUserId::try_from(annotation.sender) else {
110+ continue;
111+ };
112+ let index = match reactions.iter().position(|r| r.key == annotation.key) {
113+ Some(index) => index,
114+ None => {
115+ reactions.push(Reaction {
116+ key: annotation.key.to_owned(),
117+ shortcode: None,
118+ image: <&MxcUri>::from(annotation.key)
119+ .is_valid()
120+ .then(|| MediaId::plain(annotation.key)),
121+ count: 0,
122+ own: false,
123+ senders: Vec::new(),
124+ });
125+ reactions.len() - 1
126+ }
127+ };
128+ let reaction = &mut reactions[index];
129+ if reaction.senders.contains(&sender) {
130+ continue;
131+ }
132+ reaction.own |= current_user_id == Some(sender.as_ref());
133+ reaction.senders.push(sender);
134+ reaction.count = reaction.senders.len();
135+ if reaction.shortcode.is_none() {
136+ reaction.shortcode = annotation.shortcode.map(str::to_owned);
137+ }
138+ }
139+ reactions
140+}
141+
142+/// the sender's member data, for each event and every reply quoted in it
143+pub(crate) fn apply_profiles<'a>(
144+ events: impl Iterator<Item = &'a mut TimelineEvent>,
145+ profiles: &HashMap<OwnedUserId, Sender>,
146+) {
147+ for event in events {
148+ if let Some(sender) = profiles.get(&event.sender.user_id) {
149+ event.sender = sender.clone();
150+ }
151+ }
152+}
153+
154+/// settles every reply, from `targets` or as unavailable
155+pub(crate) fn resolve_replies<'a>(
156+ events: impl Iterator<Item = &'a mut TimelineEvent>,
157+ targets: &HashMap<OwnedEventId, TimelineEvent>,
158+) {
159+ for event in events {
160+ let Some(reply) = event.reply_to.as_mut() else {
161+ continue;
162+ };
163+ if !matches!(reply.details, ReplyDetails::Pending) {
164+ continue;
165+ }
166+ reply.details = match targets.get(&reply.event_id) {
167+ Some(target) => ReplyDetails::Ready {
168+ sender: target.sender.clone(),
169+ content: Box::new(target.content.clone()),
170+ timestamp_ms: target.timestamp_ms,
171+ },
172+ None => ReplyDetails::Unavailable,
173+ };
174+ }
175+}
176+
177+#[cfg(test)]
178+mod tests {
179+ use ruma::{room_id, user_id};
180+ use serde_json::json;
181+
182+ use super::*;
183+ use crate::TimelineContent;
184+
185+ fn message(id: &str, sender: &str, body: &str) -> Value {
186+ json!({
187+ "type": "m.room.message",
188+ "event_id": id,
189+ "sender": sender,
190+ "origin_server_ts": 1,
191+ "content": { "msgtype": "m.text", "body": body },
192+ })
193+ }
194+
195+ #[test]
196+ fn edits_and_reactions_fold_into_their_targets() {
197+ let mut edit = message("$edit", "@a:x", "* fixed");
198+ edit["content"]["m.new_content"] = json!({ "msgtype": "m.text", "body": "fixed" });
199+ edit["content"]["m.relates_to"] = json!({ "rel_type": "m.replace", "event_id": "$orig" });
200+ let mut foreign_edit = message("$foreign", "@b:x", "* hijack");
201+ foreign_edit["content"]["m.new_content"] = json!({ "msgtype": "m.text", "body": "hijack" });
202+ foreign_edit["content"]["m.relates_to"] =
203+ json!({ "rel_type": "m.replace", "event_id": "$orig" });
204+ let reaction = |id: &str, sender: &str| {
205+ json!({
206+ "type": "m.reaction",
207+ "event_id": id,
208+ "sender": sender,
209+ "origin_server_ts": 1,
210+ "content": { "m.relates_to": {
211+ "rel_type": "m.annotation", "event_id": "$orig", "key": "👍",
212+ }},
213+ })
214+ };
215+ let mut threaded = message("$threaded", "@b:x", "in a thread");
216+ threaded["content"]["m.relates_to"] =
217+ json!({ "rel_type": "m.thread", "event_id": "$orig" });
218+ let values = [
219+ message("$orig", "@a:x", "typo"),
220+ edit,
221+ foreign_edit,
222+ reaction("$r1", "@b:x"),
223+ reaction("$r2", "@b:x"),
224+ reaction("$r3", "@a:x"),
225+ threaded,
226+ json!({
227+ "type": "m.room.redaction",
228+ "event_id": "$redaction",
229+ "sender": "@a:x",
230+ "origin_server_ts": 1,
231+ "redacts": "$gone",
232+ "content": {},
233+ }),
234+ ];
235+ let page = normalise_context(&values, None, Some(user_id!("@a:x")), room_id!("!room:x"));
236+ assert_eq!(page.items.len(), 1);
237+ let TimelineItem::Event(event) = &page.items[0].1 else {
238+ panic!("expected an event");
239+ };
240+ let TimelineContent::Message(message) = &event.content else {
241+ panic!("expected a message");
242+ };
243+ assert_eq!(message.body, "fixed");
244+ assert!(message.edited);
245+ assert_eq!(event.reactions.len(), 1);
246+ assert_eq!(event.reactions[0].count, 2);
247+ assert!(event.reactions[0].own);
248+ }
249+}
Modified file: src/normalise/mod.rs+2 additions
@@ -1,3 +1,4 @@
1+mod context;
12 mod image_pack;
23 mod link_preview;
34 mod media;
@@ -6,6 +7,7 @@ mod observed;
67 mod room;
78 mod timeline;
89
10+pub(crate) use context::{apply_profiles, normalise_context, resolve_replies};
911 pub(crate) use image_pack::{
1012 ENABLED_ROOMS, ROOM_PACKS, USER_PACK, can_edit_packs, image_packs, resolve_pack_id, stored_pack,
1113 };
Modified file: src/transport.rs+8 additions−8 deletions
@@ -362,15 +362,15 @@ impl ClientHandle {
362362 &self,
363363 room_id: &RoomId,
364364 event_id: &EventId,
365- before_limit: u32,
366- after_limit: u32,
365+ before_limit: u16,
366+ after_limit: u16,
367367 ) -> Result<EventContext, Error> {
368368 match self
369369 .execute(Command::EventContext(EventContextCommand::Fetch {
370370 room_id: room_id.to_owned(),
371371 event_id: event_id.to_owned(),
372- before_limit: before_limit.min(50) as u16,
373- after_limit: after_limit.min(50) as u16,
372+ before_limit,
373+ after_limit,
374374 }))
375375 .await?
376376 {
@@ -383,11 +383,11 @@ impl ClientHandle {
383383 &self,
384384 room_id: &RoomId,
385385 token: &str,
386- limit: u32,
386+ limit: u16,
387387 ) -> Result<EventContextPage, Error> {
388388 self.event_context_page(EventContextCommand::Before {
389389 room_id: room_id.to_owned(),
390- limit: limit.min(50) as u16,
390+ limit,
391391 token: token.to_owned(),
392392 })
393393 .await
@@ -397,11 +397,11 @@ impl ClientHandle {
397397 &self,
398398 room_id: &RoomId,
399399 token: &str,
400- limit: u32,
400+ limit: u16,
401401 ) -> Result<EventContextPage, Error> {
402402 self.event_context_page(EventContextCommand::After {
403403 room_id: room_id.to_owned(),
404- limit: limit.min(50) as u16,
404+ limit,
405405 token: token.to_owned(),
406406 })
407407 .await