macltok

public

Star0

Watch0

90 commits 3 branches 0 tags 0 releases3.0 MiB

use std::{
    collections::{HashMap, VecDeque},
    sync::Arc,
};

use futures_util::future::{join_all, try_join};
use matrix_sdk::{
    Client, Room, RoomState, deserialized_responses::TimelineEvent as SdkTimelineEvent,
    room::MessagesOptions,
};
use ruma::{OwnedEventId, OwnedRoomId, OwnedUserId, UInt, push::Action};
use serde_json::Value;
use tokio::sync::{Mutex, broadcast::error::RecvError};

use crate::{
    Error, ErrorKind, Event, EventContext, EventContextCloseReason, EventContextCommand,
    EventContextPage, EventContextSession, Sender, TimelineEvent, TimelineItem,
    engine::MediaService,
    normalise::{
        MediaRegistration, MediaScope, apply_profiles, member_sender, normalise_context,
        normalise_observed, resolve_replies,
    },
    transport::EventSender,
};

const MAX_CONTEXT_LIMIT: u16 = 50;
/// a consumer that never closes its jump views still holds at most this many
const MAX_SESSIONS: usize = 32;

/// open event contexts, oldest first, each owning the media its pages registered
pub(crate) struct Sessions {
    open: Mutex<VecDeque<Session>>,
    events: EventSender,
}

struct Session {
    id: String,
    room_id: OwnedRoomId,
    scope: MediaScope,
}

impl Session {
    /// media ids carry the session, so two owners of one event never share an entry
    fn media_prefix(scope: &MediaScope) -> String {
        format!("{}#", scope.nonce())
    }
}

impl Sessions {
    pub(crate) fn new(events: EventSender) -> Self {
        Self {
            open: Mutex::default(),
            events,
        }
    }

    async fn open(
        &self,
        media: &MediaService,
        room_id: OwnedRoomId,
        scope: MediaScope,
        registrations: Vec<MediaRegistration>,
    ) -> String {
        let id = format!("context-{}", scope.nonce());
        let mut evicted = Vec::new();
        {
            let mut open = self.open.lock().await;
            media.register(&scope, registrations).await;
            open.push_back(Session {
                id: id.clone(),
                room_id,
                scope,
            });
            while open.len() > MAX_SESSIONS {
                if let Some(oldest) = open.pop_front() {
                    media.release(&oldest.scope).await;
                    evicted.push(oldest.id);
                }
            }
        }
        self.announce(evicted, EventContextCloseReason::Evicted)
            .await;
        id
    }

    /// the room and the media prefix the session's pages use
    async fn find(&self, id: &str) -> Result<(OwnedRoomId, String), Error> {
        let open = self.open.lock().await;
        let session = open
            .iter()
            .find(|session| session.id == id)
            .ok_or_else(not_open)?;
        Ok((
            session.room_id.clone(),
            Session::media_prefix(&session.scope),
        ))
    }

    /// under the lock, so a page that lands after its close registers nothing
    async fn register(
        &self,
        media: &MediaService,
        id: &str,
        registrations: Vec<MediaRegistration>,
    ) -> Result<(), Error> {
        let open = self.open.lock().await;
        let session = open
            .iter()
            .find(|session| session.id == id)
            .ok_or_else(not_open)?;
        media.register(&session.scope, registrations).await;
        Ok(())
    }

    async fn close(&self, media: &MediaService, id: &str) -> Result<(), Error> {
        let mut open = self.open.lock().await;
        let index = open
            .iter()
            .position(|session| session.id == id)
            .ok_or_else(not_open)?;
        if let Some(session) = open.remove(index) {
            media.release(&session.scope).await;
        }
        Ok(())
    }

    async fn close_left(&self, media: &MediaService, client: &Client) {
        let mut closed = Vec::new();
        {
            let mut open = self.open.lock().await;
            let mut kept = VecDeque::with_capacity(open.len());
            for session in open.drain(..) {
                let left = client.get_room(&session.room_id).is_none_or(|room| {
                    matches!(room.state(), RoomState::Left | RoomState::Banned)
                });
                if left {
                    media.release(&session.scope).await;
                    closed.push(session.id);
                } else {
                    kept.push_back(session);
                }
            }
            *open = kept;
        }
        self.announce(closed, EventContextCloseReason::RoomLeft)
            .await;
    }

