macltok

public

Star0

Watch0

31 commits 3 branches 0 tags 0 releases3.0 MiB

use std::{
    collections::HashMap,
    fmt,
    pin::Pin,
    sync::{Arc, Mutex, MutexGuard, PoisonError},
    task::{Context, Poll},
    time::Duration,
};

use futures_util::{Stream, StreamExt};
use imbl::Vector;
use matrix_sdk::{
    SendOutsideWasm,
    executor::{AbortHandle, spawn},
};
use ruma::{OwnedDeviceId, OwnedRoomId, OwnedTransactionId, OwnedUserId};
#[cfg(feature = "serde")]
use serde::{Deserialize, Serialize};
use tokio::sync::broadcast;

use crate::{
    ClientHandle, Command, ConnectionState, EncryptionState, EngineFailure, ErrorKind, Event,
    HistoryState, RoomSummary, TimelineCommand, TimelineDiff, TimelineId, TimelineItem,
    TypingState, UploadOutcome, VerificationState, VerificationTarget,
    engine::executor::EngineBoxFuture,
};

#[cfg(not(target_family = "wasm"))]
type EngineBoxStream<T> = futures_util::stream::BoxStream<'static, T>;
#[cfg(target_family = "wasm")]
type EngineBoxStream<T> = futures_util::stream::LocalBoxStream<'static, T>;

const DEFAULT_CAPACITY: usize = 256;
const RESYNC_BACKOFF: Duration = Duration::from_millis(100);
const RESYNC_BACKOFF_CAP: Duration = Duration::from_secs(10);

/// everything the engine has said so far, folded together
#[derive(Clone, Debug, PartialEq)]
#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
#[non_exhaustive]
pub struct State {
    pub user_id: Option<OwnedUserId>,
    pub session: SessionStatus,
    pub connection: ConnectionState,
    pub encryption: Option<EncryptionState>,
    pub verification: Option<VerificationState>,
    pub verification_target: Option<VerificationTarget>,
    pub rooms: Vector<RoomSummary>,
    pub timelines: imbl::HashMap<TimelineId, TimelineState>,
    pub typing: imbl::HashMap<OwnedRoomId, Vec<OwnedUserId>>,
    pub uploads: imbl::HashMap<OwnedTransactionId, UploadState>,
    /// what stopped the engine, if something went wrong
    pub engine_failure: Option<EngineFailure>,
    /// the last save of the session record that failed, until one succeeds
    pub session_save_failure: Option<String>,
}

#[derive(Clone, Debug, PartialEq)]
#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
#[non_exhaustive]
pub struct TimelineState {
    pub items: Vector<TimelineItem>,
    pub history: HistoryState,
    /// a batch didn't fit, so `items` is the last good copy until the next reset
    pub diverged: bool,
}

#[derive(Clone, Debug, Eq, PartialEq)]
#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
#[non_exhaustive]
pub struct UploadState {
    pub room_id: OwnedRoomId,
    pub uploaded_bytes: u64,
    pub total_bytes: u64,
    pub complete: bool,
    pub failure: Option<UploadFailure>,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
#[non_exhaustive]
pub struct UploadFailure {
    pub kind: ErrorKind,
    pub will_retry: bool,
}

/// the `Termination` it'll end in, as far as the events tell
#[derive(Clone, Debug, Eq, PartialEq)]
#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
#[non_exhaustive]
pub enum SessionStatus {
    Active,
    SoftLoggedOut {
        device_id: OwnedDeviceId,
    },
    AccessRevoked,
    LoggedOut,
    /// stopped with the session still good
    Stopped,
}

#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct Update {
    /// one more than the update before it, and a `State` carries the one it's current as of
    pub sequence: u64,
    pub change: StateChange,
}

#[derive(Clone, Debug)]
#[non_exhaustive]
pub enum StateChange {
    /// start over from this. a subscription opens with one and gets another after lagging
    State(Arc<State>),
    Authenticated(OwnedUserId),
    Session(SessionStatus),
    /// the record on disk may now hold a spent refresh token
    SessionSaveFailed(String),
    /// saved again after `SessionSaveFailed`
    SessionSaved,
    Connection(ConnectionState),
    Encryption(EncryptionState),
    Verification(VerificationState),
    VerificationTarget(VerificationTarget),
    /// room modes arrive with the `Rooms` that follows
    NotificationSettings,
    Rooms(Vector<RoomSummary>),
    Timeline {
        timeline_id: TimelineId,
        change: TimelineChange,
    },
    History {
        timeline_id: TimelineId,
        state: HistoryState,
    },
    Typing(TypingState),
    Upload {
        transaction_id: OwnedTransactionId,
        upload: UploadState,
    },
    UploadFinished {
        room_id: OwnedRoomId,
        transaction_id: OwnedTransactionId,
        outcome: UploadOutcome,
    },
    Failed(EngineFailure),
}

#[derive(Clone, Debug)]
#[non_exhaustive]
pub enum TimelineChange {
    Reset {
        items: Vector<TimelineItem>,
    },
    /// `items` is what applying `diffs` left
    Changed {
        diffs: Arc<[TimelineDiff]>,
        items: Vector<TimelineItem>,
    },
    /// a batch didn't fit. a resync is on its way and nothing changes until its reset
    Diverged,
    Closed,
}

