macltok

public
Star0
Watch0
139 commits 3 branches 0 tags 0 releases3.0 MiB
mod commands;
pub(crate) mod context;
pub(crate) mod executor;
mod media;
mod mutations;
mod packs;
mod prepare;
mod repair;
mod room;
mod session;
mod thread_list;
mod timeline;
mod url_preview;
mod watchers;

use std::{
    collections::{HashMap, VecDeque},
    future::Future,
    panic::AssertUnwindSafe,
    sync::Arc,
};

use futures_util::FutureExt;
use matrix_sdk::{Client, SendOutsideWasm, SessionChange};
use matrix_sdk_ui::sync_service::SyncService;
use ruma::{OwnedDeviceId, OwnedRoomId};
use thiserror::Error as ThisError;
use tokio::sync::{broadcast, mpsc, watch};

pub(crate) use media::MediaService;
pub(crate) use session::Persistence;

#[cfg(any(feature = "sqlite", feature = "indexeddb"))]
pub(crate) use crate::store_lock::StoreLock;
#[cfg(not(any(feature = "sqlite", feature = "indexeddb")))]
pub(crate) enum StoreLock {}

#[cfg(any(feature = "sqlite", feature = "indexeddb"))]
pub(crate) use crate::mentions::store::MentionsStore;
#[cfg(not(any(feature = "sqlite", feature = "indexeddb")))]
pub(crate) enum MentionsStore {}

#[cfg(any(feature = "sqlite", feature = "indexeddb"))]
pub(crate) use crate::search::store::SearchStore;
#[cfg(not(any(feature = "sqlite", feature = "indexeddb")))]
pub(crate) enum SearchStore {}

use self::{
    commands::Job,
    executor::{EngineBoxFuture, TaskSet},
    mutations::{Mutation, MutationJob},
    packs::PackEdit,
    room::{ActorNote, ActorRequest, RoomActor},
    timeline::{SessionContext, Supervisor},
};
use crate::{
    AccountCommand, ClientHandle, Command, CommandOutput, ConnectionState, EncryptionCommand,
    EngineErrorKind, EngineFailure, Error, ErrorKind, Event, EventStream, ImagePackSource,
    LifecycleCommand, MediaCommand, NotificationCommand, PackCommand, RoomCommand, Termination,
    ThreadListCommand, TimelineCommand,
    normalise::{ROOM_PACKS, ROOM_PREVIEW_URLS},
    transport::{DriverPort, Request, Responder, transport_pair},
};

#[derive(Clone, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub struct EngineConfig {
    pub command_capacity: usize,
    pub event_capacity: usize,
    /// what a thread paginates as it opens, and a room if its recent load fails. 0 for neither
    pub initial_history_limit: u16,
    pub max_media_bytes: usize,
    /// media one event context session keeps registered, least recently used dropped first
    pub max_context_media: usize,
    pub max_mentions_bytes: usize,
    pub max_mentions_entries: usize,
    pub max_mention_context_events: usize,
    pub max_mention_candidates: usize,
    /// warm a room up for sending when it's opened or someone starts typing in it
    pub prepare_to_send: bool,
    /// `/messages` pages each joined room gets per engine run for search backfill. 0 for none
    pub search_backfill_pages_per_room: u16,
    pub search_backfill_page_size: u16,
    /// the least time between two backfill requests, across every room
    pub search_backfill_interval_ms: u64,
    /// undecryptable messages kept to index once their keys turn up, newest first
    pub max_search_undecrypted: usize,
}

impl EngineConfig {
    #[must_use]
    pub const fn with_max_mentions_bytes(mut self, bytes: usize) -> Self {
        self.max_mentions_bytes = bytes;
        self
    }

    #[must_use]
    pub const fn with_max_mentions_entries(mut self, entries: usize) -> Self {
        self.max_mentions_entries = entries;
        self
    }

    #[must_use]
    pub const fn with_max_mention_context_events(mut self, events: usize) -> Self {
        self.max_mention_context_events = events;
        self
    }

    #[must_use]
    pub const fn with_max_mention_candidates(mut self, candidates: usize) -> Self {
        self.max_mention_candidates = candidates;
        self
    }

    #[must_use]
    pub const fn with_command_capacity(mut self, capacity: usize) -> Self {
        self.command_capacity = capacity;
        self
    }

    #[must_use]
    pub const fn with_event_capacity(mut self, capacity: usize) -> Self {
        self.event_capacity = capacity;
        self
    }

    #[must_use]
    pub const fn with_initial_history_limit(mut self, limit: u16) -> Self {
        self.initial_history_limit = limit;
        self
    }

    #[must_use]
    pub const fn with_max_media_bytes(mut self, bytes: usize) -> Self {
        self.max_media_bytes = bytes;
        self
    }

    #[must_use]
    pub const fn with_max_context_media(mut self, media: usize) -> Self {
        self.max_context_media = media;
        self
    }

    #[must_use]
    pub const fn with_prepare_to_send(mut self, prepare: bool) -> Self {
        self.prepare_to_send = prepare;
        self
    }

    #[must_use]
    pub const fn with_search_backfill_pages_per_room(mut self, pages: u16) -> Self {
        self.search_backfill_pages_per_room = pages;
        self
    }

    #[must_use]
    pub const fn with_search_backfill_page_size(mut self, events: u16) -> Self {
        self.search_backfill_page_size = events;
        self
    }

