macltok

public

Star0

Watch0

16 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>,
}

#[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
    #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
    SessionSaveFailed(Arc<crate::AccountError>),
    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>,
    #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
    session_store: Option<Arc<dyn crate::SessionStore>>,
}

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
    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(),
        };
        let shared = Arc::new(Shared {
            published: Mutex::new(Published {
                state: Arc::new(state),
                sequence: 0,
                updates: Some(broadcast::channel(capacity).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,
            #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
            session_store: None,
        };
        (mirror, MirrorHandle { shared })
    }

    /// saves the record on every `SessionTokensChanged` before taking another event, and
    /// calls nothing but `save`. the store can't go while the client holds it open
    #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
    #[must_use]
    pub fn with_session_store(mut self, store: Arc<dyn crate::SessionStore>) -> Self {
        self.session_store = Some(store);
        self
    }

    /// 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).await;
            }
        })
    }

    async 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))
            }),
            #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
            Event::SessionTokensChanged => self.save_session().await,
            #[cfg(not(any(feature = "sqlite", feature = "indexeddb")))]
            Event::SessionTokensChanged => {}
            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(|_| 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 timeline = state.timelines.get_mut(&timeline_id)?;
            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());
        }
    }

    #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
    async fn save_session(&mut self) {
        let Some(store) = &self.session_store else {
            return;
        };
        let saved = match self.handle.session_record() {
            Ok(record) => store.save(&record).await,
            Err(error) => Err(error),
        };
        if let Err(error) = saved {
            tracing::warn!(%error, "saving the session record failed");
            self.publish(|_| Some(StateChange::SessionSaveFailed(Arc::new(error))));
        }
    }

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