macltok

public
Star0
Watch0
129 commits 3 branches 0 tags 0 releases3.0 MiB
use std::{
    collections::{BTreeMap, BTreeSet, HashMap},
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
    time::Duration,
};

use futures_util::{StreamExt, stream::FuturesUnordered};
use matrix_sdk::{
    Client, Room, RoomState,
    event_cache::RedecryptorReport,
    sliding_sync::{ResponseCommitBatch, ResponseCommitHook},
};
use ruma::{
    OwnedRoomId, OwnedUserId, RoomId,
    events::{
        AnySyncTimelineEvent, ignored_user_list::IgnoredUserListEventContent,
        room::encrypted::OriginalSyncRoomEncryptedEvent,
    },
    serde::Raw,
};
use serde_json::Value;
use tokio::sync::{mpsc, oneshot, watch};

use super::state::{Candidate, Checkpoint, Data, SavedEvent, SyncState, Window, id, invalid, now};
#[cfg(any(feature = "sqlite", feature = "indexeddb"))]
use super::store::Opener;
use crate::{
    CommandOutput, EngineConfig, Error, ErrorKind, Event, MentionCaptureState, MentionEntry,
    MentionGapReason, MentionKey, MentionReason, MentionsCommand, MentionsPage, MentionsUpdate,
    TimelineContent, TimelineItem,
    engine::{MediaService, MentionsStore, context, executor::EngineBoxFuture},
    normalise::{MediaRegistration, MediaScope, normalise_observed, resolve_replies},
    transport::EventSender,
};

#[cfg(all(test, feature = "sqlite"))]
#[path = "service_tests.rs"]
mod tests;

const HOOK_TIMEOUT: Duration = Duration::from_secs(10);
const FLUSH_RETRY_MS: u64 = 60_000;

#[cfg_attr(not(any(feature = "sqlite", feature = "indexeddb")), allow(dead_code))]
pub(crate) struct Unopened {
    pub(crate) error: Error,
    /// set when another attempt may succeed
    #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
    pub(crate) retry: Option<Opener>,
}

struct Loaded {
    store: Option<MentionsStore>,
    #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
    reopen: Option<Opener>,
    data: Data,
    sync: SyncState,
    rooms: BTreeSet<OwnedRoomId>,
    gaps: Vec<MentionGapReason>,
}

impl Loaded {
    fn empty(gaps: Vec<MentionGapReason>) -> Self {
        Self {
            store: None,
            #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
            reopen: None,
            data: Data::default(),
            sync: SyncState::default(),
            rooms: BTreeSet::new(),
            gaps,
        }
    }
}

#[derive(Clone)]
pub(crate) struct Handle {
    requests: mpsc::Sender<Request>,
    timed_out: Arc<AtomicBool>,
}

enum Request {
    Batch(ResponseCommitBatch, oneshot::Sender<()>),
    Command(
        MentionsCommand,
        oneshot::Sender<Result<CommandOutput, Error>>,
    ),
    Stop(oneshot::Sender<()>),
}

pub(crate) struct Service {
    client: Client,
    account: OwnedUserId,
    media: Arc<MediaService>,
    /// as many media as the inbox keeps events
    media_scope: MediaScope,
    store: Option<MentionsStore>,
    #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
    reopen: Option<Opener>,
    data: Data,
    sync: SyncState,
    rooms: BTreeSet<OwnedRoomId>,
    dirty: bool,
    rooms_dirty: bool,
    unpublished: bool,
    flush_after: u64,
    ignored: BTreeSet<OwnedUserId>,
    /// kept out of everything the inbox shows, while their entries stay saved for when the
    /// room stops being direct
    direct: BTreeSet<OwnedRoomId>,
    leaving: BTreeSet<OwnedRoomId>,
    timed_out: Arc<AtomicBool>,
    config: EngineConfig,
    requests: mpsc::Receiver<Request>,
    updates: watch::Sender<MentionsPage>,
    recovery_budget: usize,
    latest_scan: bool,
    recovery_retry: u64,
    /// senders a profile pass already looked up, found or not, so one without a profile isn't
    /// asked for again every second
    profiled: BTreeSet<(OwnedRoomId, OwnedUserId)>,
    profiling: bool,
}

enum Work {
    Capture(
        String,
        u64,
        Result<(crate::EventContext, Vec<MediaRegistration>, Vec<Value>), Error>,
    ),
    Candidate(String, Result<Value, Error>),
    History(
        bool,
        Option<String>,
        Result<ruma::api::client::push::get_notifications::v3::Response, Error>,
    ),
    Profiles(
        OwnedRoomId,
        Vec<OwnedUserId>,
        HashMap<OwnedUserId, crate::Sender>,
    ),
}

impl Work {
    fn rooms(&self) -> BTreeSet<OwnedRoomId> {
        let room = |key: &str| {
            key.split_once('\n')
                .and_then(|(room, _)| OwnedRoomId::try_from(room).ok())
        };
        match self {
            Self::Capture(key, ..) | Self::Candidate(key, _) => room(key).into_iter().collect(),
            Self::History(_, _, Ok(page)) => page
                .notifications
                .iter()
                .map(|notification| notification.room_id.clone())
                .collect(),
            Self::History(..) => BTreeSet::new(),
            Self::Profiles(room_id, ..) => BTreeSet::from([room_id.clone()]),
        }
    }
}

impl Handle {
    /// never fails, a mentions failure becomes a coverage gap rather than a stalled sync
    pub(crate) fn hook(&self) -> ResponseCommitHook {
        let handle = self.clone();
        ResponseCommitHook::new(move |batch| {
            let handle = handle.clone();
            async move {
                let (reply, response) = oneshot::channel();
                let queued = handle.requests.send(Request::Batch(batch, reply));
                match matrix_sdk::timeout::timeout(queued, HOOK_TIMEOUT).await {
                    Ok(Ok(())) => {}
                    Ok(Err(_)) => {
                        tracing::warn!("mentions service stopped, sync carries on");
                        return Ok(());
                    }
                    Err(_) => {
                        handle.timed_out.store(true, Ordering::Release);
                        tracing::warn!("mentions queue stayed full, a response was dropped");
                        return Ok(());
                    }
                }
                // a queued batch still commits in order, so giving up here loses nothing
                // unless the process dies first, which the next start sees as a
                // discontinuity
                if matrix_sdk::timeout::timeout(response, HOOK_TIMEOUT)
                    .await
                    .is_err()
                {
                    tracing::warn!("mentions commit is slow, sync carries on");
                }
                Ok(())
            }
        })
    }

    pub(crate) async fn execute(&self, command: MentionsCommand) -> Result<CommandOutput, Error> {
        let (reply, response) = oneshot::channel();
        self.requests
            .send(Request::Command(command, reply))
            .await
            .map_err(|_| Error::interrupted())?;
        response.await.map_err(|_| Error::interrupted())?
    }

    pub(crate) async fn stop(&self) {
        let (reply, response) = oneshot::channel();
        if self.requests.send(Request::Stop(reply)).await.is_ok() {
            let _ = response.await;
        }
    }
}

#[cfg(any(feature = "sqlite", feature = "indexeddb"))]
fn read_store(store: &MentionsStore) -> Result<(Data, SyncState, BTreeSet<OwnedRoomId>), Error> {
    let data: Data = store.read("inbox")?.unwrap_or_default();
    let sync: SyncState = store.read("sync")?.unwrap_or_default();
    let rooms = store.read("rooms")?.unwrap_or_default();
    if sync.inbox_revision != data.revision {
        return Err(Error::new(
            ErrorKind::Store,
            "mentions checkpoint does not match its inbox",
        ));
    }
    Ok((data, sync, rooms))
}