/// drains the event stream into a `State` that any number of `MirrorHandle`s watch
pub struct Mirror {
    handle: ClientHandle,
    events: EngineBoxStream<Event>,
    shared: Arc<Shared>,
    resyncs: HashMap<TimelineId, AbortHandle>,
    rooms_fetch: Option<AbortHandle>,
}

impl fmt::Debug for Mirror {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.debug_struct("Mirror").finish_non_exhaustive()
    }
}

impl Mirror {
    /// takes any stream of the engine's events, so a host can still look at them on the way past
    pub fn new(
        handle: ClientHandle,
        events: impl Stream<Item = Event> + SendOutsideWasm + 'static,
    ) -> (Self, MirrorHandle) {
        Self::with_capacity(handle, events, DEFAULT_CAPACITY)
    }

    /// `capacity` updates can queue for a subscriber before it lags and starts over. a zero
    /// capacity counts as one
    pub fn with_capacity(
        handle: ClientHandle,
        events: impl Stream<Item = Event> + SendOutsideWasm + 'static,
        capacity: usize,
    ) -> (Self, MirrorHandle) {
        let state = State {
            user_id: None,
            session: SessionStatus::Active,
            connection: *handle.connection_state().borrow(),
            encryption: handle.encryption_state(),
            verification: None,
            verification_target: None,
            rooms: Vector::new(),
            timelines: imbl::HashMap::new(),
            typing: imbl::HashMap::new(),
            uploads: imbl::HashMap::new(),
            engine_failure: None,
            session_save_failure: None,
        };
        let shared = Arc::new(Shared {
            published: Mutex::new(Published {
                state: Arc::new(state),
                sequence: 0,
                updates: Some(broadcast::channel(capacity.max(1)).0),
                rooms_fetch: RoomsFetch::Idle,
            }),
        });
        #[cfg(not(target_family = "wasm"))]
        let events = events.boxed();
        #[cfg(target_family = "wasm")]
        let events = events.boxed_local();
        let mirror = Self {
            handle,
            events,
            shared: shared.clone(),
            resyncs: HashMap::new(),
            rooms_fetch: None,
        };
        (mirror, MirrorHandle { shared })
    }