    #[must_use]
    pub const fn with_search_backfill_interval_ms(mut self, interval_ms: u64) -> Self {
        self.search_backfill_interval_ms = interval_ms;
        self
    }

    #[must_use]
    pub const fn with_max_search_undecrypted(mut self, events: usize) -> Self {
        self.max_search_undecrypted = events;
        self
    }
}

impl Default for EngineConfig {
    fn default() -> Self {
        Self {
            command_capacity: 64,
            event_capacity: 128,
            initial_history_limit: 30,
            max_media_bytes: 20 * 1024 * 1024,
            max_context_media: 1_024,
            max_mentions_bytes: 64 * 1024 * 1024,
            max_mentions_entries: 5_000,
            max_mention_context_events: 25_000,
            max_mention_candidates: 5_000,
            prepare_to_send: true,
            search_backfill_pages_per_room: 10,
            search_backfill_page_size: 100,
            search_backfill_interval_ms: 2_000,
            max_search_undecrypted: 2_000,
        }
    }
}

#[derive(Debug, ThisError)]
pub enum EngineError {
    #[error("the Matrix client is not authenticated")]
    Unauthenticated,
    #[error("command, event, media source, and mentions capacities must be nonzero")]
    ZeroCapacity,
    #[error("could not build the sync service")]
    SyncService(#[source] matrix_sdk_ui::sync_service::Error),
    #[error("the event receiver has closed")]
    EventReceiverClosed,
    #[error("engine task {task} panicked")]
    TaskPanicked { task: String },
}

impl EngineError {
    fn failure(&self) -> EngineFailure {
        EngineFailure {
            kind: match self {
                Self::EventReceiverClosed => EngineErrorKind::EventReceiverClosed,
                Self::TaskPanicked { .. } => EngineErrorKind::Watcher,
                Self::Unauthenticated | Self::ZeroCapacity | Self::SyncService(_) => {
                    EngineErrorKind::Other
                }
            },
            message: self.to_string(),
        }
    }
}

#[derive(Clone)]
enum TaskContext {
    /// a panic here costs the caller an `Interrupted` answer and nothing else
    Command(&'static str),
    /// a panic here takes the engine with it
    Watcher(&'static str),
    RoomActor {
        room_id: OwnedRoomId,
        generation: u64,
    },
}

impl TaskContext {
    fn label(&self) -> String {
        match self {
            Self::Command(name) | Self::Watcher(name) => (*name).to_owned(),
            Self::RoomActor {
                room_id,
                generation,
            } => format!("room actor {room_id} generation {generation}"),
        }
    }
}

/// tuwunel resends every room in range each time the room list grows, so the sdk's 100 a step
/// had a 399 room account hand most rooms to the event cache four or five times
const ROOM_LIST_GROWING_BATCH: u32 = 500;

/// how long shutdown waits on the mentions and search services before aborting them with the
/// other tasks. a sync response still being processed kept mentions busy for 2 to 3.4s, and the
/// next engine can't start until this one is gone
const SERVICE_STOP_GRACE: std::time::Duration = std::time::Duration::from_millis(500);

async fn stop_within(stop: impl Future<Output = ()>, service: &'static str) {
    if matrix_sdk::timeout::timeout(stop, SERVICE_STOP_GRACE)
        .await
        .is_err()
    {
        tracing::debug!(service, "a service didn't stop in time and is aborted");
    }
}

/// tasks catch their own panics so each comes back with its context, since joining
/// through the sdk's executor gives no task id to look one up by. wasm has nothing to catch
fn panic_message(payload: Box<dyn std::any::Any + Send>) -> String {
    payload
        .downcast_ref::<&str>()
        .map(ToString::to_string)
        .or_else(|| payload.downcast_ref::<String>().cloned())
        .unwrap_or_else(|| "task panicked with a non-string payload".to_owned())
}

struct TaskRegistry {
    tasks: TaskSet<(TaskContext, Option<String>)>,
}

impl TaskRegistry {
    fn spawn(
        &mut self,
        context: TaskContext,
        task: impl Future<Output = ()> + SendOutsideWasm + 'static,
    ) {
        self.tasks.spawn(async move {
            let panic = AssertUnwindSafe(task)
                .catch_unwind()
                .await
                .err()
                .map(panic_message);
            (context, panic)
        });
    }

    fn is_empty(&self) -> bool {
        self.tasks.is_empty()
    }

    async fn join_next(&mut self) -> Option<(TaskContext, Option<String>)> {
        loop {
            if let Ok(done) = self.tasks.join_next().await? {
                return Some(done);
            }
        }
    }

    async fn shutdown(&mut self) {
        self.tasks.shutdown().await;
    }
}

enum RoomSlot {
    Active {
        generation: u64,
        sender: mpsc::Sender<ActorRequest>,
    },
    /// opens and closes that arrive while the actor winds down, replayed in order once it's gone
    Closing {
        generation: u64,
        pending: VecDeque<Deferred>,
    },
}

impl RoomSlot {
    const fn generation(&self) -> u64 {
        match self {
            Self::Active { generation, .. } | Self::Closing { generation, .. } => *generation,
        }
    }
}

enum Deferred {
    Open(Responder),
    Close(Responder),
}

/// logouts in flight, and a token rejection held back because one of them caused it
#[derive(Default)]
pub(super) struct Logouts {
    pub(super) in_flight: usize,
    pub(super) rejected: Option<Option<OwnedDeviceId>>,
}

struct Queues {
    mutations: mpsc::Receiver<MutationJob>,
    notifications: mpsc::Receiver<Job<NotificationCommand>>,
    encryption: mpsc::Receiver<Job<EncryptionCommand>>,
    thread_lists: mpsc::Receiver<Job<ThreadListCommand>>,
    session_changes: broadcast::Receiver<SessionChange>,
    persistence: Option<Persistence>,
}

/// requests other devices send us, handed from the sdk's event handler to the encryption worker
type IncomingVerifications =
    mpsc::UnboundedSender<matrix_sdk::encryption::verification::VerificationRequest>;

pub struct Engine {
    client: Client,
    sync: Arc<SyncService>,
    port: DriverPort,
    config: EngineConfig,
    media: Arc<MediaService>,
    contexts: Arc<context::Sessions>,
    tasks: TaskRegistry,
    rooms: HashMap<OwnedRoomId, RoomSlot>,
    next_generation: u64,
    notes: mpsc::UnboundedReceiver<ActorNote>,
    note_sender: mpsc::UnboundedSender<ActorNote>,
    /// the rooms sliding sync should follow, applied off the run loop since it touches the store
    subscriptions: watch::Sender<Vec<OwnedRoomId>>,
    mutations: mpsc::Sender<MutationJob>,
    notifications: mpsc::Sender<Job<NotificationCommand>>,
    encryption: mpsc::Sender<Job<EncryptionCommand>>,
    thread_lists: mpsc::Sender<Job<ThreadListCommand>>,
    queues: Option<Queues>,
    /// the client can outlive us, so the handler has to come off it when we stop
    verification_handler: Option<matrix_sdk::event_handler::EventHandlerHandle>,
    stop: mpsc::UnboundedSender<Termination>,
    stops: mpsc::UnboundedReceiver<Termination>,
    logouts: Arc<watch::Sender<Logouts>>,
    preparations: prepare::Preparations,
    /// shared with every `ClientHandle`, so the store stays locked while the client is reachable
    _store_lock: Option<Arc<StoreLock>>,
    mentions: crate::mentions::service::Handle,
    mentions_service: Option<crate::mentions::service::Service>,
    mentions_updates: Option<watch::Receiver<crate::MentionsPage>>,
    search: crate::search::Handle,
    search_service: Option<crate::search::Service>,
}

impl Engine {
    /// the escape hatch for a client you built yourself. `Session` is the usual way in
    pub async fn from_client(
        client: Client,
        config: EngineConfig,
    ) -> Result<(Self, ClientHandle, EventStream), EngineError> {
        let session_changes = client.subscribe_to_session_changes();
        Self::with_persistence(client, config, session_changes, None, None, None, None).await
    }

    /// `session_changes` is taken when the client is built, since the sdk can refresh
    /// before an engine exists
    pub(crate) async fn with_persistence(
        client: Client,
        config: EngineConfig,
        session_changes: broadcast::Receiver<SessionChange>,
        persistence: Option<Persistence>,
        store_lock: Option<StoreLock>,
        mentions: Option<Result<MentionsStore, crate::mentions::service::Unopened>>,
        search: Option<Result<SearchStore, crate::Error>>,
    ) -> Result<(Self, ClientHandle, EventStream), EngineError> {
        if client.user_id().is_none() {
            return Err(EngineError::Unauthenticated);
        }
        if config.command_capacity == 0
            || config.event_capacity == 0
            || config.max_context_media == 0
            || config.max_mentions_entries == 0
            || config.max_mention_context_events == 0
            || config.max_mention_candidates == 0
        {
            return Err(EngineError::ZeroCapacity);
        }
        let store_lock = store_lock.map(Arc::new);
        let media = Arc::new(MediaService::new(
            client.clone(),
            config.max_media_bytes,
            config.max_context_media,
        ));
        let (mentions, mentions_service, mentions_updates) =
            crate::mentions::service::Service::new(
                client.clone(),
                media.clone(),
                mentions,
                config.clone(),
            )
            .await
            // a missing user id is all it still refuses, and that was turned away above
            .map_err(|_| EngineError::Unauthenticated)?;
        let sync = Arc::new(
            SyncService::builder(client.clone())
                .with_offline_mode()
                .with_room_list_extra_required_state(vec![
                    (ROOM_PACKS.into(), "*".to_owned()),
                    (ROOM_PREVIEW_URLS.into(), String::new()),
                ])
                .with_room_list_response_commit_hook(mentions.hook())
                .with_room_list_growing_batch_size(ROOM_LIST_GROWING_BATCH)
                .build()
                .await
                .map_err(EngineError::SyncService)?,
        );
        let (handle, port, events) = transport_pair(
            client.clone(),
            sync.clone(),
            media.clone(),
            store_lock.clone(),
            config.command_capacity,
            config.event_capacity,
        );
        let (search, search_service) = crate::search::Service::new(
            client.clone(),
            media.clone(),
            search,
            port.event_sender(),
            &config,
        );
        client.send_queue().enable_upload_progress(true);
        client.send_queue().enable_global_upload_progress(true);
        let (note_sender, notes) = mpsc::unbounded_channel();
        let (stop, stops) = mpsc::unbounded_channel();
        let (mutations, mutation_queue) = mpsc::channel(config.command_capacity);
        let (notifications, notification_queue) = mpsc::channel(config.command_capacity);
        let (encryption, encryption_queue) = mpsc::channel(config.command_capacity);
        let (thread_lists, thread_list_queue) = mpsc::channel(config.command_capacity);
        let engine = Self {
            contexts: Arc::new(context::Sessions::new(port.event_sender())),
            client,
            sync,
            port,
            config,
            media,
            tasks: TaskRegistry {
                tasks: TaskSet::new(),
            },
            rooms: HashMap::new(),
            next_generation: 0,
            notes,
            note_sender,
            subscriptions: watch::Sender::new(Vec::new()),
            mutations,
            notifications,
            encryption,
            thread_lists,
            queues: Some(Queues {
                mutations: mutation_queue,
                notifications: notification_queue,
                encryption: encryption_queue,
                thread_lists: thread_list_queue,
                session_changes,
                persistence,
            }),
            verification_handler: None,
            preparations: prepare::Preparations::default(),
            stop,
            stops,
            logouts: Arc::new(watch::Sender::new(Logouts::default())),
            _store_lock: store_lock,
            mentions,
            mentions_service: Some(mentions_service),
            mentions_updates: Some(mentions_updates),
            search,
            search_service: Some(search_service),
        };
        Ok((engine, handle, events))
    }

    /// boxed because the engine's future is deep enough to blow the compiler's query depth
    /// in every crate that spawns it, which isn't something a client should have to fix
    pub fn run(self) -> EngineBoxFuture<'static, Result<Termination, EngineError>> {
        Box::pin(self.run_inner())
    }

    async fn run_inner(mut self) -> Result<Termination, EngineError> {
        let events = self.port.event_sender();
        let user_id = self
            .client
            .user_id()
            .expect("authentication was checked during construction")
            .to_owned();
        self.port
            .connection_sender()
            .send_replace(ConnectionState::Connecting);
        for event in [
            Event::Authenticated { user_id },
            Event::ConnectionChanged(ConnectionState::Connecting),
        ] {
            if events.send(event).await.is_err() {
                self.shutdown().await;
                return Err(EngineError::EventReceiverClosed);
            }
        }
        self.start_tasks();
        self.sync.start().await;

        let mut shutdown_reply = None;
        let mut handles_dropped = false;
        let mut logouts = self.logouts.subscribe();
        let ending = loop {
            tokio::select! {
                biased;
                Some(termination) = self.stops.recv() => break Ok(termination),
                () = events.closed() => break Err(EngineError::EventReceiverClosed),
                () = settled(&mut logouts), if handles_dropped => {
                    break Ok(Termination::HandlesDropped);
                }
                Some((context, panic)) = self.tasks.join_next(), if !self.tasks.is_empty() => {
                    let Some(message) = panic else { continue };
                    if let TaskContext::Command(label) = context {
                        tracing::error!(label, panic = %message, "command task panicked");
                        continue;
                    }
                    break Err(EngineError::TaskPanicked { task: context.label() });
                }
                Some(note) = self.notes.recv() => self.actor_note(note).await,
                request = self.port.recv(), if !handles_dropped => {
                    // nobody's left to ask, but a logout they asked for still deserves to land
                    let Some(request) = request else {
                        handles_dropped = true;
                        continue;
                    };
                    if let Some(reply) = self.dispatch(request).await {
                        shutdown_reply = Some(reply);
                        break Ok(Termination::ShutDown);
                    }
                }
            }
        };

        if let Err(error) = &ending {
            let _ = events.send(Event::Failed(error.failure())).await;
        }
        self.shutdown().await;
        let _ = events.send(Event::Stopped).await;
        if let Some(reply) = shutdown_reply {
            reply.done(Ok(()));
        }
        // everything else still queued answers `Interrupted` as it drops, but a shutdown did happen
        while let Some(Request { command, reply }) = self.port.try_recv() {
            if matches!(command, Command::Lifecycle(LifecycleCommand::Shutdown)) {
                reply.done(Ok(()));
            }
        }
        ending
    }

    fn start_tasks(&mut self) {
        let queues = self.queues.take().expect("engine tasks start once");
        if let Some(service) = self.mentions_service.take() {
            self.tasks
                .spawn(TaskContext::Watcher("mentions service"), service.run());
        }
        if let Some(service) = self.search_service.take() {
            self.tasks
                .spawn(TaskContext::Watcher("search service"), service.run());
        }
        if let Some(updates) = self.mentions_updates.take() {
            let events = self.port.event_sender();
            self.tasks.spawn(
                TaskContext::Watcher("mentions updates"),
                crate::mentions::service::watch_updates(updates, events),
            );
        }
        let (incoming, incoming_verifications) = mpsc::unbounded_channel();
        self.verification_handler = Some(self.verification_handler(incoming));
        let client = &self.client;
        let events = || self.port.event_sender();
        let (notices, notice_queue) = mpsc::unbounded_channel();
        let (enabling_recovery, recovery_enabling) = watch::channel(false);
        let watchers: [(&'static str, EngineBoxFuture<'static, ()>); 15] = [
            (
                "upload tracker",
                Box::pin(watchers::uploads(
                    client.clone(),
                    client.send_queue().subscribe(),
                    events(),
                )),
            ),
            (
                "room subscription watcher",
                Box::pin(watchers::room_subscriptions(
                    self.sync.clone(),
                    self.subscriptions.subscribe(),
                )),
            ),
            (
                "session change watcher",
                Box::pin(session::session(
                    client.clone(),
                    queues.session_changes,
                    queues.persistence,
                    notices,
                    client
                        .device_id()
                        .expect("authentication was checked during construction")
                        .to_owned(),
                    self.logouts.clone(),
                )),
            ),
            (
                "session notice watcher",
                Box::pin(session::session_notices(
                    notice_queue,
                    events(),
                    self.stop.clone(),
                )),
            ),
            (
                "connection state watcher",
                Box::pin(watchers::connection(
                    self.sync.state(),
                    self.sync.clone(),
                    client.clone(),
                    events(),
                    self.port.connection_sender(),
                )),
            ),
            (
                "send queue watcher",
                Box::pin(watchers::send_queue_errors(
                    client.clone(),
                    client.send_queue().subscribe_errors(),
                    self.port.connection_sender().subscribe(),
                )),
            ),
            (
                "encryption state watcher",
                Box::pin(watchers::encryption(
                    client.clone(),
                    events(),
                    recovery_enabling,
                )),
            ),
            (
                "room list watcher",
                Box::pin(watchers::rooms(client.clone(), events())),
            ),
            (
                "event context watcher",
                Box::pin(context::close_on_leave(
                    client.clone(),
                    self.media.clone(),
                    self.contexts.clone(),
                )),
            ),
            (
                "image pack watcher",
                Box::pin(watchers::image_packs(client.clone(), events())),
            ),
            (
                "notification settings watcher",
                Box::pin(watchers::notification_settings(client.clone(), events())),
            ),
            (
                "room command worker",
                Box::pin(mutations::run_scheduler(
                    client.clone(),
                    queues.mutations,
                    self.config.command_capacity,
                )),
            ),
            (
                "notification command worker",
                Box::pin(commands::run_notifications(
                    client.clone(),
                    queues.notifications,
                )),
            ),
            (
                "encryption command worker",
                Box::pin(commands::run_encryption(
                    client.clone(),
                    events(),
                    queues.encryption,
                    incoming_verifications,
                    enabling_recovery,
                )),
            ),
            (
                "thread list worker",
                Box::pin(thread_list::run(
                    client.clone(),
                    events(),
                    self.media.clone(),
                    queues.thread_lists,
                )),
            ),
        ];
        for (label, task) in watchers {
            self.tasks.spawn(TaskContext::Watcher(label), task);
        }
    }

    fn verification_handler(
        &self,
        incoming: IncomingVerifications,
    ) -> matrix_sdk::event_handler::EventHandlerHandle {
        use matrix_sdk::ruma::events::key::verification::request::ToDeviceKeyVerificationRequestEvent;
        self.client.add_event_handler(
            move |event: ToDeviceKeyVerificationRequestEvent, client: Client| {
                let incoming = incoming.clone();
                async move {
                    let request = client
                        .encryption()
                        .get_verification_request(
                            &event.sender,
                            event.content.transaction_id.as_str(),
                        )
                        .await;
                    if let Some(request) = request {
                        let _ = incoming.send(request);
                    }
                }
            },
        )
    }

    /// hands back the shutdown request's responder, which is answered once we've stopped
    async fn dispatch(&mut self, Request { command, reply }: Request) -> Option<Responder> {
        tracing::debug!(command = %executor::label(&command), "engine command");
        let client = self.client.clone();
        match command {
            Command::Lifecycle(LifecycleCommand::Shutdown) => return Some(reply),
            Command::Account(AccountCommand::Logout { remove_pushers }) => {
                let events = self.port.event_sender();
                let stop = self.stop.clone();
                let logouts = self.logouts.clone();
                logouts.send_modify(|logouts| logouts.in_flight += 1);
                self.spawn_command("logout", reply, async move {
                    for pusher in remove_pushers {
                        if let Err(error) = crate::transport::remove_pusher(&client, pusher).await {
                            tracing::debug!(%error, "removing a pusher before logout failed");
                        }
                    }
                    let result = client.logout().await;
                    if result.is_ok() {
                        let _ = events.send(Event::LoggedOut).await;
                        let _ = stop.send(Termination::LoggedOut);
                    }
                    let mut rejected = None;
                    logouts.send_if_modified(|logouts| {
                        if result.is_err() && logouts.in_flight == 1 {
                            rejected = logouts.rejected.take();
                        }
                        false
                    });
                    if let Some(soft_logout) = rejected {
                        session::reject_session(&events, &stop, soft_logout).await;
                    }
                    logouts.send_modify(|logouts| logouts.in_flight -= 1);
                    result.map_err(|source| Error::from_sdk("logout failed", source))
                });
            }
            Command::Account(AccountCommand::SetAccountData {
                event_type,
                content,
            }) => enqueue(
                &self.mutations,
                Job {
                    command: Mutation::GlobalAccountData {
                        event_type,
                        content,
                    },
                    reply,
                },
            ),
            Command::Account(AccountCommand::SetDisplayName(name)) => {
                self.spawn_command("set display name", reply, async move {
                    client
                        .account()
                        .set_display_name(name.as_deref())
                        .await
                        .map_err(|source| Error::from_sdk("setting display name failed", source))
                });
            }
            Command::Account(AccountCommand::SetAvatar(avatar)) => {
                self.spawn_command("set avatar", reply, async move {
                    client
                        .account()
                        .set_avatar_url(avatar.as_deref())
                        .await
                        .map_err(|source| Error::from_sdk("setting avatar failed", source))
                });
            }
            Command::Account(AccountCommand::UploadAvatar { mimetype, bytes }) => {
                let max_bytes = self.config.max_media_bytes;
                self.spawn_command("upload avatar", reply, async move {
                    commands::upload_avatar(&client, &mimetype, bytes, max_bytes).await
                });
            }
            Command::Account(AccountCommand::UploadMedia { mimetype, bytes }) => {
                let max_bytes = self.config.max_media_bytes;
                self.tasks
                    .spawn(TaskContext::Command("upload media"), async move {
                        let uploaded =
                            commands::upload_media(&client, &mimetype, bytes, max_bytes).await;
                        reply.send(uploaded.map(CommandOutput::Mxc));
                    });
            }
            Command::Account(AccountCommand::IgnoreUser(user_id)) => enqueue(
                &self.mutations,
                Job {
                    command: Mutation::Ignore {
                        user_id,
                        ignored: true,
                    },
                    reply,
                },
            ),
            Command::Account(AccountCommand::UnignoreUser(user_id)) => enqueue(
                &self.mutations,
                Job {
                    command: Mutation::Ignore {
                        user_id,
                        ignored: false,
                    },
                    reply,
                },
            ),
            Command::Account(AccountCommand::ReorderSpaces(order)) => enqueue(
                &self.mutations,
                Job {
                    command: Mutation::ReorderSpaces(order),
                    reply,
                },
            ),
            Command::Account(AccountCommand::RenameDevice {
                device_id,
                display_name,
            }) => self.spawn_command("rename device", reply, async move {
                commands::rename_device(&client, &device_id, &display_name).await
            }),
            Command::Account(AccountCommand::DeleteDevices { device_ids, auth }) => {
                self.spawn_command("delete devices", reply, async move {
                    commands::delete_devices(&client, &device_ids, auth).await
                });
            }
            Command::Room(RoomCommand::Open(room_id)) => {
                if self.config.prepare_to_send {
                    self.prepare_quietly(room_id.clone());
                }
                self.open_room(room_id, reply).await;
            }
            Command::Room(RoomCommand::Close(room_id)) => self.close_room(room_id, reply).await,
            Command::Room(RoomCommand::PrepareToSend(room_id)) => {
                let flight = self.preparations.join(&client, room_id);
                self.spawn_command("prepare to send", reply, flight);
            }
            Command::Room(RoomCommand::SetTyping { room_id, typing }) => {
                if typing && self.config.prepare_to_send {
                    self.prepare_quietly(room_id.clone());
                }
                self.spawn_command("typing", reply, async move {
                    commands::set_typing(&client, &room_id, typing).await
                });
            }
            Command::Room(RoomCommand::CancelSend {
                room_id,
                transaction_id,
            }) => self.spawn_command("cancel send", reply, async move {
                commands::cancel_send(&client, &room_id, &transaction_id).await
            }),
            Command::Room(RoomCommand::RetrySend {
                room_id,
                transaction_id,
            }) => self.spawn_command("retry send", reply, async move {
                commands::retry_send(&client, &room_id, &transaction_id).await
            }),
            Command::Room(command) => enqueue(
                &self.mutations,
                Job {
                    command: Mutation::Room(command),
                    reply,
                },
            ),
            Command::Timeline(command) => self.route_timeline(command, reply),
            Command::EventContext(command) => {
                let (client, media, contexts) =
                    (client.clone(), self.media.clone(), self.contexts.clone());
                self.tasks
                    .spawn(TaskContext::Command("event context"), async move {
                        reply.send(context::execute(client, media, contexts, command).await);
                    });
            }
            Command::Mentions(command) => {
                let mentions = self.mentions.clone();
                self.tasks
                    .spawn(TaskContext::Command("mentions command"), async move {
                        reply.send(mentions.execute(command).await);
                    });
            }
            Command::Search(command) => {
                let search = self.search.clone();
                self.tasks
                    .spawn(TaskContext::Command("search command"), async move {
                        reply.send(search.execute(command).await);
                    });
            }
            Command::HistoryRepair(command) => self.route_repair(command, reply),
            Command::Media(MediaCommand::Upload(upload)) => self.upload(upload, reply),
            Command::Encryption(command) => enqueue(&self.encryption, Job { command, reply }),
            Command::Notification(command) => {
                enqueue(&self.notifications, Job { command, reply });
            }
            Command::Pack(command) => self.pack(command, reply),
            Command::ThreadList(command) => enqueue(&self.thread_lists, Job { command, reply }),
        }
        None
    }

    fn spawn_command(
        &mut self,
        label: &'static str,
        reply: Responder,
        work: impl Future<Output = Result<(), Error>> + SendOutsideWasm + 'static,
    ) {
        self.tasks.spawn(TaskContext::Command(label), async move {
            reply.done(work.await);
        });
    }

    fn prepare_quietly(&mut self, room_id: OwnedRoomId) {
        let flight = self.preparations.join(&self.client, room_id);
        self.tasks
            .spawn(TaskContext::Command("prepare to send"), async move {
                let _ = flight.await;
            });
    }

    fn pack(&mut self, command: PackCommand, reply: Responder) {
        let edit = |pack, edit| Mutation::Pack { pack, edit };
        let mutation = match command {
            PackCommand::CreateRoomPack {
                room_id,
                state_key,
                display_name,
            } => edit(
                ImagePackSource::Room { room_id, state_key },
                PackEdit::Create { display_name },
            ),
            PackCommand::Delete(pack) => edit(pack, PackEdit::Delete),
            PackCommand::SetMeta {
                pack,
                display_name,
                avatar,
                usage,
            } => edit(
                pack,
                PackEdit::Meta {
                    display_name,
                    avatar,
                    usage,
                },
            ),
            PackCommand::PutImage {
                pack,
                shortcode,
                mxc,
                body,
                usage,
                info,
            } => edit(
                pack,
                PackEdit::Put {
                    shortcode,
                    mxc,
                    body,
                    usage,
                    info,
                },
            ),
            PackCommand::RenameImage { pack, from, to } => {
                edit(pack, PackEdit::Rename { from, to })
            }
            PackCommand::RemoveImage { pack, shortcode } => {
                edit(pack, PackEdit::Remove { shortcode })
            }
            PackCommand::SetEnabledEverywhere {
                room_id,
                state_key,
                enabled,
            } => Mutation::PackEnabledEverywhere {
                room_id,
                state_key,
                enabled,
            },
            command => {
                let client = self.client.clone();
                let mutations = self.mutations.clone();
                let max_bytes = self.config.max_media_bytes;
                self.tasks.spawn(TaskContext::Command("pack"), async move {
                    reply.send(packs::run_detached(client, mutations, max_bytes, command).await);
                });
                return;
            }
        };
        enqueue(
            &self.mutations,
            Job {
                command: mutation,
                reply,
            },
        );
    }

    fn upload(&mut self, upload: crate::AttachmentUpload, reply: Responder) {
        let Some(room) = self.client.get_room(upload.timeline_id.room_id()) else {
            return reply.done(Err(Error::room_unknown()));
        };
        let max_bytes = self.config.max_media_bytes;
        self.spawn_command("upload", reply, async move {
            media::queue_upload(room, upload, max_bytes).await
        });
    }

    fn route_timeline(&mut self, command: TimelineCommand, reply: Responder) {
        let sender = match self.rooms.get(command.room_id()) {
            Some(RoomSlot::Active { sender, .. }) => Some(sender.clone()),
            _ => None,
        };
        match (sender, command) {
            (Some(sender), command) => send_actor(&sender, ActorRequest::Execute(command, reply)),
            (
                None,
                TimelineCommand::MarkRead {
                    timeline_id,
                    event_id,
                },
            ) => {
                let client = self.client.clone();
                self.spawn_command("mark read", reply, async move {
                    commands::mark_read(&client, &timeline_id, event_id).await
                });
            }
            (None, _) => reply.done(Err(Error::new(ErrorKind::RoomNotOpen, "room is not open"))),
        }
    }

    fn route_repair(&mut self, command: crate::HistoryRepairCommand, reply: Responder) {
        use crate::HistoryRepairCommand;
        let room_id = match &command {
            HistoryRepairCommand::Cancel { room_id }
            | HistoryRepairCommand::RetryRecent { room_id }
            | HistoryRepairCommand::FillGap { room_id, .. } => room_id,
        };
        match self.rooms.get(room_id) {
            Some(RoomSlot::Active { sender, .. }) => {
                send_actor(sender, ActorRequest::Repair(command, reply))
            }
            Some(RoomSlot::Closing { .. }) => {
                reply.done(Err(Error::new(ErrorKind::RoomBusy, "room is closing")))
            }
            None => reply.done(Err(Error::new(ErrorKind::RoomNotOpen, "room is not open"))),
        }
    }

    async fn open_room(&mut self, room_id: OwnedRoomId, reply: Responder) {
        match self.rooms.get_mut(&room_id) {
            Some(RoomSlot::Active { sender, .. }) => send_actor(sender, ActorRequest::Open(reply)),
            Some(RoomSlot::Closing { pending, .. }) => {
                defer(pending, Deferred::Open(reply), self.config.command_capacity);
            }
            None => self.start_actor(room_id, reply).await,
        }
    }

    async fn close_room(&mut self, room_id: OwnedRoomId, reply: Responder) {
        let (generation, sender) = match self.rooms.get_mut(&room_id) {
            None => return reply.done(Ok(())),
            Some(RoomSlot::Closing { pending, .. }) => {
                return defer(
                    pending,
                    Deferred::Close(reply),
                    self.config.command_capacity,
                );
            }
            Some(RoomSlot::Active { generation, sender }) => (*generation, sender.clone()),
        };
        match sender.try_send(ActorRequest::Close(reply)) {
            Ok(()) => {}
            // the actor already left on its own, which is closed enough
            Err(mpsc::error::TrySendError::Closed(request)) => request.into_reply().done(Ok(())),
            Err(mpsc::error::TrySendError::Full(request)) => {
                return request.into_reply().done(Err(Error::new(
                    ErrorKind::RoomBusy,
                    "room command queue is full",
                )));
            }
        }
        self.rooms.insert(
            room_id,
            RoomSlot::Closing {
                generation,
                pending: VecDeque::new(),
            },
        );
        self.update_room_subscriptions();
    }

    async fn start_actor(&mut self, room_id: OwnedRoomId, reply: Responder) {
        let Some(room) = self.client.get_room(&room_id) else {
            return reply.done(Err(Error::room_unknown()));
        };
        self.next_generation = self.next_generation.wrapping_add(1);
        let generation = self.next_generation;
        let (sender, requests) = mpsc::channel(self.config.command_capacity);
        let (supervisor, panics) = Supervisor::new();
        let actor = RoomActor {
            room_id: room_id.clone(),
            generation,
            room,
            requests,
            notes: self.note_sender.clone(),
            context: SessionContext {
                events: self.port.event_sender(),
                media: self.media.clone(),
                current_user_id: self.client.user_id().map(ToOwned::to_owned),
                initial_history_limit: self.config.initial_history_limit,
                supervisor,
            },
            panics,
            opening: vec![reply],
        };
        self.rooms
            .insert(room_id.clone(), RoomSlot::Active { generation, sender });
        self.tasks.spawn(
            TaskContext::RoomActor {
                room_id,
                generation,
            },
            actor.run(),
        );
        self.update_room_subscriptions();
    }

    async fn actor_note(&mut self, note: ActorNote) {
        let (room_id, generation) = match &note {
            ActorNote::Leaving {
                room_id,
                generation,
            }
            | ActorNote::Exited {
                room_id,
                generation,
                ..
            } => (room_id.clone(), *generation),
        };
        if self
            .rooms
            .get(&room_id)
            .is_none_or(|slot| slot.generation() != generation)
        {
            return;
        }
        let ActorNote::Exited { leftovers, .. } = note else {
            if let Some(slot @ RoomSlot::Active { .. }) = self.rooms.get_mut(&room_id) {
                *slot = RoomSlot::Closing {
                    generation,
                    pending: VecDeque::new(),
                };
                self.update_room_subscriptions();
            }
            return;
        };
        let pending = match self.rooms.remove(&room_id) {
            Some(RoomSlot::Closing { pending, .. }) => pending,
            _ => VecDeque::new(),
        };
        // leftovers were sent before anything deferred, so they go first
        for request in leftovers {
            match request {
                ActorRequest::Open(reply) => self.open_room(room_id.clone(), reply).await,
                ActorRequest::Close(reply) => reply.done(Ok(())),
                ActorRequest::Execute(command, reply) => self.route_timeline(command, reply),
                ActorRequest::Repair(command, reply) => self.route_repair(command, reply),
            }
        }
        for deferred in pending {
            match deferred {
                Deferred::Open(reply) => self.open_room(room_id.clone(), reply).await,
                Deferred::Close(reply) => self.close_room(room_id.clone(), reply).await,
            }
        }
        self.update_room_subscriptions();
    }

    fn update_room_subscriptions(&self) {
        let room_ids = self
            .rooms
            .iter()
            .filter(|(_, slot)| matches!(slot, RoomSlot::Active { .. }))
            .map(|(room_id, _)| room_id.clone())
            .collect();
        self.subscriptions.send_replace(room_ids);
    }

    async fn shutdown(&mut self) {
        if let Some(handler) = self.verification_handler.take() {
            self.client.remove_event_handler(handler);
        }
        self.port.close();
        tracing::debug!(step = "sync", "engine shutdown step");
        self.sync.stop().await;
        if self.mentions_service.is_none() {
            tracing::debug!(step = "mentions", "engine shutdown step");
            stop_within(self.mentions.stop(), "mentions").await;
        }
        if self.search_service.is_none() {
            tracing::debug!(step = "search", "engine shutdown step");
            stop_within(self.search.stop(), "search").await;
        }
        tracing::debug!(step = "tasks", "engine shutdown step");
        self.tasks.shutdown().await;
        self.rooms.clear();
        tracing::debug!(step = "contexts", "engine shutdown step");
        self.contexts.clear().await;
        tracing::debug!(step = "media", "engine shutdown step");
        self.media.clear().await;
        tracing::debug!(step = "done", "engine shutdown step");
        self.port
            .connection_sender()
            .send_replace(ConnectionState::Stopped);
        let _ = self
            .port
            .event_sender()
            .send(Event::ConnectionChanged(ConnectionState::Stopped))
            .await;
    }
}

fn send_actor(sender: &mpsc::Sender<ActorRequest>, request: ActorRequest) {
    if let Err(error) = sender.try_send(request) {
        let (request, error) = match error {
            mpsc::error::TrySendError::Full(request) => (
                request,
                Error::new(ErrorKind::RoomBusy, "room command queue is full"),
            ),
            mpsc::error::TrySendError::Closed(request) => (
                request,
                Error::new(ErrorKind::RoomNotOpen, "room is not open"),
            ),
        };
        request.into_reply().done(Err(error));
    }
}

fn defer(pending: &mut VecDeque<Deferred>, deferred: Deferred, capacity: usize) {
    if pending.len() < capacity {
        return pending.push_back(deferred);
    }
    let (Deferred::Open(reply) | Deferred::Close(reply)) = deferred;
    reply.done(Err(Error::new(
        ErrorKind::RoomBusy,
        "room lifecycle queue is full",
    )));
}

async fn settled(logouts: &mut watch::Receiver<Logouts>) {
    let _ = logouts.wait_for(|logouts| logouts.in_flight == 0).await;
}

fn enqueue<T>(sender: &mpsc::Sender<Job<T>>, job: Job<T>) {
    if let Err(error) = sender.try_send(job) {
        let (job, error) = match error {
            mpsc::error::TrySendError::Full(job) => (
                job,
                Error::new(ErrorKind::QueueFull, "command queue is full"),
            ),
            mpsc::error::TrySendError::Closed(job) => (job, Error::interrupted()),
        };
        job.reply.done(Err(error));
    }
}