#[cfg(any(feature = "sqlite", feature = "indexeddb"))]
async fn open_store(store: Option<Result<MentionsStore, Unopened>>) -> Loaded {
    let mut store = match store {
        None => return Loaded::empty(Vec::new()),
        Some(Err(unopened)) => {
            tracing::error!(error = %unopened.error, "mentions store unavailable, the inbox runs in memory");
            let mut loaded = Loaded::empty(vec![MentionGapReason::StoreFailed]);
            loaded.reopen = unopened.retry;
            return loaded;
        }
        Some(Ok(store)) => store,
    };
    let (data, sync, rooms) = match read_store(&store) {
        Ok(loaded) => loaded,
        Err(error) => {
            tracing::warn!(%error, "resetting an unreadable mentions inbox");
            if let Err(error) = store.reset().await {
                tracing::error!(%error, "mentions store unavailable, the inbox runs in memory");
                return Loaded::empty(vec![MentionGapReason::StoreFailed]);
            }
            Default::default()
        }
    };
    let gaps = if store.was_reset() {
        vec![MentionGapReason::StoreReset]
    } else {
        Vec::new()
    };
    Loaded {
        store: Some(store),
        reopen: None,
        data,
        sync,
        rooms,
        gaps,
    }
}

#[cfg(not(any(feature = "sqlite", feature = "indexeddb")))]
async fn open_store(store: Option<Result<MentionsStore, Unopened>>) -> Loaded {
    match store {
        None => Loaded::empty(Vec::new()),
        Some(Err(_)) => Loaded::empty(vec![MentionGapReason::StoreFailed]),
        Some(Ok(store)) => match store {},
    }
}

impl Service {
    pub(crate) async fn new(
        client: Client,
        media: Arc<MediaService>,
        store: Option<Result<MentionsStore, Unopened>>,
        config: EngineConfig,
    ) -> Result<(Handle, Self, watch::Receiver<MentionsPage>), Error> {
        let account = client
            .user_id()
            .ok_or_else(|| {
                Error::new(
                    ErrorKind::Unauthenticated,
                    "mentions require an authenticated account",
                )
            })?
            .to_owned();
        let mut loaded = open_store(store).await;
        for reason in &loaded.gaps {
            loaded.data.gap(None, *reason, None);
        }
        let direct = direct_rooms(&client);
        let page = loaded.data.page(&account, None, 50, &direct)?;
        let media_scope = MediaScope::bounded(config.max_mention_context_events);
        media
            .register(
                &media_scope,
                page.entries
                    .iter()
                    .filter_map(|entry| loaded.data.events.get(&id(&entry.key)))
                    .flat_map(|event| event.media.clone())
                    .collect(),
            )
            .await;
        let (updates, receiver) = watch::channel(page);
        let (sender, requests) = mpsc::channel(config.command_capacity.max(1));
        let timed_out = Arc::new(AtomicBool::new(false));
        let mut service = Self {
            client,
            account,
            media,
            media_scope,
            store: loaded.store,
            #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
            reopen: loaded.reopen,
            data: loaded.data,
            sync: loaded.sync,
            rooms: loaded.rooms,
            dirty: !loaded.gaps.is_empty(),
            rooms_dirty: false,
            unpublished: false,
            flush_after: 0,
            ignored: BTreeSet::new(),
            direct,
            leaving: BTreeSet::new(),
            timed_out: timed_out.clone(),
            config,
            requests,
            updates,
            recovery_budget: 20,
            latest_scan: true,
            recovery_retry: 0,
            profiled: BTreeSet::new(),
            profiling: false,
        };
        service.ignored = service.ignored_users().await;
        Ok((
            Handle {
                requests: sender,
                timed_out,
            },
            service,
            receiver,
        ))
    }

    async fn ignored_users(&self) -> BTreeSet<OwnedUserId> {
        self.client
            .account()
            .account_data::<IgnoredUserListEventContent>()
            .await
            .ok()
            .flatten()
            .and_then(|raw| raw.deserialize().ok())
            .map(|content| content.ignored_users.into_keys().collect())
            .unwrap_or_default()
    }

    async fn publish(&mut self) {
        self.unpublished = false;
        let Ok(page) = self.data.page(&self.account, None, 50, &self.direct) else {
            return;
        };
        self.media
            .register(
                &self.media_scope,
                page.entries
                    .iter()
                    .filter_map(|entry| self.data.events.get(&id(&entry.key)))
                    .flat_map(|event| event.media.clone())
                    .collect(),
            )
            .await;
        self.updates.send_replace(page);
    }

    /// writes the sync record, with the inbox when `dirty` and observed rooms when
    /// `rooms_dirty`, in one commit so a checkpoint never lands without the changes it
    /// covers
    async fn flush(&mut self) -> Result<(), Error> {
        if self.dirty {
            self.advance()?;
        }
        #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
        if let Some(store) = &mut self.store {
            use super::store::Write;

            let encode = |error| {
                Error::with_source(
                    ErrorKind::Store,
                    "encoding the mentions inbox failed",
                    error,
                )
            };
            self.sync.inbox_revision = self.data.revision;
            let sync = serde_json::value::to_raw_value(&self.sync).map_err(encode)?;
            let rooms = if self.rooms_dirty {
                Some(serde_json::value::to_raw_value(&self.rooms).map_err(encode)?)
            } else {
                None
            };
            let mut writes = Vec::new();
            if self.dirty {
                self.data
                    .coverage
                    .gaps
                    .retain(|gap| gap.reason != MentionGapReason::StoreFailed);
                let replaced: &[&str] = if rooms.is_some() {
                    &["inbox", "sync", "rooms"]
                } else {
                    &["inbox", "sync"]
                };
                let available = store
                    .available(replaced)
                    .saturating_sub(store.sealed_bound("sync", sync.get().len()))
                    .saturating_sub(
                        rooms
                            .as_ref()
                            .map_or(0, |rooms| store.sealed_bound("rooms", rooms.get().len())),
                    );
                let mut inbox = serde_json::value::to_raw_value(&self.data).map_err(encode)?;
                let mut size = store.sealed_bound("inbox", inbox.get().len());
                if size > available {
                    while size > available {
                        let Some(freed) = self.data.evict_one() else {
                            break;
                        };
                        size = size.saturating_sub(freed * 3);
                    }
                    inbox = serde_json::value::to_raw_value(&self.data).map_err(encode)?;
                }
                writes.push(Write::Set("inbox".to_owned(), inbox));
            }
            writes.push(Write::Set("sync".to_owned(), sync));
            if let Some(rooms) = rooms {
                writes.push(Write::Set("rooms".to_owned(), rooms));
            }
            store.commit(store.revision(), writes).await?;
        }
        self.rooms_dirty = false;
        if self.dirty {
            self.dirty = false;
            self.publish().await;
        }
        Ok(())
    }

    fn advance(&mut self) -> Result<(), Error> {
        self.data.revision = self
            .data
            .revision
            .checked_add(1)
            .ok_or_else(|| invalid("mentions revision exhausted"))?;
        self.data.bound(
            self.config.max_mentions_entries,
            self.config.max_mention_context_events,
            self.config.max_mention_candidates,
        );
        Ok(())
    }

    async fn apply(&mut self, data: Data) -> Result<(), Error> {
        let dirty = std::mem::replace(&mut self.dirty, true);
        let previous = std::mem::replace(&mut self.data, data);
        let result = self.flush().await;
        if result.is_err() {
            self.data = previous;
            self.dirty = dirty;
        }
        result
    }

    async fn store_failed(&mut self, error: Error) {
        tracing::error!(%error, "saving the mentions inbox failed");
        self.data.gap(None, MentionGapReason::StoreFailed, None);
        self.dirty = true;
        self.flush_after = now().saturating_add(FLUSH_RETRY_MS);
        self.publish().await;
    }