    /// resolves once the event stream ends, which is just after `Stopped`
    pub fn run(mut self) -> EngineBoxFuture<'static, ()> {
        Box::pin(async move {
            self.refetch_rooms();
            while let Some(event) = self.events.next().await {
                self.apply(event);
            }
        })
    }

    fn apply(&mut self, event: Event) {
        match event {
            Event::ConnectionChanged(connection) => self.publish(|state| {
                state.connection = connection;
                Some(StateChange::Connection(connection))
            }),
            Event::Authenticated { user_id } => self.publish(|state| {
                state.user_id = Some(user_id.clone());
                Some(StateChange::Authenticated(user_id))
            }),
            Event::SessionSaveFailed { message } => self.publish(|state| {
                state.session_save_failure = Some(message.clone());
                Some(StateChange::SessionSaveFailed(message))
            }),
            Event::SessionTokensChanged => self.publish(|state| {
                state.session_save_failure.take()?;
                Some(StateChange::SessionSaved)
            }),
            Event::SessionAccessRejected { soft_logout } => self.publish(|state| {
                state.session = match soft_logout {
                    Some(device_id) => SessionStatus::SoftLoggedOut { device_id },
                    None => SessionStatus::AccessRevoked,
                };
                Some(StateChange::Session(state.session.clone()))
            }),
            Event::LoggedOut => self.publish(|state| {
                state.session = SessionStatus::LoggedOut;
                Some(StateChange::Session(SessionStatus::LoggedOut))
            }),
            Event::Stopped => {
                self.cancel_resyncs();
                self.publish(|state| {
                    if state.session != SessionStatus::Active {
                        return None;
                    }
                    state.session = SessionStatus::Stopped;
                    Some(StateChange::Session(SessionStatus::Stopped))
                });
            }
            Event::Failed(failure) => self.publish(|state| {
                state.engine_failure = Some(failure.clone());
                Some(StateChange::Failed(failure))
            }),
            Event::NotificationSettingsChanged => {
                self.publish(|_| Some(StateChange::NotificationSettings));
            }
            Event::RoomsChanged => self.refetch_rooms(),
            Event::EncryptionChanged(encryption) => self.publish(|state| {
                state.encryption = Some(encryption.clone());
                Some(StateChange::Encryption(encryption))
            }),
            Event::VerificationChanged(verification) => self.publish(|state| {
                state.verification = Some(verification.clone());
                Some(StateChange::Verification(verification))
            }),
            Event::VerificationTargetChanged(target) => self.publish(|state| {
                state.verification_target = Some(target.clone());
                Some(StateChange::VerificationTarget(target))
            }),
            Event::TimelineReset { timeline_id, items } => {
                self.cancel_resync(&timeline_id);
                self.publish(|state| {
                    let history = state
                        .timelines
                        .get(&timeline_id)
                        .map_or(HistoryState::Idle, |timeline| timeline.history.clone());
                    let timeline = TimelineState {
                        items: items.clone(),
                        history,
                        diverged: false,
                    };
                    state.timelines.insert(timeline_id.clone(), timeline);
                    Some(StateChange::Timeline {
                        timeline_id,
                        change: TimelineChange::Reset { items },
                    })
                });
            }
            Event::TimelineChanged { timeline_id, diffs } => self.apply_diffs(timeline_id, diffs),
            Event::HistoryChanged {
                timeline_id,
                state: history,
            } => self.publish(|state| {
                state.timelines.get_mut(&timeline_id)?.history = history.clone();
                Some(StateChange::History {
                    timeline_id,
                    state: history,
                })
            }),
            Event::TimelineClosed { timeline_id } => {
                self.cancel_resync(&timeline_id);
                self.publish(|state| {
                    state.timelines.remove(&timeline_id)?;
                    Some(StateChange::Timeline {
                        timeline_id,
                        change: TimelineChange::Closed,
                    })
                });
            }
            Event::TypingChanged(typing) => self.publish(|state| {
                if typing.users.is_empty() {
                    state.typing.remove(&typing.room_id);
                } else {
                    state
                        .typing
                        .insert(typing.room_id.clone(), typing.users.clone());
                }
                Some(StateChange::Typing(typing))
            }),
            Event::MediaUploadProgress {
                room_id,
                transaction_id,
                uploaded_bytes,
                total_bytes,
                complete,
            } => self.publish(|state| {
                let upload = UploadState {
                    room_id,
                    uploaded_bytes,
                    total_bytes,
                    complete,
                    failure: None,
                };
                state.uploads.insert(transaction_id.clone(), upload.clone());
                Some(StateChange::Upload {
                    transaction_id,
                    upload,
                })
            }),
            Event::MediaUploadFailed {
                room_id,
                transaction_id,
                kind,
                will_retry,
            } => self.publish(|state| {
                let upload = state
                    .uploads
                    .entry(transaction_id.clone())
                    .or_insert_with(|| UploadState {
                        room_id,
                        uploaded_bytes: 0,
                        total_bytes: 0,
                        complete: false,
                        failure: None,
                    });
                upload.failure = Some(UploadFailure { kind, will_retry });
                Some(StateChange::Upload {
                    transaction_id,
                    upload: upload.clone(),
                })
            }),
            Event::MediaUploadFinished {
                room_id,
                transaction_id,
                outcome,
            } => self.publish(|state| {
                state.uploads.remove(&transaction_id);
                Some(StateChange::UploadFinished {
                    room_id,
                    transaction_id,
                    outcome,
                })
            }),
        }
    }

    fn apply_diffs(&mut self, timeline_id: TimelineId, diffs: Vec<TimelineDiff>) {
        let mut diverged = false;
        self.publish(|state| {
            let Some(timeline) = state.timelines.get_mut(&timeline_id) else {
                diverged = true;
                state.timelines.insert(
                    timeline_id.clone(),
                    TimelineState {
                        items: Vector::new(),
                        history: HistoryState::Idle,
                        diverged: true,
                    },
                );
                return Some(StateChange::Timeline {
                    timeline_id: timeline_id.clone(),
                    change: TimelineChange::Diverged,
                });
            };
            if timeline.diverged {
                return None;
            }
            let diffs: Arc<[TimelineDiff]> = diffs.into();
            let change = match TimelineDiff::try_apply_all(diffs.to_vec(), &mut timeline.items) {
                Ok(()) => TimelineChange::Changed {
                    diffs,
                    items: timeline.items.clone(),
                },
                Err(error) => {
                    tracing::debug!(?timeline_id, length = error.length, "timeline diverged");
                    timeline.diverged = true;
                    diverged = true;
                    TimelineChange::Diverged
                }
            };
            Some(StateChange::Timeline {
                timeline_id: timeline_id.clone(),
                change,
            })
        });
        if diverged {
            let task = spawn(resync(self.handle.clone(), timeline_id.clone()));
            self.resyncs.insert(timeline_id, task.abort_handle());
        }
    }

    fn refetch_rooms(&mut self) {
        let mut published = self.shared.lock();
        match published.rooms_fetch {
            RoomsFetch::Idle => {
                published.rooms_fetch = RoomsFetch::Running;
                drop(published);
                let task = spawn(fetch_rooms(self.handle.clone(), self.shared.clone()));
                self.rooms_fetch = Some(task.abort_handle());
            }
            RoomsFetch::Running => published.rooms_fetch = RoomsFetch::Again,
            RoomsFetch::Again => {}
        }
    }

    fn cancel_resync(&mut self, timeline_id: &TimelineId) {
        if let Some(task) = self.resyncs.remove(timeline_id) {
            task.abort();
        }
    }

    fn cancel_resyncs(&mut self) {
        for (_, task) in self.resyncs.drain() {
            task.abort();
        }
    }

    fn publish(&self, change: impl FnOnce(&mut State) -> Option<StateChange>) {
        self.shared.publish(change);
    }
}

impl Drop for Mirror {
    fn drop(&mut self) {
        self.cancel_resyncs();
        if let Some(task) = self.rooms_fetch.take() {
            task.abort();
        }
        self.shared.lock().updates = None;
    }
}

