macltok

public

Star0

Fork 0

Watch0

53 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `89bc9f8d246d`

tarball

atagen <boss@atagen.co>2026-10-24 13:40 UTC

6 files changed+357-2

timeline: add independent event context pagination

Pinned-message views need bounded history without moving a room timeline cursor. The context worker now owns focused requests, opaque directional pagination, normalisation, and distinct unavailable-event errors.

Added file: src/engine/context.rs+224 additions@@ -0,0 +1,224 @@
+use std::sync::Arc;
+
+use matrix_sdk::{room::MessagesOptions, Room};
+use matrix_sdk_ui::timeline::{RoomExt, TimelineEventFocusThreadMode, TimelineFocus};
+use ruma::{uint, OwnedEventId};
+
+use crate::{
+    normalise::{normalise_timeline, MediaScope},
+    Error, ErrorKind, EventContext, EventContextCommand, EventContextDirection, EventContextPage,
+    TimelineItem,
+};
+
+const MAX_CONTEXT_LIMIT: u16 = 50;
+
+pub(crate) async fn execute(
+    room: Room,
+    current_user_id: Option<ruma::OwnedUserId>,
+    command: EventContextCommand,
+) -> Result<crate::CommandOutput, Error> {
+    if let EventContextCommand::Before { token, limit, .. }
+    | EventContextCommand::After { token, limit, .. } = &command
+    {
+        validate_limit(*limit)?;
+        if token.is_empty() {
+            return Err(Error::new(
+                ErrorKind::Invalid,
+                "context pagination token is empty",
+            ));
+        }
+        return execute_page(
+            room,
+            current_user_id,
+            token,
+            *limit,
+            matches!(command, EventContextCommand::Before { .. }),
+        )
+        .await;
+    }
+
+    let (event_id, limit) = match &command {
+        EventContextCommand::Fetch {
+            event_id,
+            before_limit,
+            after_limit,
+            ..
+        } => {
+            validate_limit(*before_limit)?;
+            validate_limit(*after_limit)?;
+            (event_id.clone(), (*before_limit).max(*after_limit))
+        }
+        EventContextCommand::Before { .. } | EventContextCommand::After { .. } => unreachable!(),
+    };
+    validate_limit(limit)?;
+
+    let raw_context = if matches!(command, EventContextCommand::Fetch { .. }) {
+        Some(
+            room.event_with_context(&event_id, false, uint!(0), 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
+                    }
+                })?,
+        )
+    } else {
+        None
+    };
+
+    let focus = TimelineFocus::Event {
+        target: event_id.clone(),
+        num_context_events: limit,
+        thread_mode: TimelineEventFocusThreadMode::Automatic {
+            hide_threaded_events: true,
+        },
+    };
+    let timeline = room
+        .timeline_builder()
+        .with_focus(focus)
+        .build()
+        .await
+        .map_err(|source| {
+            Error::with_source(ErrorKind::Timeline, "building event context failed", source)
+        })?;
+
+    if let EventContextCommand::Fetch { .. } = command {
+        let raw = raw_context.expect("fetch context response");
+        let normalized = normalize_items(timeline.items().await, current_user_id.as_deref());
+        let (before, event, after) = split_items(normalized, &event_id);
+        if event.is_none() && raw.event.is_none() {
+            return Err(Error::new(ErrorKind::EventNotFound, "event was not found"));
+        }
+        return Ok(crate::CommandOutput::EventContext(EventContext {
+            events_before: before,
+            event,
+            events_after: after,
+            start_token: raw.prev_batch_token,
+            end_token: raw.next_batch_token,
+        }));
+    }
+
+    unreachable!("fetch context returned above")
+}
+
+async fn execute_page(
+    room: Room,
+    current_user_id: Option<ruma::OwnedUserId>,
+    token: &str,
+    limit: u16,
+    backwards: bool,
+) -> Result<crate::CommandOutput, Error> {
+    let mut options = if backwards {
+        MessagesOptions::backward()
+    } else {
+        MessagesOptions::forward()
+    };
+    options = options.from(token);
+    options.limit = ruma::UInt::from(limit);
+    let messages = room
+        .messages(options)
+        .await
+        .map_err(|source| Error::from_sdk("fetching event context page failed", source))?;
+
+    let ids: Vec<_> = messages
+        .chunk
+        .iter()
+        .filter_map(|event| event.event_id())
+        .collect();
+    let normalized = if let Some(target) = ids.first() {
+        let focus = TimelineFocus::Event {
+            target: target.clone(),
+            num_context_events: limit,
+            thread_mode: TimelineEventFocusThreadMode::Automatic {
+                hide_threaded_events: false,
+            },
+        };
+        let timeline = room
+            .timeline_builder()
+            .with_focus(focus)
+            .build()
+            .await
+            .map_err(|source| {
+                Error::with_source(
+                    ErrorKind::Timeline,
+                    "building event context page failed",
+                    source,
+                )
+            })?;
+        normalize_items(timeline.items().await, current_user_id.as_deref())
+    } else {
+        Vec::new()
+    };
+
+    let mut by_id = std::collections::HashMap::new();
+    for item in normalized {
+        if let TimelineItem::Event(event) = &item {
+            if let Some(event_id) = &event.event_id {
+                by_id.insert(event_id.clone(), item);
+            }
+        }
+    }
+    let mut events: Vec<_> = ids.iter().filter_map(|id| by_id.remove(id)).collect();
+    if backwards {
+        events.reverse();
+    }
+    let next_token = messages.end.clone();
+    let end_reached = next_token.is_none();
+    let direction = if backwards {
+        EventContextDirection::Before
+    } else {
+        EventContextDirection::After
+    };
+    Ok(crate::CommandOutput::EventContextPage(EventContextPage {
+        direction,
+        events,
+        next_token,
+        end_reached,
+    }))
+}
+
+fn validate_limit(limit: u16) -> Result<(), Error> {
+    if limit > MAX_CONTEXT_LIMIT {
+        Err(Error::new(ErrorKind::Invalid, "context limit exceeds 50"))
+    } else {
+        Ok(())
+    }
+}
+
+fn normalize_items(
+    items: imbl::Vector<Arc<matrix_sdk_ui::timeline::TimelineItem>>,
+    current_user_id: Option<&ruma::UserId>,
+) -> Vec<TimelineItem> {
+    let scope = MediaScope::new();
+    normalise_timeline(items.iter(), current_user_id, &scope)
+        .items
+        .into_iter()
+        .collect()
+}
+
+fn split_items(
+    items: Vec<TimelineItem>,
+    event_id: &OwnedEventId,
+) -> (Vec<TimelineItem>, Option<TimelineItem>, Vec<TimelineItem>) {
+    let items: Vec<_> = items
+        .into_iter()
+        .filter(|item| matches!(item, TimelineItem::Event(_)))
+        .collect();
+    let target = items.iter().position(|item| {
+        matches!(item, TimelineItem::Event(event) if event.event_id.as_ref() == Some(event_id))
+    });
+    let Some(index) = target else {
+        return (items, None, Vec::new());
+    };
+    let mut before = items;
+    let after = before.split_off(index + 1);
+    let event = before.pop();
+    (before, event, after)
+}