    /// the store is back after failing at startup, what it held and what was built in
    /// memory since are merged so nothing is lost
    #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
    async fn reopen(&mut self) {
        let Some(opener) = &self.reopen else {
            return;
        };
        let store = match opener.open().await {
            Ok(store) => store,
            Err(super::store::OpenError::Unavailable(error)) => {
                tracing::warn!(%error, "the mentions store is still unavailable");
                self.flush_after = now().saturating_add(FLUSH_RETRY_MS);
                return;
            }
            Err(super::store::OpenError::Newer(error)) => {
                tracing::error!(%error, "the mentions store belongs to a newer macltok");
                self.reopen = None;
                return;
            }
        };
        let mut store = store;
        let (mut data, mut sync, mut rooms) = match read_store(&store) {
            Ok(loaded) => loaded,
            Err(error) => {
                tracing::warn!(%error, "resetting an unreadable mentions inbox");
                if let Err(error) = store.reset().await {
                    tracing::warn!(%error, "the mentions store is still unavailable");
                    self.flush_after = now().saturating_add(FLUSH_RETRY_MS);
                    return;
                }
                self.data.gap(None, MentionGapReason::StoreReset, None);
                Default::default()
            }
        };
        data.absorb(std::mem::take(&mut self.data));
        sync.checkpoints
            .extend(std::mem::take(&mut self.sync.checkpoints));
        rooms.extend(std::mem::take(&mut self.rooms));
        self.data = data;
        self.sync = sync;
        self.rooms = rooms;
        self.store = Some(store);
        self.reopen = None;
        self.dirty = true;
        self.rooms_dirty = true;
        self.flush_after = 0;
    }

    async fn command(&mut self, command: MentionsCommand) -> Result<CommandOutput, Error> {
        match command {
            MentionsCommand::Page { cursor, limit } => {
                let page = self
                    .data
                    .page(&self.account, cursor.as_deref(), limit, &self.direct)?;
                self.media
                    .register(
                        &self.media_scope,
                        page.entries
                            .iter()
                            .filter_map(|entry| self.data.events.get(&id(&entry.key)))
                            .flat_map(|event| event.media.clone())
                            .collect(),
                    )
                    .await;
                Ok(CommandOutput::MentionsPage(page))
            }
            MentionsCommand::Context(key) => {
                let (context, media) = self.data.context(&key)?;
                self.media.register(&self.media_scope, media).await;
                Ok(CommandOutput::MentionContext(context))
            }
            MentionsCommand::RetryContext(key) => {
                let room = self
                    .client
                    .get_room(&key.room_id)
                    .filter(|room| room.state() == RoomState::Joined)
                    .ok_or_else(|| {
                        Error::new(
                            ErrorKind::Forbidden,
                            "saved context cannot be refetched after leaving the room",
                        )
                    })?;
                drop(room);
                let mut next = self.data.clone();
                let entry = next
                    .entries
                    .get_mut(&id(&key))
                    .ok_or_else(|| Error::new(ErrorKind::EventNotFound, "mention was not found"))?;
                entry.capture = MentionCaptureState::Pending;
                entry.revision = self.data.revision + 1;
                next.windows.entry(id(&key)).or_insert(Window {
                    before: Vec::new(),
                    target: Some(id(&key)),
                    after: Vec::new(),
                    start: None,
                    end: None,
                });
                self.apply(next).await?;
                let (context, media) = self.data.context(&key)?;
                self.media.register(&self.media_scope, media).await;
                Ok(CommandOutput::MentionContext(context))
            }
            MentionsCommand::SetRead { keys, read } => {
                let mut next = self.data.clone();
                for key in keys
                    .iter()
                    .filter(|key| !self.direct.contains(&key.room_id))
                {
                    if let Some(entry) = next.entries.get_mut(&id(key)) {
                        entry.read = read;
                        entry.revision = self.data.revision + 1;
                    }
                }
                self.apply(next).await?;
                Ok(CommandOutput::Done)
            }
            MentionsCommand::ReadThrough(boundary) => {
                if boundary.account != self.account || boundary.revision > self.data.revision {
                    return Err(invalid("invalid mentions read boundary"));
                }
                let mut next = self.data.clone();
                for entry in next.entries.values_mut().filter(|entry| {
                    entry.discovery_revision <= boundary.revision
                        && !self.direct.contains(&entry.key.room_id)
                }) {
                    entry.read = true;
                    entry.revision = self.data.revision + 1;
                }
                self.apply(next).await?;
                Ok(CommandOutput::Done)
            }
            MentionsCommand::Recover => {
                self.recovery_budget = 20;
                self.latest_scan = true;
                self.recovery_retry = 0;
                let mut next = self.data.clone();
                next.coverage.recovering = true;
                next.coverage.historical_supported = None;
                self.apply(next).await?;
                Ok(CommandOutput::Done)
            }
        }
    }

    fn observable(&self, room_id: &RoomId) -> Option<Room> {
        self.client
            .get_room(room_id)
            .filter(|room| room.state() == RoomState::Joined || self.leaving.contains(room_id))
    }

    async fn batch(&mut self, batch: ResponseCommitBatch) {
        let checkpoint = self.sync.checkpoints.get(&batch.connection_id).cloned();
        if checkpoint
            .as_ref()
            .is_some_and(|checkpoint| checkpoint.pos == batch.next_pos)
        {
            return;
        }
        self.ignored = self.ignored_users().await;
        self.leaving = batch.response.rooms.left.keys().cloned().collect();
        let room_ids: BTreeSet<_> = batch
            .raw_rooms
            .keys()
            .chain(batch.response.rooms.joined.keys())
            .chain(batch.response.rooms.left.keys())
            .cloned()
            .collect();
        let before = self.data.scope(&room_ids);
        let mut next = std::mem::take(&mut self.data);
        let mut changed = false;
        if self.timed_out.swap(false, Ordering::AcqRel) {
            next.gap(None, MentionGapReason::HookTimedOut, None);
        }
        // a crash after the commit but before the sdk saved `pos` replays from the
        // previous position, and that response is a superset of the one committed
        let continuous = match &checkpoint {
            None => batch.previous_pos.is_none(),
            Some(checkpoint) => {
                batch.previous_pos.as_ref() == Some(&checkpoint.pos)
                    || batch.previous_pos == checkpoint.previous
            }
        };
        if !continuous {
            next.gap(None, MentionGapReason::SyncDiscontinuity, None);
        }
        for (room_id, raw) in &batch.raw_rooms {
            if raw.limited && self.rooms.contains(room_id) {
                next.gap(
                    Some(room_id.clone()),
                    MentionGapReason::LimitedTimeline,
                    raw.prev_batch.clone(),
                );
            }
        }
        for room_id in &room_ids {
            if self.observable(room_id).is_none() {
                continue;
            }
            self.rooms_dirty |= self.rooms.insert(room_id.clone());
            let mut observed = BTreeMap::new();
            if let Some(raw) = batch.raw_rooms.get(room_id) {
                for event in &raw.timeline {
                    let Ok(value) = serde_json::from_str::<Value>(event.json().get()) else {
                        continue;
                    };
                    if let Some(event_id) = value["event_id"].as_str() {
                        observed.insert(event_id.to_owned(), value);
                    }
                }
            }
            let processed = batch
                .response
                .rooms
                .joined
                .get(room_id)
                .map(|update| &update.timeline.events)
                .into_iter()
                .chain(
                    batch
                        .response
                        .rooms
                        .left
                        .get(room_id)
                        .map(|update| &update.timeline.events),
                )
                .flatten();
            for event in processed {
                let Ok(value) = serde_json::from_str::<Value>(event.raw().json().get()) else {
                    continue;
                };
                if let Some(event_id) = value["event_id"].as_str() {
                    observed.insert(event_id.to_owned(), value);
                }
            }
            let mut observed: Vec<_> = observed.into_values().collect();
            observed.sort_by_key(|value| value["origin_server_ts"].as_u64().unwrap_or(0));
            for value in observed {
                self.observe(&mut next, room_id, value).await;
            }
        }
        for room_id in batch.response.rooms.left.keys() {
            self.rooms_dirty |= self.rooms.remove(room_id);
            next.gap(Some(room_id.clone()), MentionGapReason::RoomLeft, None);
            next.candidates
                .retain(|_, candidate| &candidate.room_id != room_id);
            let revision = next.revision + 1;
            for entry in next.entries.values_mut().filter(|entry| {
                &entry.key.room_id == room_id && entry.capture != MentionCaptureState::Unavailable
            }) {
                entry.capture = MentionCaptureState::Unavailable;
                entry.revision = revision;
            }
        }
        let reevaluate: Vec<_> = room_ids
            .iter()
            .flat_map(|room_id| {
                next.entries
                    .range(format!("{room_id}\n")..format!("{room_id}\u{b}"))
                    .filter_map(|(key, _)| next.events.get(key).and_then(|event| event.raw.clone()))
                    .map(move |raw| (room_id.clone(), raw))
            })
            .collect();
        for (room_id, raw) in reevaluate {
            self.observe(&mut next, &room_id, raw).await;
        }
        changed |= self.filter_ignored(&mut next);
        self.leaving.clear();
        self.sync.checkpoints.insert(
            batch.connection_id,
            Checkpoint {
                pos: batch.next_pos,
                previous: batch.previous_pos,
            },
        );
        changed |= next.scope(&room_ids) != before;
        self.data = next;
        self.dirty |= changed;
        if now() < self.flush_after {
            if changed && self.advance().is_ok() {
                self.publish().await;
            }
        } else if let Err(error) = self.flush().await {
            self.store_failed(error).await;
        }
    }