/// asks until the engine answers, since a busy room or a full queue is only ever for now.
/// a reset or a close for the timeline aborts it, and the engine sends a reset before it answers
async fn resync(handle: ClientHandle, timeline_id: TimelineId) {
    let mut backoff = RESYNC_BACKOFF;
    loop {
        let command = Command::Timeline(TimelineCommand::Resync {
            timeline_id: timeline_id.clone(),
        });
        match handle.execute(command).await {
            Ok(_) => return,
            Err(error) if error.kind == ErrorKind::Interrupted => return,
            Err(error) => {
                tracing::debug!(%error, ?timeline_id, "resync refused, trying again");
                matrix_sdk::sleep::sleep(backoff).await;
                backoff = (backoff * 2).min(RESYNC_BACKOFF_CAP);
            }
        }
    }
}

async fn fetch_rooms(handle: ClientHandle, shared: Arc<Shared>) {
    loop {
        let rooms: Vector<RoomSummary> = handle.rooms().await.into();
        shared.publish(|state| {
            state.rooms = rooms.clone();
            Some(StateChange::Rooms(rooms))
        });
        let mut published = shared.lock();
        if published.rooms_fetch != RoomsFetch::Again {
            published.rooms_fetch = RoomsFetch::Idle;
            return;
        }
        published.rooms_fetch = RoomsFetch::Running;
    }
}

/// cheap to clone, one per frontend
#[derive(Clone, Debug)]
pub struct MirrorHandle {
    shared: Arc<Shared>,
}

impl MirrorHandle {
    #[must_use]
    pub fn snapshot(&self) -> Arc<State> {
        self.shared.lock().state.clone()
    }

    /// opens with the current state, then every update after it. ends once the mirror stops
    #[must_use]
    pub fn subscribe(&self) -> Subscription {
        let (first, receiving) = self.shared.clone().subscribe();
        let rest = futures_util::stream::unfold(receiving, |(shared, receiver)| async move {
            let mut receiver = receiver?;
            match receiver.recv().await {
                Ok(update) => Some((update, (shared, Some(receiver)))),
                Err(broadcast::error::RecvError::Lagged(missed)) => {
                    tracing::debug!(missed, "mirror subscriber lagged, starting it over");
                    Some(shared.subscribe())
                }
                Err(broadcast::error::RecvError::Closed) => None,
            }
        });
        let updates = futures_util::stream::once(std::future::ready(first)).chain(rest);
        #[cfg(not(target_family = "wasm"))]
        let updates = updates.boxed();
        #[cfg(target_family = "wasm")]
        let updates = updates.boxed_local();
        Subscription(updates)
    }
}

pub struct Subscription(EngineBoxStream<Update>);

impl fmt::Debug for Subscription {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("Subscription")
            .finish_non_exhaustive()
    }
}

impl Stream for Subscription {
    type Item = Update;

    fn poll_next(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Update>> {
        self.0.poll_next_unpin(context)
    }
}

#[derive(Debug)]
struct Shared {
    published: Mutex<Published>,
}

#[derive(Debug)]
struct Published {
    state: Arc<State>,
    sequence: u64,
    /// gone once the mirror stops, which ends every subscription
    updates: Option<broadcast::Sender<Update>>,
    rooms_fetch: RoomsFetch,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum RoomsFetch {
    Idle,
    Running,
    Again,
}

impl Shared {
    fn lock(&self) -> MutexGuard<'_, Published> {
        self.published
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
    }

    /// state, sequence and send move together under the one lock, which is what lets
    /// `subscribe` take a snapshot and a receiver with nothing between them
    fn publish(&self, change: impl FnOnce(&mut State) -> Option<StateChange>) {
        let mut published = self.lock();
        let Some(change) = change(Arc::make_mut(&mut published.state)) else {
            return;
        };
        published.sequence += 1;
        let update = Update {
            sequence: published.sequence,
            change,
        };
        if let Some(updates) = &published.updates {
            let _ = updates.send(update);
        }
    }