Modified file: src/engine/mod.rs+16 additions@@ -1,4 +1,5 @@
 mod commands;
+mod context;
 pub(crate) mod executor;
 mod media;
 mod mutations;
@@ -742,6 +743,21 @@ impl Engine {
                 },
             ),
             Command::Timeline(command) => self.route_timeline(command, reply),
+            Command::EventContext(command) => {
+                let room_id = match &command {
+                    crate::EventContextCommand::Fetch { room_id, .. }
+                    | crate::EventContextCommand::Before { room_id, .. }
+                    | crate::EventContextCommand::After { room_id, .. } => room_id,
+                };
+                let Some(room) = client.get_room(room_id) else {
+                    reply.done(Err(Error::room_unknown()));
+                    return None;
+                };
+                let current_user_id = client.user_id().map(ToOwned::to_owned);
+                self.tasks.spawn(TaskContext::Command("event context"), async move {
+                    reply.send(context::execute(room, current_user_id, command).await);
+                });
+            }
             Command::Media(MediaCommand::Upload(upload)) => self.upload(upload, reply),
             Command::Encryption(command) => enqueue(&self.encryption, Job { command, reply }),
             Command::Notification(command) => {

Modified file: src/error.rs+1

@@ -53,6 +53,7 @@ pub enum ErrorKind {

5353 RoomUnknown,

5454 RoomNotOpen,

5555 TimelineNotOpen,

56+ EventNotFound,

5657 RoomBusy,

5758 QueueFull,

5859 EmptyBody,

Modified file: src/lib.rs+2 additions−1 deletions@@ -62,7 +62,8 @@ pub use model::{
     Sender, SpaceChild, Termination, ThreadSummary, TimelineCommand, TimelineContent, TimelineDiff,
     TimelineDiffError, TimelineEvent, TimelineId, TimelineItem, TimelineItemId, TypingState,
     UploadDetails, UploadOutcome, UploadThumbnail, UtdCause, VerificationAction,
-    VerificationFailure, VerificationState, VerificationTarget,
+    VerificationFailure, VerificationState, VerificationTarget, EventContext, EventContextCommand,
+    EventContextDirection, EventContextPage,
 };
 pub use ruma;
 pub use transport::{ClientHandle, EventStream, Pending, SpaceHierarchy};

Modified file: src/model.rs+53 additions@@ -23,6 +23,7 @@ pub enum Command {
     Account(AccountCommand),
     Room(RoomCommand),
     Timeline(TimelineCommand),
+    EventContext(EventContextCommand),
     Media(MediaCommand),
     Encryption(EncryptionCommand),
     Notification(NotificationCommand),
@@ -30,6 +31,28 @@ pub enum Command {
     Pack(PackCommand),
 }
 
+#[derive(Clone, Debug, Eq, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[non_exhaustive]
+pub enum EventContextCommand {
+    Fetch {
+        room_id: OwnedRoomId,
+        event_id: OwnedEventId,
+        before_limit: u16,
+        after_limit: u16,
+    },
+    Before {
+        room_id: OwnedRoomId,
+        limit: u16,
+        token: String,
+    },
+    After {
+        room_id: OwnedRoomId,
+        limit: u16,
+        token: String,
+    },
+}
+
 #[derive(Debug)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 #[non_exhaustive]
@@ -640,6 +663,36 @@ pub enum CommandOutput {
         shortcodes: Vec<String>,
         skipped: Vec<String>,
     },
+    EventContext(EventContext),
+    EventContextPage(EventContextPage),
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[non_exhaustive]
+pub struct EventContext {
+    pub events_before: Vec<TimelineItem>,
+    pub event: Option<TimelineItem>,
+    pub events_after: Vec<TimelineItem>,
+    pub start_token: Option<String>,
+    pub end_token: Option<String>,
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+pub enum EventContextDirection {
+    Before,
+    After,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[non_exhaustive]
+pub struct EventContextPage {
+    pub direction: EventContextDirection,
+    pub events: Vec<TimelineItem>,
+    pub next_token: Option<String>,
+    pub end_reached: bool,
 }
 
 #[derive(Clone, Debug, Eq, PartialEq)]

Modified file: src/transport.rs+61 additions−1 deletions@@ -28,7 +28,8 @@ use tokio::sync::{OnceCell, mpsc, oneshot, watch};
 
 use crate::{
     AccountCommand, AttachmentUpload, Command, CommandOutput, ConnectionState, CreateRoomOptions,
-    EncryptionCommand, EncryptionState, Error, ErrorKind, Event, EventRef, ImagePackSource,
+    EncryptionCommand, EncryptionState, Error, ErrorKind, Event, EventContext, EventContextCommand,
+    EventContextPage, EventRef, ImagePackSource,
     ImagePacks, InteractiveAuth, InteractiveAuthOutcome, LifecycleCommand, LinkPreview, Media,
     MediaCommand, MediaId, MediaLocation, MediaSize, MemberFilter, MemberSummary, MessageContent,
     MessageEditContent, NotificationCommand, OwnDevice, PackCommand, PackImageInfo, PackUsage,
@@ -325,6 +326,65 @@ impl ClientHandle {
         .await
     }
 
+    pub async fn event_context(
+        &self,
+        room_id: &RoomId,
+        event_id: &EventId,
+        before_limit: u32,
+        after_limit: u32,
+    ) -> Result<EventContext, Error> {
+        match self
+            .execute(Command::EventContext(EventContextCommand::Fetch {
+                room_id: room_id.to_owned(),
+                event_id: event_id.to_owned(),
+                before_limit: before_limit.min(50) as u16,
+                after_limit: after_limit.min(50) as u16,
+            }))
+            .await?
+        {
+            CommandOutput::EventContext(context) => Ok(context),
+            output => Err(unexpected("event context", &output)),
+        }
+    }
+
+    pub async fn event_context_before(
+        &self,
+        room_id: &RoomId,
+        token: &str,
+        limit: u32,
+    ) -> Result<EventContextPage, Error> {
+        self.event_context_page(EventContextCommand::Before {
+            room_id: room_id.to_owned(),
+            limit: limit.min(50) as u16,
+            token: token.to_owned(),
+        })
+        .await
+    }
+
+    pub async fn event_context_after(
+        &self,
+        room_id: &RoomId,
+        token: &str,
+        limit: u32,
+    ) -> Result<EventContextPage, Error> {
+        self.event_context_page(EventContextCommand::After {
+            room_id: room_id.to_owned(),
+            limit: limit.min(50) as u16,
+            token: token.to_owned(),
+        })
+        .await
+    }
+
+    async fn event_context_page(
+        &self,
+        command: EventContextCommand,
+    ) -> Result<EventContextPage, Error> {
+        match self.execute(Command::EventContext(command)).await? {
+            CommandOutput::EventContextPage(page) => Ok(page),
+            output => Err(unexpected("event context page", &output)),
+        }
+    }
+
     pub async fn send_message(
         &self,
         timeline_id: TimelineId,

macltok

Commit 89bc9f8d246d

Commit `89bc9f8d246d`