    fn filter_ignored(&self, next: &mut Data) -> bool {
        let removed: Vec<_> = next
            .entries
            .iter()
            .filter(|(_, entry)| {
                matches!(&entry.event, TimelineItem::Event(event) if self.ignored.contains(&event.sender.user_id))
            })
            .map(|(key, _)| key.clone())
            .collect();
        if removed.is_empty() {
            return false;
        }
        for key in &removed {
            next.entries.remove(key);
            next.windows.remove(key);
        }
        next.collect_events();
        true
    }

    async fn observe(&self, data: &mut Data, room_id: &OwnedRoomId, mut value: Value) {
        let Some(event_id) = value["event_id"]
            .as_str()
            .and_then(|value| value.parse().ok())
        else {
            return;
        };
        let key = MentionKey {
            room_id: room_id.clone(),
            event_id,
        };
        let identity = id(&key);
        let Some(timestamp) = value["origin_server_ts"].as_u64() else {
            return;
        };
        data.coverage.earliest_timestamp_ms = Some(
            data.coverage
                .earliest_timestamp_ms
                .map_or(timestamp, |old| old.min(timestamp)),
        );
        if value["type"] == "m.room.redaction" {
            if let Some(target) = value["redacts"]
                .as_str()
                .or_else(|| value["content"]["redacts"].as_str())
                .and_then(|value| value.parse().ok())
            {
                let target = id(&MentionKey {
                    room_id: room_id.clone(),
                    event_id: target,
                });
                data.redacted.insert(target.clone(), timestamp);
                data.entries.remove(&target);
                data.windows.remove(&target);
                data.candidates.remove(&target);
                data.edits.remove(&target);
                data.candidates.retain(|_, candidate| {
                    candidate.room_id != *room_id
                        || candidate.raw["content"]["m.relates_to"]["event_id"].as_str()
                            != target.split_once('\n').map(|(_, event)| event)
                });
                if let Some(saved) = data.events.get_mut(&target) {
                    redact(saved);
                }
                data.touch_contexts(&target);
            }
            return;
        }
        if data.redacted.contains_key(&identity)
            || value["unsigned"].get("redacted_because").is_some()
        {
            data.redacted.insert(identity.clone(), timestamp);
            data.entries.remove(&identity);
            data.windows.remove(&identity);
            data.candidates.remove(&identity);
            if let Some(saved) = data.events.get_mut(&identity) {
                redact(saved);
            }
            data.touch_contexts(&identity);
            return;
        }
        if value["content"]["m.relates_to"]["rel_type"] == "m.replace" {
            if let Some(target) = value["content"]["m.relates_to"]["event_id"]
                .as_str()
                .and_then(|value| value.parse().ok())
            {
                let target = id(&MentionKey {
                    room_id: room_id.clone(),
                    event_id: target,
                });
                if data.redacted.contains_key(&target) {
                    return;
                }
                let newer = data.edits.get(&target).is_none_or(|old| {
                    (old["origin_server_ts"].as_u64(), old["event_id"].as_str())
                        < (Some(timestamp), value["event_id"].as_str())
                });
                if newer {
                    if let Some(original) =
                        data.events.get(&target).and_then(|saved| saved.raw.clone())
                    {
                        if original["sender"] != value["sender"] {
                            return;
                        }
                        data.edits.insert(target.clone(), value.clone());
                        value = edited(original, &value);
                        return Box::pin(self.observe(data, room_id, value)).await;
                    }
                    data.candidates
                        .entry(identity)
                        .and_modify(|candidate| candidate.raw = value.clone())
                        .or_insert_with(|| Candidate::new(room_id.clone(), value, 0));
                }
            }
            return;
        }
        if let Some(edit) = value["unsigned"]["m.relations"].get("m.replace").cloned() {
            if edit["sender"] == value["sender"]
                && edit["content"]["m.new_content"].is_object()
                && edit["content"]["m.relates_to"]["event_id"]
                    .as_str()
                    .is_none_or(|target| target == key.event_id.as_str())
                && edit["origin_server_ts"].as_u64().is_some()
                && data.edits.get(&identity).is_none_or(|old| {
                    (old["origin_server_ts"].as_u64(), old["event_id"].as_str())
                        < (edit["origin_server_ts"].as_u64(), edit["event_id"].as_str())
                })
            {
                data.edits.insert(identity.clone(), edit);
            }
        }
        if let Some(edit) = data.edits.get(&identity) {
            if value["sender"] == edit["sender"] {
                value = edited(value, edit);
            }
        }
        if let Some(edit) = value["unsigned"].get("macltok_edit").cloned() {
            data.edits.insert(identity.clone(), edit);
        }
        if value["type"] == "m.room.encrypted" {
            data.candidates
                .entry(identity)
                .or_insert_with(|| Candidate::new(room_id.clone(), value, 0));
            return;
        }
        let Some((mut item, media)) = normalise_observed(&value, Some(&self.account), &identity)
        else {
            return;
        };
        // the commit hook waits on this actor, so a reply settles here only from what the
        // inbox already holds, and the capture fetches profiles and anything else later
        if let TimelineItem::Event(event) = &mut item
            && let Some(reply) = &event.reply_to
        {
            let target = id(&MentionKey {
                room_id: room_id.clone(),
                event_id: reply.event_id.clone(),
            });
            let targets: HashMap<_, _> = match data.events.get(&target).map(|saved| &saved.item) {
                Some(TimelineItem::Event(saved)) => {
                    HashMap::from([(reply.event_id.clone(), (**saved).clone())])
                }
                _ => HashMap::new(),
            };
            resolve_replies(std::iter::once(&mut **event), &targets);
        }
        if let Some(saved) = data.events.get(&identity) {
            improve(&mut item, &saved.item);
        }
        self.supplement(data, &key, &item, &media, &value);
        let (reasons, suppressed, unresolved) = self.classify(data, room_id, &value).await;
        if unresolved {
            data.candidates
                .entry(identity.clone())
                .or_insert_with(|| Candidate::new(room_id.clone(), value.clone(), now() + 30_000));
        } else {
            data.candidates.remove(&identity);
        }
        if reasons.is_empty() {
            data.entries.remove(&identity);
            data.windows.remove(&identity);
            if data.events.contains_key(&identity) || data.edits.contains_key(&identity) {
                data.events.insert(
                    identity,
                    SavedEvent {
                        item,
                        media,
                        raw: Some(value),
                    },
                );
            }
            return;
        }
        let previous = data.entries.get(&identity);
        let event_changed = previous.is_none_or(|entry| entry.event != item);
        let changed = event_changed
            || previous.is_some_and(|entry| {
                entry.reasons != reasons || entry.notification_suppressed != suppressed
            });
        let entry = MentionEntry {
            key,
            timestamp_ms: timestamp,
            discovered_at_ms: previous.map_or_else(now, |entry| entry.discovered_at_ms),
            discovery_revision: previous
                .map_or(data.revision + 1, |entry| entry.discovery_revision),
            revision: if changed {
                data.revision + 1
            } else {
                previous.unwrap().revision
            },
            event: item.clone(),
            reasons,
            read: previous.is_some_and(|entry| entry.read),
            capture: previous.map_or(MentionCaptureState::Pending, |entry| {
                if event_changed {
                    MentionCaptureState::Pending
                } else {
                    entry.capture.clone()
                }
            }),
            room_name: self.client.get_room(room_id).and_then(|room| room.name()),
            notification_suppressed: suppressed,
        };
        data.entries.insert(identity.clone(), entry);
        data.windows.entry(identity.clone()).or_insert(Window {
            before: Vec::new(),
            target: Some(identity.clone()),
            after: Vec::new(),
            start: None,
            end: None,
        });
        data.events.insert(
            identity,
            SavedEvent {
                item,
                media,
                raw: Some(value),
            },
        );
    }