    async fn announce(&self, ids: Vec<String>, reason: EventContextCloseReason) {
        for context_id in ids {
            let event = Event::EventContextClosed {
                context_id,
                reason: reason.clone(),
            };
            if self.events.send(event).await.is_err() {
                return;
            }
        }
    }

    /// the media service is cleared beside this, so nothing is released one by one
    pub(crate) async fn clear(&self) {
        self.open.lock().await.clear();
    }
}

fn not_open() -> Error {
    Error::new(
        ErrorKind::EventContextNotOpen,
        "event context is closed or was never opened",
    )
}

/// leaving a room closes its event contexts
pub(super) async fn close_on_leave(
    client: Client,
    media: Arc<MediaService>,
    sessions: Arc<Sessions>,
) {
    let mut info = client.room_info_notable_update_receiver();
    loop {
        match info.recv().await {
            Ok(_) | Err(RecvError::Lagged(_)) => sessions.close_left(&media, &client).await,
            Err(RecvError::Closed) => return,
        }
    }
}

pub(crate) async fn execute(
    client: Client,
    media: Arc<MediaService>,
    sessions: Arc<Sessions>,
    command: EventContextCommand,
) -> Result<crate::CommandOutput, Error> {
    let current_user_id = client.user_id().map(ToOwned::to_owned);
    let room = |room_id: &OwnedRoomId| client.get_room(room_id).ok_or_else(Error::room_unknown);
    let (context_id, token, limit, backwards) = match command {
        EventContextCommand::Fetch {
            room_id,
            event_id,
            before_limit,
            after_limit,
        } => {
            let scope = media.context_scope();
            let (context, registrations, _) = capture_in(
                room(&room_id)?,
                current_user_id,
                event_id,
                before_limit,
                after_limit,
                &Session::media_prefix(&scope),
            )
            .await?;
            let context_id = sessions.open(&media, room_id, scope, registrations).await;
            return Ok(crate::CommandOutput::EventContext(EventContextSession {
                context_id,
                context,
            }));
        }
        EventContextCommand::Close { context_id } => {
            sessions.close(&media, &context_id).await?;
            return Ok(crate::CommandOutput::Done);
        }
        EventContextCommand::Before {
            context_id,
            token,
            limit,
        } => (context_id, token, limit, true),
        EventContextCommand::After {
            context_id,
            token,
            limit,
        } => (context_id, token, limit, false),
    };
    if limit == 0 {
        return Err(Error::new(
            ErrorKind::Invalid,
            "context page limit must be at least 1",
        ));
    }
    validate_limit(limit)?;
    if token.is_empty() {
        return Err(Error::new(
            ErrorKind::Invalid,
            "context pagination token is empty",
        ));
    }
    let (room_id, media_prefix) = sessions.find(&context_id).await?;
    let room = room(&room_id)?;
    let side = fetch_side(&room, Some(&token), limit, backwards).await?;
    let mut page = Assembly::new(
        &room,
        current_user_id.as_deref(),
        &side.values,
        &side.highlights,
        None,
        &media_prefix,
    )
    .await;
    let events = page.take(0..side.values.len());
    sessions.register(&media, &context_id, page.media).await?;
    Ok(crate::CommandOutput::EventContextPage(EventContextPage {
        events,
        next_token: side.token,
    }))
}

pub(crate) async fn capture(
    room: Room,
    current_user_id: Option<OwnedUserId>,
    event_id: OwnedEventId,
    before_limit: u16,
    after_limit: u16,
) -> Result<(EventContext, Vec<MediaRegistration>, Vec<Value>), Error> {
    capture_in(room, current_user_id, event_id, before_limit, after_limit, "").await
}