    fn subscribe(self: Arc<Self>) -> (Update, (Arc<Self>, Option<broadcast::Receiver<Update>>)) {
        let published = self.lock();
        let update = Update {
            sequence: published.sequence,
            change: StateChange::State(published.state.clone()),
        };
        let receiver = published.updates.as_ref().map(broadcast::Sender::subscribe);
        drop(published);
        (update, (self, receiver))
    }
}

1	use std::{
2	collections::HashMap,
3	fmt,
4	pin::Pin,
5	sync::{Arc, Mutex, MutexGuard, PoisonError},
6	task::{Context, Poll},
7	time::Duration,
8	};
9
10	use futures_util::{Stream, StreamExt};
11	use imbl::Vector;
12	use matrix_sdk::{
13	SendOutsideWasm,
14	executor::{AbortHandle, spawn},
15	};
16	use ruma::{OwnedDeviceId, OwnedRoomId, OwnedTransactionId, OwnedUserId};
17	#[cfg(feature = "serde")]
18	use serde::{Deserialize, Serialize};
19	use tokio::sync::broadcast;
20
21	use crate::{
22	ClientHandle, Command, ConnectionState, EncryptionState, EngineFailure, ErrorKind, Event,
23	HistoryState, RoomSummary, TimelineCommand, TimelineDiff, TimelineId, TimelineItem,
24	TypingState, UploadOutcome, VerificationState, VerificationTarget,
25	engine::executor::EngineBoxFuture,
26	};
27
28	#[cfg(not(target_family = "wasm"))]
29	type EngineBoxStream<T> = futures_util::stream::BoxStream<'static, T>;
30	#[cfg(target_family = "wasm")]
31	type EngineBoxStream<T> = futures_util::stream::LocalBoxStream<'static, T>;
32
33	const DEFAULT_CAPACITY: usize = 256;
34	const RESYNC_BACKOFF: Duration = Duration::from_millis(100);
35	const RESYNC_BACKOFF_CAP: Duration = Duration::from_secs(10);
36
37	/// everything the engine has said so far, folded together
38	#[derive(Clone, Debug, PartialEq)]
39	#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
40	#[non_exhaustive]
41	pub struct State {
42	pub user_id: Option<OwnedUserId>,
43	pub session: SessionStatus,
44	pub connection: ConnectionState,
45	pub encryption: Option<EncryptionState>,
46	pub verification: Option<VerificationState>,
47	pub verification_target: Option<VerificationTarget>,
48	pub rooms: Vector<RoomSummary>,
49	pub timelines: imbl::HashMap<TimelineId, TimelineState>,
50	pub typing: imbl::HashMap<OwnedRoomId, Vec<OwnedUserId>>,
51	pub uploads: imbl::HashMap<OwnedTransactionId, UploadState>,
52	/// what stopped the engine, if something went wrong
53	pub engine_failure: Option<EngineFailure>,
54	/// the last save of the session record that failed, until one succeeds
55	pub session_save_failure: Option<String>,
56	}
57
58	#[derive(Clone, Debug, PartialEq)]
59	#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
60	#[non_exhaustive]
61	pub struct TimelineState {
62	pub items: Vector<TimelineItem>,
63	pub history: HistoryState,
64	/// a batch didn't fit, so `items` is the last good copy until the next reset
65	pub diverged: bool,
66	}
67
68	#[derive(Clone, Debug, Eq, PartialEq)]
69	#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
70	#[non_exhaustive]
71	pub struct UploadState {
72	pub room_id: OwnedRoomId,
73	pub uploaded_bytes: u64,
74	pub total_bytes: u64,
75	pub complete: bool,
76	pub failure: Option<UploadFailure>,
77	}
78
79	#[derive(Clone, Copy, Debug, Eq, PartialEq)]
80	#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
81	#[non_exhaustive]
82	pub struct UploadFailure {
83	pub kind: ErrorKind,
84	pub will_retry: bool,
85	}
86
87	/// the `Termination` it'll end in, as far as the events tell
88	#[derive(Clone, Debug, Eq, PartialEq)]
89	#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
90	#[non_exhaustive]
91	pub enum SessionStatus {
92	Active,
93	SoftLoggedOut {
94	device_id: OwnedDeviceId,
95	},
96	AccessRevoked,
97	LoggedOut,
98	/// stopped with the session still good
99	Stopped,
100	}
101
102	#[derive(Clone, Debug)]
103	#[non_exhaustive]
104	pub struct Update {
105	/// one more than the update before it, and a `State` carries the one it's current as of
106	pub sequence: u64,
107	pub change: StateChange,
108	}
109
110	#[derive(Clone, Debug)]
111	#[non_exhaustive]
112	pub enum StateChange {
113	/// start over from this. a subscription opens with one and gets another after lagging
114	State(Arc<State>),
115	Authenticated(OwnedUserId),
116	Session(SessionStatus),
117	/// the record on disk may now hold a spent refresh token
118	SessionSaveFailed(String),
119	/// saved again after `SessionSaveFailed`
120	SessionSaved,
121	Connection(ConnectionState),
122	Encryption(EncryptionState),
123	Verification(VerificationState),
124	VerificationTarget(VerificationTarget),
125	/// room modes arrive with the `Rooms` that follows
126	NotificationSettings,
127	Rooms(Vector<RoomSummary>),
128	Timeline {
129	timeline_id: TimelineId,
130	change: TimelineChange,
131	},
132	History {
133	timeline_id: TimelineId,
134	state: HistoryState,
135	},
136	Typing(TypingState),
137	Upload {
138	transaction_id: OwnedTransactionId,
139	upload: UploadState,
140	},
141	UploadFinished {
142	room_id: OwnedRoomId,
143	transaction_id: OwnedTransactionId,
144	outcome: UploadOutcome,
145	},
146	Failed(EngineFailure),
147	}
148
149	#[derive(Clone, Debug)]
150	#[non_exhaustive]
151	pub enum TimelineChange {
152	Reset {
153	items: Vector<TimelineItem>,
154	},
155	/// `items` is what applying `diffs` left
156	Changed {
157	diffs: Arc<[TimelineDiff]>,
158	items: Vector<TimelineItem>,
159	},
160	/// a batch didn't fit. a resync is on its way and nothing changes until its reset
161	Diverged,
162	Closed,
163	}
164
165	/// drains the event stream into a `State` that any number of `MirrorHandle`s watch
166	pub struct Mirror {
167	handle: ClientHandle,
168	events: EngineBoxStream<Event>,
169	shared: Arc<Shared>,
170	resyncs: HashMap<TimelineId, AbortHandle>,
171	rooms_fetch: Option<AbortHandle>,
172	}
173
174	impl fmt::Debug for Mirror {
175	fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
176	formatter.debug_struct("Mirror").finish_non_exhaustive()
177	}
178	}
179
180	impl Mirror {
181	/// takes any stream of the engine's events, so a host can still look at them on the way past
182	pub fn new(
183	handle: ClientHandle,
184	events: impl Stream<Item = Event> + SendOutsideWasm + 'static,
185	) -> (Self, MirrorHandle) {
186	Self::with_capacity(handle, events, DEFAULT_CAPACITY)
187	}
188
189	/// `capacity` updates can queue for a subscriber before it lags and starts over. a zero
190	/// capacity counts as one
191	pub fn with_capacity(
192	handle: ClientHandle,
193	events: impl Stream<Item = Event> + SendOutsideWasm + 'static,
194	capacity: usize,
195	) -> (Self, MirrorHandle) {
196	let state = State {
197	user_id: None,
198	session: SessionStatus::Active,
199	connection: *handle.connection_state().borrow(),
200	encryption: handle.encryption_state(),
201	verification: None,
202	verification_target: None,
203	rooms: Vector::new(),
204	timelines: imbl::HashMap::new(),
205	typing: imbl::HashMap::new(),
206	uploads: imbl::HashMap::new(),
207	engine_failure: None,
208	session_save_failure: None,
209	};
210	let shared = Arc::new(Shared {
211	published: Mutex::new(Published {
212	state: Arc::new(state),
213	sequence: 0,
214	updates: Some(broadcast::channel(capacity.max(1)).0),
215	rooms_fetch: RoomsFetch::Idle,
216	}),
217	});
218	#[cfg(not(target_family = "wasm"))]
219	let events = events.boxed();
220	#[cfg(target_family = "wasm")]
221	let events = events.boxed_local();
222	let mirror = Self {
223	handle,
224	events,
225	shared: shared.clone(),
226	resyncs: HashMap::new(),
227	rooms_fetch: None,
228	};
229	(mirror, MirrorHandle { shared })
230	}
231
232	/// resolves once the event stream ends, which is just after `Stopped`
233	pub fn run(mut self) -> EngineBoxFuture<'static, ()> {
234	Box::pin(async move {
235	self.refetch_rooms();
236	while let Some(event) = self.events.next().await {
237	self.apply(event);
238	}
239	})
240	}
241
242	fn apply(&mut self, event: Event) {
243	match event {
244	Event::ConnectionChanged(connection) => self.publish(\|state\| {
245	state.connection = connection;
246	Some(StateChange::Connection(connection))
247	}),
248	Event::Authenticated { user_id } => self.publish(\|state\| {
249	state.user_id = Some(user_id.clone());
250	Some(StateChange::Authenticated(user_id))
251	}),
252	Event::SessionSaveFailed { message } => self.publish(\|state\| {
253	state.session_save_failure = Some(message.clone());
254	Some(StateChange::SessionSaveFailed(message))
255	}),
256	Event::SessionTokensChanged => self.publish(\|state\| {
257	state.session_save_failure.take()?;
258	Some(StateChange::SessionSaved)
259	}),
260	Event::SessionAccessRejected { soft_logout } => self.publish(\|state\| {
261	state.session = match soft_logout {
262	Some(device_id) => SessionStatus::SoftLoggedOut { device_id },
263	None => SessionStatus::AccessRevoked,
264	};
265	Some(StateChange::Session(state.session.clone()))
266	}),
267	Event::LoggedOut => self.publish(\|state\| {
268	state.session = SessionStatus::LoggedOut;
269	Some(StateChange::Session(SessionStatus::LoggedOut))
270	}),
271	Event::Stopped => {
272	self.cancel_resyncs();
273	self.publish(\|state\| {
274	if state.session != SessionStatus::Active {
275	return None;
276	}
277	state.session = SessionStatus::Stopped;
278	Some(StateChange::Session(SessionStatus::Stopped))
279	});
280	}
281	Event::Failed(failure) => self.publish(\|state\| {
282	state.engine_failure = Some(failure.clone());
283	Some(StateChange::Failed(failure))
284	}),
285	Event::NotificationSettingsChanged => {
286	self.publish(\|_\| Some(StateChange::NotificationSettings));
287	}
288	Event::RoomsChanged => self.refetch_rooms(),
289	Event::EncryptionChanged(encryption) => self.publish(\|state\| {
290	state.encryption = Some(encryption.clone());
291	Some(StateChange::Encryption(encryption))
292	}),
293	Event::VerificationChanged(verification) => self.publish(\|state\| {
294	state.verification = Some(verification.clone());
295	Some(StateChange::Verification(verification))
296	}),
297	Event::VerificationTargetChanged(target) => self.publish(\|state\| {
298	state.verification_target = Some(target.clone());
299	Some(StateChange::VerificationTarget(target))
300	}),
301	Event::TimelineReset { timeline_id, items } => {
302	self.cancel_resync(&timeline_id);
303	self.publish(\|state\| {
304	let history = state
305	.timelines
306	.get(&timeline_id)
307	.map_or(HistoryState::Idle, \|timeline\| timeline.history.clone());
308	let timeline = TimelineState {
309	items: items.clone(),
310	history,
311	diverged: false,
312	};
313	state.timelines.insert(timeline_id.clone(), timeline);
314	Some(StateChange::Timeline {
315	timeline_id,
316	change: TimelineChange::Reset { items },
317	})
318	});
319	}
320	Event::TimelineChanged { timeline_id, diffs } => self.apply_diffs(timeline_id, diffs),
321	Event::HistoryChanged {
322	timeline_id,
323	state: history,
324	} => self.publish(\|state\| {
325	state.timelines.get_mut(&timeline_id)?.history = history.clone();
326	Some(StateChange::History {
327	timeline_id,
328	state: history,
329	})
330	}),
331	Event::TimelineClosed { timeline_id } => {
332	self.cancel_resync(&timeline_id);
333	self.publish(\|state\| {
334	state.timelines.remove(&timeline_id)?;
335	Some(StateChange::Timeline {
336	timeline_id,
337	change: TimelineChange::Closed,
338	})
339	});
340	}
341	Event::TypingChanged(typing) => self.publish(\|state\| {
342	if typing.users.is_empty() {
343	state.typing.remove(&typing.room_id);
344	} else {
345	state
346	.typing
347	.insert(typing.room_id.clone(), typing.users.clone());
348	}
349	Some(StateChange::Typing(typing))
350	}),
351	Event::MediaUploadProgress {
352	room_id,
353	transaction_id,
354	uploaded_bytes,
355	total_bytes,
356	complete,
357	} => self.publish(\|state\| {
358	let upload = UploadState {
359	room_id,
360	uploaded_bytes,
361	total_bytes,
362	complete,
363	failure: None,
364	};
365	state.uploads.insert(transaction_id.clone(), upload.clone());
366	Some(StateChange::Upload {
367	transaction_id,
368	upload,
369	})
370	}),
371	Event::MediaUploadFailed {
372	room_id,
373	transaction_id,
374	kind,
375	will_retry,
376	} => self.publish(\|state\| {
377	let upload = state
378	.uploads
379	.entry(transaction_id.clone())
380	.or_insert_with(\|\| UploadState {
381	room_id,
382	uploaded_bytes: 0,
383	total_bytes: 0,
384	complete: false,
385	failure: None,
386	});
387	upload.failure = Some(UploadFailure { kind, will_retry });
388	Some(StateChange::Upload {
389	transaction_id,
390	upload: upload.clone(),
391	})
392	}),
393	Event::MediaUploadFinished {
394	room_id,
395	transaction_id,
396	outcome,
397	} => self.publish(\|state\| {
398	state.uploads.remove(&transaction_id);
399	Some(StateChange::UploadFinished {
400	room_id,
401	transaction_id,
402	outcome,
403	})
404	}),
405	}
406	}
407
408	fn apply_diffs(&mut self, timeline_id: TimelineId, diffs: Vec<TimelineDiff>) {
409	let mut diverged = false;
410	self.publish(\|state\| {
411	let Some(timeline) = state.timelines.get_mut(&timeline_id) else {
412	diverged = true;
413	state.timelines.insert(
414	timeline_id.clone(),
415	TimelineState {
416	items: Vector::new(),
417	history: HistoryState::Idle,
418	diverged: true,
419	},
420	);
421	return Some(StateChange::Timeline {
422	timeline_id: timeline_id.clone(),
423	change: TimelineChange::Diverged,
424	});
425	};
426	if timeline.diverged {
427	return None;
428	}
429	let diffs: Arc<[TimelineDiff]> = diffs.into();
430	let change = match TimelineDiff::try_apply_all(diffs.to_vec(), &mut timeline.items) {
431	Ok(()) => TimelineChange::Changed {
432	diffs,
433	items: timeline.items.clone(),
434	},
435	Err(error) => {
436	tracing::debug!(?timeline_id, length = error.length, "timeline diverged");
437	timeline.diverged = true;
438	diverged = true;
439	TimelineChange::Diverged
440	}
441	};
442	Some(StateChange::Timeline {
443	timeline_id: timeline_id.clone(),
444	change,
445	})
446	});
447	if diverged {
448	let task = spawn(resync(self.handle.clone(), timeline_id.clone()));
449	self.resyncs.insert(timeline_id, task.abort_handle());
450	}
451	}
452
453	fn refetch_rooms(&mut self) {
454	let mut published = self.shared.lock();
455	match published.rooms_fetch {
456	RoomsFetch::Idle => {
457	published.rooms_fetch = RoomsFetch::Running;
458	drop(published);
459	let task = spawn(fetch_rooms(self.handle.clone(), self.shared.clone()));
460	self.rooms_fetch = Some(task.abort_handle());
461	}
462	RoomsFetch::Running => published.rooms_fetch = RoomsFetch::Again,
463	RoomsFetch::Again => {}
464	}
465	}
466
467	fn cancel_resync(&mut self, timeline_id: &TimelineId) {
468	if let Some(task) = self.resyncs.remove(timeline_id) {
469	task.abort();
470	}
471	}
472
473	fn cancel_resyncs(&mut self) {
474	for (_, task) in self.resyncs.drain() {
475	task.abort();
476	}
477	}
478
479	fn publish(&self, change: impl FnOnce(&mut State) -> Option<StateChange>) {
480	self.shared.publish(change);
481	}
482	}
483
484	impl Drop for Mirror {
485	fn drop(&mut self) {
486	self.cancel_resyncs();
487	if let Some(task) = self.rooms_fetch.take() {
488	task.abort();
489	}
490	self.shared.lock().updates = None;
491	}
492	}
493
494	/// asks until the engine answers, since a busy room or a full queue is only ever for now.
495	/// a reset or a close for the timeline aborts it, and the engine sends a reset before it answers
496	async fn resync(handle: ClientHandle, timeline_id: TimelineId) {
497	let mut backoff = RESYNC_BACKOFF;
498	loop {
499	let command = Command::Timeline(TimelineCommand::Resync {
500	timeline_id: timeline_id.clone(),
501	});
502	match handle.execute(command).await {
503	Ok(_) => return,
504	Err(error) if error.kind == ErrorKind::Interrupted => return,
505	Err(error) => {
506	tracing::debug!(%error, ?timeline_id, "resync refused, trying again");
507	matrix_sdk::sleep::sleep(backoff).await;
508	backoff = (backoff * 2).min(RESYNC_BACKOFF_CAP);
509	}
510	}
511	}
512	}
513
514	async fn fetch_rooms(handle: ClientHandle, shared: Arc<Shared>) {
515	loop {
516	let rooms: Vector<RoomSummary> = handle.rooms().await.into();
517	shared.publish(\|state\| {
518	state.rooms = rooms.clone();
519	Some(StateChange::Rooms(rooms))
520	});
521	let mut published = shared.lock();
522	if published.rooms_fetch != RoomsFetch::Again {
523	published.rooms_fetch = RoomsFetch::Idle;
524	return;
525	}
526	published.rooms_fetch = RoomsFetch::Running;
527	}
528	}
529
530	/// cheap to clone, one per frontend
531	#[derive(Clone, Debug)]
532	pub struct MirrorHandle {
533	shared: Arc<Shared>,
534	}
535
536	impl MirrorHandle {
537	#[must_use]
538	pub fn snapshot(&self) -> Arc<State> {
539	self.shared.lock().state.clone()
540	}
541
542	/// opens with the current state, then every update after it. ends once the mirror stops
543	#[must_use]
544	pub fn subscribe(&self) -> Subscription {
545	let (first, receiving) = self.shared.clone().subscribe();
546	let rest = futures_util::stream::unfold(receiving, \|(shared, receiver)\| async move {
547	let mut receiver = receiver?;
548	match receiver.recv().await {
549	Ok(update) => Some((update, (shared, Some(receiver)))),
550	Err(broadcast::error::RecvError::Lagged(missed)) => {
551	tracing::debug!(missed, "mirror subscriber lagged, starting it over");
552	Some(shared.subscribe())
553	}
554	Err(broadcast::error::RecvError::Closed) => None,
555	}
556	});
557	let updates = futures_util::stream::once(std::future::ready(first)).chain(rest);
558	#[cfg(not(target_family = "wasm"))]
559	let updates = updates.boxed();
560	#[cfg(target_family = "wasm")]
561	let updates = updates.boxed_local();
562	Subscription(updates)
563	}
564	}
565
566	pub struct Subscription(EngineBoxStream<Update>);
567
568	impl fmt::Debug for Subscription {
569	fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
570	formatter
571	.debug_struct("Subscription")
572	.finish_non_exhaustive()
573	}
574	}
575
576	impl Stream for Subscription {
577	type Item = Update;
578
579	fn poll_next(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Update>> {
580	self.0.poll_next_unpin(context)
581	}
582	}
583
584	#[derive(Debug)]
585	struct Shared {
586	published: Mutex<Published>,
587	}
588
589	#[derive(Debug)]
590	struct Published {
591	state: Arc<State>,
592	sequence: u64,
593	/// gone once the mirror stops, which ends every subscription
594	updates: Option<broadcast::Sender<Update>>,
595	rooms_fetch: RoomsFetch,
596	}
597
598	#[derive(Clone, Copy, Debug, Eq, PartialEq)]
599	enum RoomsFetch {
600	Idle,
601	Running,
602	Again,
603	}
604
605	impl Shared {
606	fn lock(&self) -> MutexGuard<'_, Published> {
607	self.published
608	.lock()
609	.unwrap_or_else(PoisonError::into_inner)
610	}
611
612	/// state, sequence and send move together under the one lock, which is what lets
613	/// `subscribe` take a snapshot and a receiver with nothing between them
614	fn publish(&self, change: impl FnOnce(&mut State) -> Option<StateChange>) {
615	let mut published = self.lock();
616	let Some(change) = change(Arc::make_mut(&mut published.state)) else {
617	return;
618	};
619	published.sequence += 1;
620	let update = Update {
621	sequence: published.sequence,
622	change,
623	};
624	if let Some(updates) = &published.updates {
625	let _ = updates.send(update);
626	}
627	}
628
629	fn subscribe(self: Arc<Self>) -> (Update, (Arc<Self>, Option<broadcast::Receiver<Update>>)) {
630	let published = self.lock();
631	let update = Update {
632	sequence: published.sequence,
633	change: StateChange::State(published.state.clone()),
634	};
635	let receiver = published.updates.as_ref().map(broadcast::Sender::subscribe);
636	drop(published);
637	(update, (self, receiver))
638	}
639	}