    fn supplement(
        &self,
        data: &mut Data,
        key: &MentionKey,
        item: &TimelineItem,
        media: &[MediaRegistration],
        raw: &Value,
    ) {
        let TimelineItem::Event(event) = item else {
            return;
        };
        let identity = id(key);
        if data
            .events
            .get(&identity)
            .is_some_and(|saved| saved.item != *item)
        {
            data.touch_contexts(&identity);
        }
        let mut retained = data.events.contains_key(&identity);
        let saved_events = &data.events;
        for (target, window) in &mut data.windows {
            let Some(entry) = data.entries.get_mut(target) else {
                continue;
            };
            if entry.key.room_id != key.room_id
                || (entry.timestamp_ms, target.as_str()) >= (event.timestamp_ms, identity.as_str())
                || window.after.contains(&identity)
            {
                continue;
            }
            let previous = window.after.clone();
            window.after.push(identity.clone());
            window.after.sort_by_key(|key| {
                if key == &identity {
                    (event.timestamp_ms, key.clone())
                } else {
                    (
                        saved_events
                            .get(key)
                            .and_then(|saved| match &saved.item {
                                TimelineItem::Event(event) => Some(event.timestamp_ms),
                                _ => None,
                            })
                            .unwrap_or(0),
                        key.clone(),
                    )
                }
            });
            window.after.truncate(10);
            if window.after == previous {
                continue;
            }
            window.end = None;
            retained |= window.after.contains(&identity);
            if entry.capture == MentionCaptureState::Partial
                && window.after.len() == 10
                && (window.before.len() == 10 || window.start.is_none())
            {
                entry.capture = MentionCaptureState::Ready;
            }
            entry.revision = data.revision + 1;
        }
        if retained {
            data.events.insert(
                identity,
                SavedEvent {
                    item: item.clone(),
                    media: media.to_vec(),
                    raw: Some(raw.clone()),
                },
            );
        }
    }

    async fn classify(
        &self,
        data: &Data,
        room_id: &OwnedRoomId,
        value: &Value,
    ) -> (Vec<MentionReason>, bool, bool) {
        let Some(sender) = value["sender"]
            .as_str()
            .and_then(|sender| sender.parse::<OwnedUserId>().ok())
        else {
            return (vec![], false, false);
        };
        if sender == self.account
            || self.ignored.contains(&sender)
            || value["type"] != "m.room.message"
        {
            return (vec![], false, false);
        }
        let Some(room) = self.observable(room_id) else {
            return (vec![], false, false);
        };
        let raw: Raw<AnySyncTimelineEvent> =
            Raw::from_json_string(value.to_string()).expect("serialised event");
        let actions = room.event_push_actions(&raw).await.ok().flatten();
        let notify = actions.as_ref().is_some_and(|actions| {
            actions
                .iter()
                .any(|action| matches!(action, ruma::push::Action::Notify))
        });
        let highlight = actions.as_ref().is_some_and(|actions| actions.iter().any(|action| matches!(action, ruma::push::Action::SetTweak(ruma::push::Tweak::Highlight(value)) if value != &ruma::push::HighlightTweakValue::No)));
        let direct = value["content"]["m.mentions"]["user_ids"]
            .as_array()
            .is_some_and(|users| {
                users
                    .iter()
                    .any(|user| user.as_str() == Some(self.account.as_str()))
            });
        let mut reasons = Vec::new();
        if direct {
            reasons.push(MentionReason::Direct);
        }
        let mut unresolved = false;
        if let Some(reply) = value["content"]["m.relates_to"]["m.in_reply_to"]["event_id"]
            .as_str()
            .and_then(|value| value.parse().ok())
        {
            let target = data.events.get(&id(&MentionKey {
                room_id: room_id.clone(),
                event_id: reply,
            }));
            let reply_to_self = value["unsigned"]["macltok_reply_sender"].as_str() == Some(self.account.as_str()) || target.is_some_and(|target| matches!(&target.item, TimelineItem::Event(event) if event.sender.user_id == self.account));
            if reply_to_self && (direct || (notify && highlight)) {
                reasons.push(MentionReason::Reply);
            } else if target.is_none()
                && value["unsigned"]["macltok_reply_sender"].as_str().is_none()
                && (direct || (notify && highlight))
            {
                unresolved = true;
            }
        }
        if value["content"]["m.mentions"]["room"] == true {
            match room.power_levels().await {
                Ok(levels) if levels.user_can_trigger_room_notification(&sender) => {
                    if notify && highlight {
                        reasons.push(MentionReason::Room);
                    } else if actions.is_none() {
                        unresolved = true;
                    }
                }
                Err(_) => unresolved = true,
                _ => {}
            }
        }
        if reasons.is_empty() && notify && highlight {
            reasons.push(MentionReason::Keyword);
        }
        (reasons, !notify, unresolved)
    }