async fn capture_in(
    room: Room,
    current_user_id: Option<OwnedUserId>,
    event_id: OwnedEventId,
    before_limit: u16,
    after_limit: u16,
    media_prefix: &str,
) -> Result<(EventContext, Vec<MediaRegistration>, Vec<Value>), Error> {
    validate_limit(before_limit)?;
    validate_limit(after_limit)?;
    let raw = room
        .event_with_context(&event_id, false, UInt::from(0_u16), None)
        .await
        .map_err(|source| {
            let error = Error::from_sdk("fetching event context failed", source);
            if error
                .server
                .as_ref()
                .is_some_and(|server| server.status == 404)
            {
                Error::new(ErrorKind::EventNotFound, "event was not found")
            } else {
                error
            }
        })?;
    let (target, target_highlight) = raw
        .event
        .as_ref()
        .and_then(raw_event)
        .ok_or_else(|| Error::new(ErrorKind::EventNotFound, "event was not found"))?;
    let (before, after) = try_join(
        fetch_side(&room, raw.prev_batch_token.as_deref(), before_limit, true),
        fetch_side(&room, raw.next_batch_token.as_deref(), after_limit, false),
    )
    .await?;
    let target_index = before.values.len();
    let mut values = before.values;
    values.push(target);
    values.extend(after.values);
    let mut highlights = before.highlights;
    highlights.push(target_highlight);
    highlights.extend(after.highlights);
    let mut page = Assembly::new(
        &room,
        current_user_id.as_deref(),
        &values,
        &highlights,
        Some(target_index),
        media_prefix,
    )
    .await;
    let events_before = page.take(0..target_index);
    let event = page
        .take(target_index..target_index + 1)
        .pop()
        .ok_or_else(|| Error::new(ErrorKind::EventNotFound, "event could not be read"))?;
    let events_after = page.take(target_index + 1..values.len());
    Ok((
        EventContext {
            events_before,
            event,
            events_after,
            start_token: before.token,
            end_token: after.token,
        },
        page.media,
        values,
    ))
}

struct Side {
    /// in timeline order
    values: Vec<Value>,
    /// beside `values`, from the push actions the sdk worked out as it fetched
    highlights: Vec<bool>,
    token: Option<String>,
}

fn raw_event(event: &SdkTimelineEvent) -> Option<(Value, bool)> {
    let value = serde_json::from_str(event.raw().json().get()).ok()?;
    let highlight = event
        .push_actions()
        .is_some_and(|actions| actions.iter().any(Action::is_highlight));
    Some((value, highlight))
}

async fn fetch_side(
    room: &Room,
    token: Option<&str>,
    limit: u16,
    backwards: bool,
) -> Result<Side, Error> {
    let Some(token) = token else {
        return Ok(Side {
            values: Vec::new(),
            highlights: Vec::new(),
            token: None,
        });
    };
    if limit == 0 {
        return Ok(Side {
            values: Vec::new(),
            highlights: Vec::new(),
            token: Some(token.to_owned()),
        });
    }
    let mut options = if backwards {
        MessagesOptions::backward()
    } else {
        MessagesOptions::forward()
    };
    options = options.from(token);
    options.limit = UInt::from(limit);
    let messages = room
        .messages(options)
        .await
        .map_err(|source| Error::from_sdk("fetching event context events failed", source))?;
    let mut events: Vec<_> = messages.chunk.iter().filter_map(raw_event).collect();
    if backwards {
        events.reverse();
    }
    let (values, highlights) = events.into_iter().unzip();
    Ok(Side {
        values,
        highlights,
        token: messages.end,
    })
}

/// a window of raw events normalised together, so relations inside it land on their targets
struct Assembly {
    items: Vec<(usize, TimelineItem)>,
    media: Vec<MediaRegistration>,
}

