macltok

mod commands;

pub(crate) mod executor;

mod media;

mod mutations;

mod prepare;

mod room;

mod session;

mod timeline;

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 {}

use self::{

    commands::Job,

    executor::{EngineBoxFuture, TaskSet},

    mutations::{Mutation, MutationJob},

    room::{ActorNote, ActorRequest, RoomActor},

    timeline::{SessionContext, Supervisor},

};

use crate::{

    AccountCommand, ClientHandle, Command, ConnectionState, EncryptionCommand, EngineErrorKind,

    EngineFailure, Error, ErrorKind, Event, EventStream, LifecycleCommand, MediaCommand,

    NotificationCommand, RoomCommand, Termination, TimelineCommand,

    transport::{DriverPort, Request, Responder, transport_pair},

};

#[derive(Clone, Debug, Eq, PartialEq)]

#[non_exhaustive]

pub struct EngineConfig {

    pub command_capacity: usize,

    pub event_capacity: usize,

    pub initial_history_limit: u16,

    pub max_media_bytes: usize,

    pub max_media_sources: usize,

    /// warm a room up for sending when it's opened or someone starts typing in it

    pub prepare_to_send: bool,

}

impl EngineConfig {

    #[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_media_sources(mut self, sources: usize) -> Self {

        self.max_media_sources = sources;

        self

    }

    #[must_use]

    pub const fn with_prepare_to_send(mut self, prepare: bool) -> Self {

        self.prepare_to_send = prepare;

        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_media_sources: 2_048,

            prepare_to_send: true,

        }

    }

}

#[derive(Debug, ThisError)]

pub enum EngineError {

    #[error("the Matrix client is not authenticated")]

    Unauthenticated,

    #[error("command, event, and media source 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}"),

        }

    }

}

/// 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>>,

    session_changes: broadcast::Receiver<SessionChange>,

    persistence: Option<Persistence>,

    upload_reports: mpsc::Receiver<watchers::RoomUpload>,

}

/// 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>,

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

    queues: Option<Queues>,

    /// one progress forwarder per room that has uploaded, and whether it's subscribed yet

    uploads: HashMap<OwnedRoomId, watch::Receiver<bool>>,

    upload_reports: mpsc::Sender<watchers::RoomUpload>,

    /// 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>>,

}

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).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>,

    ) -> 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_media_sources == 0

        {

            return Err(EngineError::ZeroCapacity);

        }

        let sync = Arc::new(

            SyncService::builder(client.clone())

                .with_offline_mode()

                .build()

                .await

                .map_err(EngineError::SyncService)?,

        );

        let store_lock = store_lock.map(Arc::new);

        let media = Arc::new(MediaService::new(

            client.clone(),

            config.max_media_bytes,

            config.max_media_sources,

        ));

        let (handle, port, events) = transport_pair(

            client.clone(),

            sync.clone(),

            media.clone(),

            store_lock.clone(),

            config.command_capacity,

            config.event_capacity,

        );

        client.send_queue().enable_upload_progress(true);

        let (note_sender, notes) = mpsc::unbounded_channel();

        let (stop, stops) = mpsc::unbounded_channel();

        let (upload_reports, upload_report_queue) = mpsc::channel(config.event_capacity);

        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 engine = Self {

            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,

            queues: Some(Queues {

                mutations: mutation_queue,

                notifications: notification_queue,

                encryption: encryption_queue,

                session_changes,

                persistence,

                upload_reports: upload_report_queue,

            }),

            upload_reports,

            uploads: HashMap::new(),

            verification_handler: None,

            preparations: prepare::Preparations::default(),

            stop,

            stops,

            logouts: Arc::new(watch::Sender::new(Logouts::default())),

            _store_lock: store_lock,

        };

        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.resume_upload_forwarders().await;

        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");

        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 watchers: [(&'static str, EngineBoxFuture<'static, ()>); 12] = [

            (

                "upload tracker",

                Box::pin(watchers::uploads(

                    client.clone(),

                    client.send_queue().subscribe(),

                    queues.upload_reports,

                    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())),

            ),

            (

                "room list watcher",

                Box::pin(watchers::rooms(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,

                )),

            ),

        ];

        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> {

        let client = self.client.clone();

        match command {

            Command::Lifecycle(LifecycleCommand::Shutdown) => return Some(reply),

            Command::Account(AccountCommand::Logout) => {

                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 {

                    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::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::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 });

            }

        }

        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 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 mut ready = self.upload_forwarder(&room);

        let max_bytes = self.config.max_media_bytes;

        // queueing before the forwarder listens would lose the first progress ticks

        self.spawn_command("upload", reply, async move {

            ready

                .wait_for(|subscribed| *subscribed)

                .await

                .map_err(|_| Error::new(ErrorKind::Store, "reading the media send queue failed"))?;

            media::queue_upload(room, upload, max_bytes).await

        });

    }

    fn upload_forwarder(&mut self, room: &matrix_sdk::Room) -> watch::Receiver<bool> {

        self.uploads.retain(|_, ready| ready.has_changed().is_ok());

        if let Some(ready) = self.uploads.get(room.room_id()) {

            return ready.clone();

        }

        let (subscribed, ready) = watch::channel(false);

        self.tasks.spawn(

            TaskContext::Command("upload watcher"),

            watchers::room_uploads(room.clone(), self.upload_reports.clone(), subscribed),

        );

        self.uploads

            .insert(room.room_id().to_owned(), ready.clone());

        ready

    }

    /// the sdk resumes uploads left over from last time, and they deserve progress too

    async fn resume_upload_forwarders(&mut self) {

        let Ok(echoes) = self.client.send_queue().local_echoes().await else {

            return;

        };

        for room_id in echoes.keys() {

            if let Some(room) = self.client.get_room(room_id) {

                self.upload_forwarder(&room);

            }

        }

    }

    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"))),

        }

    }

    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) {