    pub(crate) async fn run(mut self) {
        let reports_client = self.client.clone();
        let reports = reports_client
            .event_cache()
            .subscribe_to_decryption_reports();
        futures_util::pin_mut!(reports);
        let mut ignores = self.client.subscribe_to_ignore_user_list_changes();
        let mut work: FuturesUnordered<EngineBoxFuture<'static, Work>> = FuturesUnordered::new();
        let mut captures = BTreeSet::new();
        let mut retries = BTreeSet::new();
        let mut history = false;
        let mut next_scan = now() + 60_000;
        let mut next_expiry = 0;
        let mut tick = Box::pin(matrix_sdk::sleep::sleep(Duration::from_secs(1)));
        loop {
            self.schedule(&mut work, &mut captures, &mut retries, &mut history);
            tokio::select! {
                biased;
                request = self.requests.recv() => match request {
                    Some(Request::Stop(reply)) => {
                        if self.dirty && let Err(error) = self.flush().await {
                            tracing::error!(%error, "saving the mentions inbox at shutdown failed");
                        }
                        self.media.release(&self.media_scope).await;
                        let _ = reply.send(());
                        return;
                    }
                    Some(Request::Batch(batch, reply)) => { self.batch(batch).await; let _ = reply.send(()); }
                    Some(Request::Command(command, reply)) => { let result = self.command(command).await; let _ = reply.send(result); }
                    None => return,
                },
                Some(result) = work.next(), if !work.is_empty() => {
                    let rooms = result.rooms();
                    let before = self.data.scope(&rooms);
                    let mut next = std::mem::take(&mut self.data);
                    // a candidate that failed again only moved its retry time, which needs no write
                    let changed = self.complete(&mut next, result, &mut captures, &mut retries, &mut history).await
                        && next.scope(&rooms) != before;
                    self.data = next;
                    self.dirty |= changed;
                    self.unpublished |= changed;
                }
                Some(report) = reports.next() => {
                    let room = match report { Ok(RedecryptorReport::ResolvedUtds { room_id, .. }) => Some(room_id), _ => None };
                    for candidate in self.data.candidates.values_mut().filter(|candidate| candidate.encrypted() && room.as_ref().is_none_or(|room| *room == candidate.room_id)) { candidate.retry_at_ms = 0; }
                }
                Some(_) = ignores.next() => {
                    self.ignored = self.ignored_users().await;
                    let mut next = std::mem::take(&mut self.data);
                    let changed = self.filter_ignored(&mut next);
                    self.data = next;
                    self.dirty |= changed;
                    self.unpublished |= changed;
                }
                () = &mut tick => {
                    tick = Box::pin(matrix_sdk::sleep::sleep(Duration::from_secs(1)));
                    self.schedule_profiles(&mut work);
                    if now() >= next_scan && !history { self.recovery_budget = 20; self.latest_scan = true; next_scan = now() + 60_000; }
                    let direct = direct_rooms(&self.client);
                    if direct != self.direct { self.direct = direct; self.dirty = true; self.unpublished = true; }
                    if now() >= next_expiry { let expired = self.data.expire_candidates(now()); self.dirty |= expired; self.unpublished |= expired; next_expiry = now() + 60_000; }
                    #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
                    if self.store.is_none() && self.reopen.is_some() && now() >= self.flush_after { self.reopen().await; }
                    if self.dirty && now() >= self.flush_after {
                        if let Err(error) = self.flush().await { self.store_failed(error).await; }
                    } else if self.unpublished && self.advance().is_ok() {
                        self.publish().await;
                    }
                }
            }
        }
    }

    async fn complete(
        &mut self,
        next: &mut Data,
        result: Work,
        captures: &mut BTreeSet<String>,
        retries: &mut BTreeSet<String>,
        history: &mut bool,
    ) -> bool {
        match result {
            Work::Capture(key, revision, result) => {
                captures.remove(&key);
                let observed = result
                    .as_ref()
                    .ok()
                    .map(|(_, _, observed)| observed.clone())
                    .unwrap_or_default();
                let room_id = next
                    .entries
                    .get(&key)
                    .filter(|entry| entry.revision == revision)
                    .map(|entry| entry.key.room_id.clone())
                    .filter(|room_id| {
                        self.client
                            .get_room(room_id)
                            .is_some_and(|room| room.state() == RoomState::Joined)
                    });
                self.captured(next, key, revision, result);
                if let Some(room_id) = room_id {
                    for value in observed {
                        let identity = value["event_id"]
                            .as_str()
                            .map(|event_id| format!("{room_id}\n{event_id}"));
                        if identity.as_ref().is_some_and(|key| {
                            next.events
                                .get(key)
                                .is_some_and(|event| event.raw.is_some())
                        }) {
                            continue;
                        }
                        self.observe(next, &room_id, value).await;
                    }
                }
            }
            Work::Candidate(key, result) => {
                retries.remove(&key);
                if let Some(mut candidate) = next.candidates.remove(&key) {
                    match result {
                        Ok(value) => {
                            self.observe(next, &candidate.room_id, value).await;
                            if let Some(unresolved) = next.candidates.get_mut(&key) {
                                unresolved.attempts = candidate.attempts.saturating_add(1);
                                unresolved.first_seen_ms = candidate.first_seen_ms;
                                unresolved.retry_at_ms =
                                    now() + (1000u64 << unresolved.attempts.min(8));
                            }
                        }
                        Err(error) => {
                            if !candidate.encrypted()
                                && (matches!(
                                    error.kind,
                                    ErrorKind::EventNotFound | ErrorKind::Forbidden
                                ) || error
                                    .server
                                    .as_ref()
                                    .is_some_and(|server| matches!(server.status, 403 | 404)))
                            {
                                next.gap(
                                    Some(candidate.room_id),
                                    MentionGapReason::ContextUnavailable,
                                    None,
                                );
                            } else {
                                candidate.attempts = candidate.attempts.saturating_add(1);
                                candidate.retry_at_ms =
                                    retry_at(&error, 1000u64 << candidate.attempts.min(8));
                                next.candidates.insert(key, candidate);
                                return false;
                            }
                        }
                    }
                }
            }
            Work::Profiles(room_id, users, found) => {
                self.profiling = false;
                self.profiled
                    .extend(users.into_iter().map(|user| (room_id.clone(), user)));
                let bare = |item: &TimelineItem| match item {
                    TimelineItem::Event(event)
                        if event.sender.display_name.is_none() && event.sender.avatar.is_none() =>
                    {
                        found.get(&event.sender.user_id).cloned()
                    }
                    _ => None,
                };
                let prefix = format!("{room_id}\n");
                let mut repaired = Vec::new();
                for (key, saved) in next.events.range_mut(prefix.clone()..) {
                    if !key.starts_with(&prefix) {
                        break;
                    }
                    if let Some(sender) = bare(&saved.item)
                        && let TimelineItem::Event(event) = &mut saved.item
                    {
                        event.sender = sender;
                        repaired.push(key.clone());
                    }
                }
                let revision = next.revision + 1;
                for (key, entry) in next.entries.range_mut(prefix.clone()..) {
                    if !key.starts_with(&prefix) {
                        break;
                    }
                    if let Some(sender) = bare(&entry.event)
                        && let TimelineItem::Event(event) = &mut entry.event
                    {
                        event.sender = sender;
                        entry.revision = revision;
                    }
                }
                for key in &repaired {
                    next.touch_contexts(key);
                }
            }
            Work::History(latest, from, result) => {
                *history = false;
                self.recovery_budget = self.recovery_budget.saturating_sub(1);
                match result {
                    Ok(page) => {
                        next.coverage.historical_supported = Some(true);
                        let mut reached_anchor = false;
                        for notification in page.notifications {
                            if let Some(event_id) = notification
                                .event
                                .get_field::<String>("event_id")
                                .ok()
                                .flatten()
                            {
                                let identity = format!("{}\n{}", notification.room_id, event_id);
                                if latest && from.is_none() && next.latest_head.is_none() {
                                    next.latest_head = Some(identity.clone());
                                }
                                if latest && next.latest_anchor.as_ref() == Some(&identity) {
                                    reached_anchor = true;
                                    break;
                                }
                            }
                            if self
                                .client
                                .get_room(&notification.room_id)
                                .is_some_and(|room| room.state() == RoomState::Joined)
                                && let Ok(value) =
                                    serde_json::from_str(notification.event.json().get())
                            {
                                self.observe(next, &notification.room_id, value).await;
                            }
                        }
                        let token = page
                            .next_token
                            .filter(|token| Some(token) != from.as_ref() && !reached_anchor);
                        if latest {
                            next.latest_cursor = token.clone();
                            if token.is_none() {
                                self.latest_scan = false;
                                next.latest_anchor =
                                    next.latest_head.take().or(next.latest_anchor.take());
                            }
                        } else {
                            next.history_cursor = token.clone();
                            next.history_exhausted = token.is_none();
                        }
                    }
                    Err(error) => {
                        if error
                            .server
                            .as_ref()
                            .is_some_and(|server| matches!(server.status, 404 | 405 | 501))
                        {
                            next.coverage.historical_supported = Some(false);
                            self.recovery_budget = 0;
                        }
                        if error
                            .server
                            .as_ref()
                            .is_some_and(|server| server.status == 400)
                        {
                            if latest {
                                next.latest_cursor = None;
                                next.latest_head = None;
                            } else {
                                next.history_cursor = None;
                            }
                        }
                        next.gap(None, MentionGapReason::NotificationHistoryUnavailable, from);
                        self.recovery_retry = retry_at(&error, 60_000);
                    }
                }
                next.coverage.recovering = self.recovery_budget > 0
                    && (self.latest_scan || !next.history_exhausted)
                    && next.coverage.historical_supported != Some(false);
            }
        }
        true
    }