impl Assembly {
    async fn new(
        room: &Room,
        current_user_id: Option<&ruma::UserId>,
        values: &[Value],
        highlights: &[bool],
        keep: Option<usize>,
        media_prefix: &str,
    ) -> Self {
        let page = normalise_context(
            values,
            keep,
            current_user_id,
            room.room_id(),
            media_prefix,
        );
        let mut items = page.items;
        for (index, item) in &mut items {
            if let TimelineItem::Event(event) = item {
                event.highlight = highlights[*index];
            }
        }
        let mut media = page.media;
        let mut targets: HashMap<OwnedEventId, TimelineEvent> = HashMap::new();
        let mut missing = Vec::new();
        for (_, item) in &items {
            if let TimelineItem::Event(event) = item
                && let Some(event_id) = &event.event_id
            {
                targets.insert(event_id.clone(), (**event).clone());
            }
        }
        for event in events(&mut items) {
            if let Some(reply) = &event.reply_to
                && !targets.contains_key(&reply.event_id)
                && !missing.contains(&reply.event_id)
            {
                missing.push(reply.event_id.clone());
            }
        }
        let fetched = join_all(
            missing
                .into_iter()
                .map(|event_id| fetch_reply(room, current_user_id, event_id, media_prefix)),
        )
        .await;
        for (event, registrations) in fetched.into_iter().flatten() {
            media.extend(registrations);
            if let Some(event_id) = event.event_id.clone() {
                targets.insert(event_id, event);
            }
        }
        let profiles = profiles(
            room,
            items
                .iter()
                .filter_map(|(_, item)| match item {
                    TimelineItem::Event(event) => Some(&event.sender.user_id),
                    _ => None,
                })
                .chain(targets.values().map(|event| &event.sender.user_id)),
        )
        .await;
        apply_profiles(targets.values_mut(), &profiles);
        apply_profiles(events(&mut items), &profiles);
        resolve_replies(events(&mut items), &targets);
        Self { items, media }
    }

    /// the items that came from raw events in `range`
    fn take(&mut self, range: std::ops::Range<usize>) -> Vec<TimelineItem> {
        let (taken, kept) = std::mem::take(&mut self.items)
            .into_iter()
            .partition(|(index, _)| range.contains(index));
        self.items = kept;
        taken.into_iter().map(|(_, item)| item).collect()
    }
}

fn events(items: &mut [(usize, TimelineItem)]) -> impl Iterator<Item = &mut TimelineEvent> {
    items.iter_mut().filter_map(|(_, item)| match item {
        TimelineItem::Event(event) => Some(&mut **event),
        _ => None,
    })
}

async fn fetch_reply(
    room: &Room,
    current_user_id: Option<&ruma::UserId>,
    event_id: OwnedEventId,
    media_prefix: &str,
) -> Option<(TimelineEvent, Vec<MediaRegistration>)> {
    let event = room
        .load_or_fetch_event(&event_id, None)
        .await
        .inspect_err(|error| tracing::debug!(%event_id, %error, "fetching a reply target failed"))
        .ok()?;
    let value: Value = serde_json::from_str(event.raw().json().get()).ok()?;
    let identity = format!("{media_prefix}{}\n{event_id}", room.room_id());
    match normalise_observed(&value, current_user_id, &identity)? {
        (TimelineItem::Event(event), media) => Some((*event, media)),
        _ => None,
    }
}

/// under lazy-loaded members a sender may be unknown locally, and `get_member` fetches the
/// member list once for the room when it is
async fn profiles<'a>(
    room: &Room,
    user_ids: impl Iterator<Item = &'a OwnedUserId>,
) -> HashMap<OwnedUserId, Sender> {
    let mut unique: Vec<&OwnedUserId> = user_ids.collect();
    unique.sort();
    unique.dedup();
    join_all(unique.into_iter().map(|user_id| async move {
        Some((user_id.clone(), member_sender(room, user_id).await?))
    }))
    .await
    .into_iter()
    .flatten()
    .collect()
}

fn validate_limit(limit: u16) -> Result<(), Error> {
    if limit > MAX_CONTEXT_LIMIT {
        Err(Error::new(ErrorKind::Invalid, "context limit exceeds 50"))
    } else {
        Ok(())
    }
}