    /// rows observed after their window's capture never got a profile, so one room at a time
    /// looks its bare senders up in the member store. get_member would fetch /members for every room
    /// whose members aren't synced, one after another, and that can starve sync
    fn schedule_profiles(&mut self, work: &mut FuturesUnordered<EngineBoxFuture<'static, Work>>) {
        if self.profiling {
            return;
        }
        let mut wanted: Option<(OwnedRoomId, BTreeSet<OwnedUserId>)> = None;
        let items = self
            .data
            .events
            .iter()
            .map(|(key, saved)| (key, &saved.item))
            .chain(
                self.data
                    .entries
                    .iter()
                    .map(|(key, entry)| (key, &entry.event)),
            );
        for (key, item) in items {
            let TimelineItem::Event(event) = item else {
                continue;
            };
            if event.sender.display_name.is_some() || event.sender.avatar.is_some() {
                continue;
            }
            let Some(room_id) = key
                .split_once('\n')
                .and_then(|(room, _)| OwnedRoomId::try_from(room).ok())
            else {
                continue;
            };
            if self.direct.contains(&room_id)
                || self
                    .profiled
                    .contains(&(room_id.clone(), event.sender.user_id.clone()))
            {
                continue;
            }
            match &mut wanted {
                None => wanted = Some((room_id, BTreeSet::from([event.sender.user_id.clone()]))),
                Some((room, users)) if *room == room_id && users.len() < 50 => {
                    users.insert(event.sender.user_id.clone());
                }
                Some(_) => {}
            }
        }
        let Some((room_id, users)) = wanted else {
            return;
        };
        let Some(room) = self
            .client
            .get_room(&room_id)
            .filter(|room| room.state() == RoomState::Joined)
        else {
            self.profiled
                .extend(users.into_iter().map(|user| (room_id.clone(), user)));
            return;
        };
        self.profiling = true;
        let users: Vec<_> = users.into_iter().collect();
        work.push(Box::pin(async move {
            let found = futures_util::future::join_all(users.iter().map(|user_id| {
                let room = &room;
                async move {
                    crate::normalise::stored_member_sender(room, user_id)
                        .await
                        .map(|sender| (user_id.clone(), sender))
                }
            }))
            .await
            .into_iter()
            .flatten()
            .collect();
            Work::Profiles(room_id, users, found)
        }));
    }

    fn schedule(
        &mut self,
        work: &mut FuturesUnordered<EngineBoxFuture<'static, Work>>,
        captures: &mut BTreeSet<String>,
        retries: &mut BTreeSet<String>,
        history: &mut bool,
    ) {
        for (key, entry) in self.data.entries.iter().filter(|(_, entry)| matches!(entry.capture, MentionCaptureState::Pending) || matches!(entry.capture, MentionCaptureState::Failed {retry_at_ms, ..} if retry_at_ms <= now())).take(128) {
            if captures.len() >= 2 { break; }
            if captures.contains(key) || self.direct.contains(&entry.key.room_id) { continue; }
            let Some(room) = self.client.get_room(&entry.key.room_id).filter(|room| room.state() == RoomState::Joined) else { continue; };
            captures.insert(key.clone());
            let key = key.clone(); let event_id = entry.key.event_id.clone(); let account = self.account.clone(); let revision = entry.revision;
            work.push(Box::pin(async move { Work::Capture(key, revision, context::capture(room, Some(account), event_id, 10, 10).await) }));
        }
        for (key, candidate) in &self.data.candidates {
            if retries.len() >= 2 {
                break;
            }
            if candidate.retry_at_ms > now() || retries.contains(key) {
                continue;
            }
            let Some(room) = self
                .client
                .get_room(&candidate.room_id)
                .filter(|room| room.state() == RoomState::Joined)
            else {
                continue;
            };
            let key = key.clone();
            let candidate = candidate.clone();
            let key_for_work = key.clone();
            retries.insert(key);
            work.push(Box::pin(async move {
                let result = async {
                    if candidate.raw["type"] == "m.room.encrypted" {
                        let raw: Raw<OriginalSyncRoomEncryptedEvent> =
                            Raw::from_json_string(candidate.raw.to_string()).map_err(|error| {
                                Error::with_source(
                                    ErrorKind::Invalid,
                                    "invalid encrypted mention candidate",
                                    error,
                                )
                            })?;
                        let event = room.decrypt_event(&raw, None).await.map_err(|error| {
                            Error::from_sdk("decrypting a mention candidate failed", error)
                        })?;
                        let value: Value =
                            serde_json::from_str(event.raw().json().get()).map_err(|error| {
                                Error::with_source(
                                    ErrorKind::Invalid,
                                    "invalid decrypted mention",
                                    error,
                                )
                            })?;
                        if value["type"] == "m.room.encrypted" {
                            return Err(Error::new(
                                ErrorKind::Encryption,
                                "mention candidate still awaits keys",
                            ));
                        }
                        return Ok(value);
                    }
                    let target = candidate.raw["content"]["m.relates_to"]["event_id"]
                        .as_str()
                        .filter(|_| {
                            candidate.raw["content"]["m.relates_to"]["rel_type"] == "m.replace"
                        });
                    if let Some(target) =
                        target.and_then(|target| target.parse::<ruma::OwnedEventId>().ok())
                    {
                        let event = room.event(&target, None).await.map_err(|error| {
                            Error::from_sdk("fetching an edited mention failed", error)
                        })?;
                        let original: Value = serde_json::from_str(event.raw().json().get())
                            .map_err(|error| {
                                Error::with_source(
                                    ErrorKind::Invalid,
                                    "invalid edited mention",
                                    error,
                                )
                            })?;
                        if original["sender"] == candidate.raw["sender"] {
                            let mut original = edited(original, &candidate.raw);
                            original["unsigned"]["macltok_edit"] = candidate.raw;
                            return Ok(original);
                        }
                        return Err(Error::new(
                            ErrorKind::Forbidden,
                            "an edit belongs to a different sender",
                        ));
                    }
                    let mut value = candidate.raw;
                    if let Some(target) =
                        value["content"]["m.relates_to"]["m.in_reply_to"]["event_id"]
                            .as_str()
                            .and_then(|target| target.parse::<ruma::OwnedEventId>().ok())
                    {
                        let event = room.event(&target, None).await.map_err(|error| {
                            Error::from_sdk("fetching a mention reply target failed", error)
                        })?;
                        let target: Value = serde_json::from_str(event.raw().json().get())
                            .map_err(|error| {
                                Error::with_source(
                                    ErrorKind::Invalid,
                                    "invalid mention reply target",
                                    error,
                                )
                            })?;
                        value["unsigned"]["macltok_reply_sender"] = target["sender"].clone();
                    }
                    Ok(value)
                }
                .await;
                Work::Candidate(key_for_work, result)
            }));
        }
        if !*history
            && self.recovery_budget > 0
            && now() >= self.recovery_retry
            && (self.latest_scan || !self.data.history_exhausted)
            && self.data.coverage.historical_supported != Some(false)
        {
            *history = true;
            let latest = self.latest_scan;
            let from = if latest {
                self.data.latest_cursor.clone()
            } else {
                self.data.history_cursor.clone()
            };
            let client = self.client.clone();
            work.push(Box::pin(async move {
                let mut request = ruma::api::client::push::get_notifications::v3::Request::new();
                request.from = from.clone();
                request.limit = Some(ruma::uint!(50));
                let result = client.send(request).await.map_err(|error| {
                    Error::from_sdk("fetching mention history failed", error.into())
                });
                Work::History(latest, from, result)
            }));
        }
    }