1	use std::{
2	collections::{HashMap, VecDeque},
3	sync::Arc,
4	};
5
6	use futures_util::future::{join_all, try_join};
7	use matrix_sdk::{
8	Client, Room, RoomState, deserialized_responses::TimelineEvent as SdkTimelineEvent,
9	room::MessagesOptions,
10	};
11	use ruma::{OwnedEventId, OwnedRoomId, OwnedUserId, UInt, push::Action};
12	use serde_json::Value;
13	use tokio::sync::{Mutex, broadcast::error::RecvError};
14
15	use crate::{
16	Error, ErrorKind, Event, EventContext, EventContextCloseReason, EventContextCommand,
17	EventContextPage, EventContextSession, Sender, TimelineEvent, TimelineItem,
18	engine::MediaService,
19	normalise::{
20	MediaRegistration, MediaScope, apply_profiles, member_sender, normalise_context,
21	normalise_observed, resolve_replies,
22	},
23	transport::EventSender,
24	};
25
26	const MAX_CONTEXT_LIMIT: u16 = 50;
27	/// a consumer that never closes its jump views still holds at most this many
28	const MAX_SESSIONS: usize = 32;
29
30	/// open event contexts, oldest first, each owning the media its pages registered
31	pub(crate) struct Sessions {
32	open: Mutex<VecDeque<Session>>,
33	events: EventSender,
34	}
35
36	struct Session {
37	id: String,
38	room_id: OwnedRoomId,
39	scope: MediaScope,
40	}
41
42	impl Session {
43	/// media ids carry the session, so two owners of one event never share an entry
44	fn media_prefix(scope: &MediaScope) -> String {
45	format!("{}#", scope.nonce())
46	}
47	}
48
49	impl Sessions {
50	pub(crate) fn new(events: EventSender) -> Self {
51	Self {
52	open: Mutex::default(),
53	events,
54	}
55	}
56
57	async fn open(
58	&self,
59	media: &MediaService,
60	room_id: OwnedRoomId,
61	scope: MediaScope,
62	registrations: Vec<MediaRegistration>,
63	) -> String {
64	let id = format!("context-{}", scope.nonce());
65	let mut evicted = Vec::new();
66	{
67	let mut open = self.open.lock().await;
68	media.register(&scope, registrations).await;
69	open.push_back(Session {
70	id: id.clone(),
71	room_id,
72	scope,
73	});
74	while open.len() > MAX_SESSIONS {
75	if let Some(oldest) = open.pop_front() {
76	media.release(&oldest.scope).await;
77	evicted.push(oldest.id);
78	}
79	}
80	}
81	self.announce(evicted, EventContextCloseReason::Evicted)
82	.await;
83	id
84	}
85
86	/// the room and the media prefix the session's pages use
87	async fn find(&self, id: &str) -> Result<(OwnedRoomId, String), Error> {
88	let open = self.open.lock().await;
89	let session = open
90	.iter()
91	.find(\|session\| session.id == id)
92	.ok_or_else(not_open)?;
93	Ok((
94	session.room_id.clone(),
95	Session::media_prefix(&session.scope),
96	))
97	}
98
99	/// under the lock, so a page that lands after its close registers nothing
100	async fn register(
101	&self,
102	media: &MediaService,
103	id: &str,
104	registrations: Vec<MediaRegistration>,
105	) -> Result<(), Error> {
106	let open = self.open.lock().await;
107	let session = open
108	.iter()
109	.find(\|session\| session.id == id)
110	.ok_or_else(not_open)?;
111	media.register(&session.scope, registrations).await;
112	Ok(())
113	}
114
115	async fn close(&self, media: &MediaService, id: &str) -> Result<(), Error> {
116	let mut open = self.open.lock().await;
117	let index = open
118	.iter()
119	.position(\|session\| session.id == id)
120	.ok_or_else(not_open)?;
121	if let Some(session) = open.remove(index) {
122	media.release(&session.scope).await;
123	}
124	Ok(())
125	}
126
127	async fn close_left(&self, media: &MediaService, client: &Client) {
128	let mut closed = Vec::new();
129	{
130	let mut open = self.open.lock().await;
131	let mut kept = VecDeque::with_capacity(open.len());
132	for session in open.drain(..) {
133	let left = client.get_room(&session.room_id).is_none_or(\|room\| {
134	matches!(room.state(), RoomState::Left \| RoomState::Banned)
135	});
136	if left {
137	media.release(&session.scope).await;
138	closed.push(session.id);
139	} else {
140	kept.push_back(session);
141	}
142	}
143	*open = kept;
144	}
145	self.announce(closed, EventContextCloseReason::RoomLeft)
146	.await;
147	}
148
149	async fn announce(&self, ids: Vec<String>, reason: EventContextCloseReason) {
150	for context_id in ids {
151	let event = Event::EventContextClosed {
152	context_id,
153	reason: reason.clone(),
154	};
155	if self.events.send(event).await.is_err() {
156	return;
157	}
158	}
159	}
160
161	/// the media service is cleared beside this, so nothing is released one by one
162	pub(crate) async fn clear(&self) {
163	self.open.lock().await.clear();
164	}
165	}
166
167	fn not_open() -> Error {
168	Error::new(
169	ErrorKind::EventContextNotOpen,
170	"event context is closed or was never opened",
171	)
172	}
173
174	/// leaving a room closes its event contexts
175	pub(super) async fn close_on_leave(
176	client: Client,
177	media: Arc<MediaService>,
178	sessions: Arc<Sessions>,
179	) {
180	let mut info = client.room_info_notable_update_receiver();
181	loop {
182	match info.recv().await {
183	Ok(_) \| Err(RecvError::Lagged(_)) => sessions.close_left(&media, &client).await,
184	Err(RecvError::Closed) => return,
185	}
186	}
187	}
188
189	pub(crate) async fn execute(
190	client: Client,
191	media: Arc<MediaService>,
192	sessions: Arc<Sessions>,
193	command: EventContextCommand,
194	) -> Result<crate::CommandOutput, Error> {
195	let current_user_id = client.user_id().map(ToOwned::to_owned);
196	let room = \|room_id: &OwnedRoomId\| client.get_room(room_id).ok_or_else(Error::room_unknown);
197	let (context_id, token, limit, backwards) = match command {
198	EventContextCommand::Fetch {
199	room_id,
200	event_id,
201	before_limit,
202	after_limit,
203	} => {
204	let scope = media.context_scope();
205	let (context, registrations, _) = capture_in(
206	room(&room_id)?,
207	current_user_id,
208	event_id,
209	before_limit,
210	after_limit,
211	&Session::media_prefix(&scope),
212	)
213	.await?;
214	let context_id = sessions.open(&media, room_id, scope, registrations).await;
215	return Ok(crate::CommandOutput::EventContext(EventContextSession {
216	context_id,
217	context,
218	}));
219	}
220	EventContextCommand::Close { context_id } => {
221	sessions.close(&media, &context_id).await?;
222	return Ok(crate::CommandOutput::Done);
223	}
224	EventContextCommand::Before {
225	context_id,
226	token,
227	limit,
228	} => (context_id, token, limit, true),
229	EventContextCommand::After {
230	context_id,
231	token,
232	limit,
233	} => (context_id, token, limit, false),
234	};
235	if limit == 0 {
236	return Err(Error::new(
237	ErrorKind::Invalid,
238	"context page limit must be at least 1",
239	));
240	}
241	validate_limit(limit)?;
242	if token.is_empty() {
243	return Err(Error::new(
244	ErrorKind::Invalid,
245	"context pagination token is empty",
246	));
247	}
248	let (room_id, media_prefix) = sessions.find(&context_id).await?;
249	let room = room(&room_id)?;
250	let side = fetch_side(&room, Some(&token), limit, backwards).await?;
251	let mut page = Assembly::new(
252	&room,
253	current_user_id.as_deref(),
254	&side.values,
255	&side.highlights,
256	None,
257	&media_prefix,
258	)
259	.await;
260	let events = page.take(0..side.values.len());
261	sessions.register(&media, &context_id, page.media).await?;
262	Ok(crate::CommandOutput::EventContextPage(EventContextPage {
263	events,
264	next_token: side.token,
265	}))
266	}
267
268	pub(crate) async fn capture(
269	room: Room,
270	current_user_id: Option<OwnedUserId>,
271	event_id: OwnedEventId,
272	before_limit: u16,
273	after_limit: u16,
274	) -> Result<(EventContext, Vec<MediaRegistration>, Vec<Value>), Error> {
275	capture_in(room, current_user_id, event_id, before_limit, after_limit, "").await
276	}
277
278	async fn capture_in(
279	room: Room,
280	current_user_id: Option<OwnedUserId>,
281	event_id: OwnedEventId,
282	before_limit: u16,
283	after_limit: u16,
284	media_prefix: &str,
285	) -> Result<(EventContext, Vec<MediaRegistration>, Vec<Value>), Error> {
286	validate_limit(before_limit)?;
287	validate_limit(after_limit)?;
288	let raw = room
289	.event_with_context(&event_id, false, UInt::from(0_u16), None)
290	.await
291	.map_err(\|source\| {
292	let error = Error::from_sdk("fetching event context failed", source);
293	if error
294	.server
295	.as_ref()
296	.is_some_and(\|server\| server.status == 404)
297	{
298	Error::new(ErrorKind::EventNotFound, "event was not found")
299	} else {
300	error
301	}
302	})?;
303	let (target, target_highlight) = raw
304	.event
305	.as_ref()
306	.and_then(raw_event)
307	.ok_or_else(\|\| Error::new(ErrorKind::EventNotFound, "event was not found"))?;
308	let (before, after) = try_join(
309	fetch_side(&room, raw.prev_batch_token.as_deref(), before_limit, true),
310	fetch_side(&room, raw.next_batch_token.as_deref(), after_limit, false),
311	)
312	.await?;
313	let target_index = before.values.len();
314	let mut values = before.values;
315	values.push(target);
316	values.extend(after.values);
317	let mut highlights = before.highlights;
318	highlights.push(target_highlight);
319	highlights.extend(after.highlights);
320	let mut page = Assembly::new(
321	&room,
322	current_user_id.as_deref(),
323	&values,
324	&highlights,
325	Some(target_index),
326	media_prefix,
327	)
328	.await;
329	let events_before = page.take(0..target_index);
330	let event = page
331	.take(target_index..target_index + 1)
332	.pop()
333	.ok_or_else(\|\| Error::new(ErrorKind::EventNotFound, "event could not be read"))?;
334	let events_after = page.take(target_index + 1..values.len());
335	Ok((
336	EventContext {
337	events_before,
338	event,
339	events_after,
340	start_token: before.token,
341	end_token: after.token,
342	},
343	page.media,
344	values,
345	))
346	}
347
348	struct Side {
349	/// in timeline order
350	values: Vec<Value>,
351	/// beside `values`, from the push actions the sdk worked out as it fetched
352	highlights: Vec<bool>,
353	token: Option<String>,
354	}
355
356	fn raw_event(event: &SdkTimelineEvent) -> Option<(Value, bool)> {
357	let value = serde_json::from_str(event.raw().json().get()).ok()?;
358	let highlight = event
359	.push_actions()
360	.is_some_and(\|actions\| actions.iter().any(Action::is_highlight));
361	Some((value, highlight))
362	}
363
364	async fn fetch_side(
365	room: &Room,
366	token: Option<&str>,
367	limit: u16,
368	backwards: bool,
369	) -> Result<Side, Error> {
370	let Some(token) = token else {
371	return Ok(Side {
372	values: Vec::new(),
373	highlights: Vec::new(),
374	token: None,
375	});
376	};
377	if limit == 0 {
378	return Ok(Side {
379	values: Vec::new(),
380	highlights: Vec::new(),
381	token: Some(token.to_owned()),
382	});
383	}
384	let mut options = if backwards {
385	MessagesOptions::backward()
386	} else {
387	MessagesOptions::forward()
388	};
389	options = options.from(token);
390	options.limit = UInt::from(limit);
391	let messages = room
392	.messages(options)
393	.await
394	.map_err(\|source\| Error::from_sdk("fetching event context events failed", source))?;
395	let mut events: Vec<_> = messages.chunk.iter().filter_map(raw_event).collect();
396	if backwards {
397	events.reverse();
398	}
399	let (values, highlights) = events.into_iter().unzip();
400	Ok(Side {
401	values,
402	highlights,
403	token: messages.end,
404	})
405	}
406
407	/// a window of raw events normalised together, so relations inside it land on their targets
408	struct Assembly {
409	items: Vec<(usize, TimelineItem)>,
410	media: Vec<MediaRegistration>,
411	}
412
413	impl Assembly {
414	async fn new(
415	room: &Room,
416	current_user_id: Option<&ruma::UserId>,
417	values: &[Value],
418	highlights: &[bool],
419	keep: Option<usize>,
420	media_prefix: &str,
421	) -> Self {
422	let page = normalise_context(
423	values,
424	keep,
425	current_user_id,
426	room.room_id(),
427	media_prefix,
428	);
429	let mut items = page.items;
430	for (index, item) in &mut items {
431	if let TimelineItem::Event(event) = item {
432	event.highlight = highlights[*index];
433	}
434	}
435	let mut media = page.media;
436	let mut targets: HashMap<OwnedEventId, TimelineEvent> = HashMap::new();
437	let mut missing = Vec::new();
438	for (_, item) in &items {
439	if let TimelineItem::Event(event) = item
440	&& let Some(event_id) = &event.event_id
441	{
442	targets.insert(event_id.clone(), (**event).clone());
443	}
444	}
445	for event in events(&mut items) {
446	if let Some(reply) = &event.reply_to
447	&& !targets.contains_key(&reply.event_id)
448	&& !missing.contains(&reply.event_id)
449	{
450	missing.push(reply.event_id.clone());
451	}
452	}
453	let fetched = join_all(
454	missing
455	.into_iter()
456	.map(\|event_id\| fetch_reply(room, current_user_id, event_id, media_prefix)),
457	)
458	.await;
459	for (event, registrations) in fetched.into_iter().flatten() {
460	media.extend(registrations);
461	if let Some(event_id) = event.event_id.clone() {
462	targets.insert(event_id, event);
463	}
464	}
465	let profiles = profiles(
466	room,
467	items
468	.iter()
469	.filter_map(\|(_, item)\| match item {
470	TimelineItem::Event(event) => Some(&event.sender.user_id),
471	_ => None,
472	})
473	.chain(targets.values().map(\|event\| &event.sender.user_id)),
474	)
475	.await;
476	apply_profiles(targets.values_mut(), &profiles);
477	apply_profiles(events(&mut items), &profiles);
478	resolve_replies(events(&mut items), &targets);
479	Self { items, media }
480	}
481
482	/// the items that came from raw events in `range`
483	fn take(&mut self, range: std::ops::Range<usize>) -> Vec<TimelineItem> {
484	let (taken, kept) = std::mem::take(&mut self.items)
485	.into_iter()
486	.partition(\|(index, _)\| range.contains(index));
487	self.items = kept;
488	taken.into_iter().map(\|(_, item)\| item).collect()
489	}
490	}
491
492	fn events(items: &mut [(usize, TimelineItem)]) -> impl Iterator<Item = &mut TimelineEvent> {
493	items.iter_mut().filter_map(\|(_, item)\| match item {
494	TimelineItem::Event(event) => Some(&mut **event),
495	_ => None,
496	})
497	}
498
499	async fn fetch_reply(
500	room: &Room,
501	current_user_id: Option<&ruma::UserId>,
502	event_id: OwnedEventId,
503	media_prefix: &str,
504	) -> Option<(TimelineEvent, Vec<MediaRegistration>)> {
505	let event = room
506	.load_or_fetch_event(&event_id, None)
507	.await
508	.inspect_err(\|error\| tracing::debug!(%event_id, %error, "fetching a reply target failed"))
509	.ok()?;
510	let value: Value = serde_json::from_str(event.raw().json().get()).ok()?;
511	let identity = format!("{media_prefix}{}\n{event_id}", room.room_id());
512	match normalise_observed(&value, current_user_id, &identity)? {
513	(TimelineItem::Event(event), media) => Some((*event, media)),
514	_ => None,
515	}
516	}
517
518	/// under lazy-loaded members a sender may be unknown locally, and `get_member` fetches the
519	/// member list once for the room when it is
520	async fn profiles<'a>(
521	room: &Room,
522	user_ids: impl Iterator<Item = &'a OwnedUserId>,
523	) -> HashMap<OwnedUserId, Sender> {
524	let mut unique: Vec<&OwnedUserId> = user_ids.collect();
525	unique.sort();
526	unique.dedup();
527	join_all(unique.into_iter().map(\|user_id\| async move {
528	Some((user_id.clone(), member_sender(room, user_id).await?))
529	}))
530	.await
531	.into_iter()
532	.flatten()
533	.collect()
534	}
535
536	fn validate_limit(limit: u16) -> Result<(), Error> {
537	if limit > MAX_CONTEXT_LIMIT {
538	Err(Error::new(ErrorKind::Invalid, "context limit exceeds 50"))
539	} else {
540	Ok(())
541	}
542	}