    fn captured(
        &self,
        data: &mut Data,
        key: String,
        revision: u64,
        result: Result<(crate::EventContext, Vec<MediaRegistration>, Vec<Value>), Error>,
    ) {
        let Some(entry) = data.entries.get(&key) else {
            return;
        };
        if entry.revision != revision {
            return;
        }
        let room_id = entry.key.room_id.clone();
        if !self
            .client
            .get_room(&room_id)
            .is_some_and(|room| room.state() == RoomState::Joined)
        {
            return;
        }
        match result {
            Ok((context, media, _)) => {
                data.capture_attempts.remove(&key);
                let live_after = data
                    .windows
                    .get(&key)
                    .map(|window| window.after.clone())
                    .unwrap_or_default();
                let mut insert = |item: TimelineItem| -> Option<String> {
                    let TimelineItem::Event(event) = &item else {
                        return None;
                    };
                    let event_id = event.event_id.clone()?;
                    let event_key = id(&MentionKey {
                        room_id: room_id.clone(),
                        event_id,
                    });
                    let registrations = media_for_item(&item, &media);
                    let mut saved = SavedEvent {
                        item,
                        media: registrations,
                        raw: None,
                    };
                    if let Some(existing) = data.events.get(&event_key) {
                        let captured = saved.item;
                        saved = existing.clone();
                        improve(&mut saved.item, &captured);
                    }
                    if data.redacted.contains_key(&event_key) {
                        redact(&mut saved);
                    }
                    data.events.insert(event_key.clone(), saved);
                    Some(event_key)
                };
                let before: Vec<_> = context
                    .events_before
                    .into_iter()
                    .filter_map(&mut insert)
                    .collect();
                let captured_target = context.event.clone();
                let target = insert(context.event);
                let mut after: Vec<_> = context
                    .events_after
                    .into_iter()
                    .filter_map(&mut insert)
                    .collect();
                let extended = live_after.iter().any(|key| !after.contains(key));
                after.extend(live_after);
                after.sort_by_key(|key| {
                    (
                        data.events
                            .get(key)
                            .and_then(|saved| match &saved.item {
                                TimelineItem::Event(event) => Some(event.timestamp_ms),
                                _ => None,
                            })
                            .unwrap_or(0),
                        key.clone(),
                    )
                });
                after.dedup();
                after.truncate(10);
                let entry = data.entries.get_mut(&key).unwrap();
                improve(&mut entry.event, &captured_target);
                entry.capture = if target.is_none() {
                    MentionCaptureState::Unavailable
                } else if after.len() < 10 || (before.len() < 10 && context.start_token.is_some()) {
                    MentionCaptureState::Partial
                } else {
                    MentionCaptureState::Ready
                };
                entry.revision = data.revision + 1;
                data.windows.insert(
                    key,
                    Window {
                        before,
                        target,
                        after,
                        start: context.start_token,
                        end: if extended { None } else { context.end_token },
                    },
                );
            }
            Err(error) => {
                let attempts = data.capture_attempts.entry(key.clone()).or_default();
                *attempts = attempts.saturating_add(1);
                let delay = (1000u64 << (*attempts).min(11)).min(3_600_000);
                let entry = data.entries.get_mut(&key).unwrap();
                entry.capture = if matches!(
                    error.kind,
                    ErrorKind::EventNotFound | ErrorKind::Forbidden | ErrorKind::RoomUnknown
                ) || error
                    .server
                    .as_ref()
                    .is_some_and(|server| matches!(server.status, 403 | 404))
                {
                    MentionCaptureState::Unavailable
                } else {
                    MentionCaptureState::Failed {
                        message: error.to_string(),
                        retry_at_ms: retry_at(&error, delay),
                    }
                };
                entry.revision = data.revision + 1;
            }
        }
    }
}

fn direct_rooms(client: &Client) -> BTreeSet<OwnedRoomId> {
    client
        .rooms()
        .into_iter()
        .filter(|room| room.direct_targets_length() > 0)
        .map(|room| room.room_id().to_owned())
        .collect()
}

fn edited(mut original: Value, edit: &Value) -> Value {
    original["content"] = edit["content"]["m.new_content"].clone();
    original["unsigned"]["macltok_edited"] = Value::Bool(true);
    original
}

fn retry_at(error: &Error, fallback_ms: u64) -> u64 {
    let mut source = std::error::Error::source(error);
    while let Some(current) = source {
        if let Some(sdk) = current.downcast_ref::<matrix_sdk::Error>() {
            if let Some(ruma::api::error::ErrorKind::LimitExceeded(data)) =
                sdk.client_api_error_kind()
            {
                if let Some(delay) = data.retry_after {
                    let delay = match delay {
                        ruma::api::error::RetryAfter::Delay(delay) => delay,
                        ruma::api::error::RetryAfter::DateTime(time) => time
                            .duration_since(ruma::time::SystemTime::now())
                            .unwrap_or_default(),
                    };
                    return now().saturating_add(
                        u64::try_from(delay.as_millis())
                            .unwrap_or(u64::MAX)
                            .max(fallback_ms),
                    );
                }
            }
        }
        source = current.source();
    }
    now().saturating_add(fallback_ms)
}

/// the capture fetched what observing at sync time couldn't, so its profile and reply win
/// over the saved ones, while the saved content keeps its own edits and redactions. observing
/// again calls it the other way round, or every sync in the room wipes the profile
fn improve(saved: &mut TimelineItem, captured: &TimelineItem) {
    let (TimelineItem::Event(saved), TimelineItem::Event(captured)) = (saved, captured) else {
        return;
    };
    if saved.event_id != captured.event_id {
        return;
    }
    if captured.sender.display_name.is_some() || captured.sender.avatar.is_some() {
        saved.sender = captured.sender.clone();
    }
    if let (Some(reply), Some(fetched)) = (&mut saved.reply_to, &captured.reply_to)
        && !matches!(reply.details, crate::ReplyDetails::Ready { .. })
        && matches!(fetched.details, crate::ReplyDetails::Ready { .. })
    {
        reply.details = fetched.details.clone();
    }
}

fn redact(event: &mut SavedEvent) {
    if let TimelineItem::Event(item) = &mut event.item {
        item.content = TimelineContent::Redacted;
        item.reply_to = None;
        item.reactions.clear();
    }
    event.media.clear();
    event.raw = None;
}

fn media_for_item(
    item: &TimelineItem,
    registrations: &[MediaRegistration],
) -> Vec<MediaRegistration> {
    let TimelineItem::Event(event) = item else {
        return Vec::new();
    };
    let mut ids = Vec::new();
    match &event.content {
        TimelineContent::Message(message) => {
            ids.extend(message.media.iter().map(|media| &media.id));
            ids.extend(
                message
                    .link_previews
                    .iter()
                    .filter_map(|preview| preview.image.as_ref().map(|image| &image.media_id)),
            );
        }
        TimelineContent::Sticker { media, .. } => ids.push(&media.id),
        _ => {}
    }
    registrations
        .iter()
        .filter(|registration| ids.contains(&&registration.id))
        .cloned()
        .collect()
}

pub(crate) async fn watch_updates(mut updates: watch::Receiver<MentionsPage>, events: EventSender) {
    let mut previous: BTreeMap<String, MentionEntry> = BTreeMap::new();
    loop {
        let snapshot = updates.borrow_and_update().clone();
        let current: BTreeMap<_, _> = snapshot
            .entries
            .iter()
            .map(|entry| (id(&entry.key), entry.clone()))
            .collect();
        let upserts = current
            .iter()
            .filter(|(key, entry)| previous.get(*key) != Some(entry))
            .map(|(_, entry)| entry.clone())
            .collect();
        let removed = previous
            .iter()
            .filter(|(key, _)| !current.contains_key(*key))
            .map(|(_, entry)| entry.key.clone())
            .collect();
        previous = current;
        if events
            .send(Event::MentionsChanged(MentionsUpdate {
                snapshot,
                upserts,
                removed,
            }))
            .await
            .is_err()
        {
            return;
        }
        if updates.changed().await.is_err() {
            return;
        }
    }
}