macltok

public
Star0
Watch0
45 commits 3 branches 0 tags 0 releases3.0 MiB
use std::{
    collections::HashSet,
    fmt,
    future::Future,
    pin::Pin,
    sync::{Arc, Mutex},
    task::{Context, Poll},
};

use matrix_sdk::{Client, RoomMemberships};
use matrix_sdk_ui::{
    notification_client::{
        NotificationClient, NotificationProcessSetup, NotificationStatus, RawNotificationEvent,
    },
    spaces::{SpaceRoomList, SpaceService, room_list::SpaceRoomListPaginationState},
    sync_service::SyncService,
};
use ruma::{
    EventId, OwnedDeviceId, OwnedEventId, OwnedMxcUri, OwnedRoomId, OwnedRoomOrAliasId,
    OwnedServerName, OwnedTransactionId, OwnedUserId, RoomId, UserId,
    events::{
        AnyGlobalAccountDataEventContent, AnyRoomAccountDataEvent, AnyRoomAccountDataEventContent,
        GlobalAccountDataEventType, RoomAccountDataEventType,
    },
    serde::Raw,
};
use tokio::sync::{OnceCell, mpsc, oneshot, watch};

use crate::{
    AccountCommand, AttachmentUpload, Command, CommandOutput, ConnectionState, CreateRoomOptions,
    EncryptionCommand, EncryptionState, Error, ErrorKind, Event, EventRef, ImagePacks,
    InteractiveAuth, InteractiveAuthOutcome, LifecycleCommand, LinkPreview, Media, MediaCommand,
    MediaId, MediaLocation, MediaSize, MemberFilter, MemberSummary, MessageContent,
    MessageEditContent, NotificationCommand, OwnDevice, PowerLevelChanges, Profile,
    PushNotification, PushNotificationEvent, PushNotificationStatus, RoomCommand, RoomListFilter,
    RoomNotificationMode, RoomNotificationSettings, RoomPowerLevels, RoomSummary, SpaceChild,
    TimelineCommand, TimelineId, VerificationAction, VerificationTarget,
    encryption::encryption_state,
    engine::{MediaService, StoreLock},
    model::{PushNotificationRoom, PushNotificationSender},
    normalise::{image_packs, member_summary, room_summaries, space_child, url_previews_allowed},
};

#[derive(Clone)]
pub struct ClientHandle {
    requests: mpsc::Sender<Request>,
    client: Client,
    connection: watch::Receiver<ConnectionState>,
    notification_client: Arc<OnceCell<NotificationClient>>,
    sync_service: Arc<SyncService>,
    media: Arc<MediaService>,
    spaces: Arc<OnceCell<SpaceService>>,
    members_checked: Arc<Mutex<HashSet<OwnedRoomId>>>,
    _store_lock: Option<Arc<StoreLock>>,
}

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

/// a command's answer. drop it if you don't care, the command still runs
#[derive(Debug)]
#[must_use = "dropping it is fine but then you'll never know"]
pub struct Pending(oneshot::Receiver<Result<CommandOutput, Error>>);

impl Future for Pending {
    type Output = Result<CommandOutput, Error>;

    fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
        Pin::new(&mut self.0)
            .poll(context)
            .map(|result| result.unwrap_or_else(|_| Err(Error::interrupted())))
    }
}

impl ClientHandle {
    /// current credentials, it holds secrets. the engine saves it for you unless the
    /// session was built `without_persistence`
    #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
    pub fn session_record(&self) -> Result<crate::SessionRecord, crate::AccountError> {
        crate::account::session_record(&self.client)
    }

    /// the bearer token. it rotates on every refresh so ask again after a 401 rather than
    /// keeping it, and keep it in memory and away from all but the homeserver's media endpoints
    pub fn access_token(&self) -> Option<String> {
        self.client.access_token()
    }

    /// queue a command, waiting for room in the queue if there isn't any
    pub async fn send(&self, command: Command) -> Result<Pending, Error> {
        let (request, pending) = Request::new(command);
        self.requests
            .send(request)
            .await
            .map_err(|_| Error::interrupted())?;
        Ok(pending)
    }

    /// queue a command or fail with `QueueFull` right now
    pub fn try_send(&self, command: Command) -> Result<Pending, Error> {
        let (request, pending) = Request::new(command);
        self.requests
            .try_send(request)
            .map_err(|error| match error {
                mpsc::error::TrySendError::Full(_) => {
                    Error::new(ErrorKind::QueueFull, "the command queue is full")
                }
                mpsc::error::TrySendError::Closed(_) => Error::interrupted(),
            })?;
        Ok(pending)
    }

    /// queue a command and wait for its answer. timeline sends answer once the
    /// sdk send queue has them, delivery shows up later as the item's send state
    pub async fn execute(&self, command: Command) -> Result<CommandOutput, Error> {
        self.send(command).await?.await
    }

    async fn run(&self, command: Command) -> Result<(), Error> {
        self.execute(command).await.map(drop)
    }

    async fn run_for_room(&self, command: RoomCommand) -> Result<OwnedRoomId, Error> {
        match self.execute(Command::Room(command)).await? {
            CommandOutput::Room(room_id) => Ok(room_id),
            output => Err(Error::new(
                ErrorKind::Other,
                format!("expected a room but got {output:?}"),
            )),
        }
    }

    pub async fn shutdown(&self) -> Result<(), Error> {
        self.run(Command::Lifecycle(LifecycleCommand::Shutdown))
            .await
    }

    pub async fn logout(&self) -> Result<(), Error> {
        self.run(Command::Account(AccountCommand::Logout)).await
    }

    pub async fn delete_devices(
        &self,
        device_ids: Vec<OwnedDeviceId>,
        auth: Option<InteractiveAuth>,
    ) -> Result<InteractiveAuthOutcome, Error> {
        match self
            .run(Command::Account(AccountCommand::DeleteDevices {
                device_ids,
                auth,
            }))
            .await
        {
            Ok(()) => Ok(InteractiveAuthOutcome::Completed),
            Err(Error {
                challenge: Some(challenge),
                ..
            }) => Ok(InteractiveAuthOutcome::Challenge(*challenge)),
            Err(error) => Err(error),
        }
    }

    pub async fn rename_device(
        &self,
        device_id: OwnedDeviceId,
        display_name: impl Into<String>,
    ) -> Result<(), Error> {
        self.run(Command::Account(AccountCommand::RenameDevice {
            device_id,
            display_name: display_name.into(),
        }))
        .await
    }

    pub async fn set_global_account_data(
        &self,
        event_type: GlobalAccountDataEventType,
        content: Raw<AnyGlobalAccountDataEventContent>,
    ) -> Result<(), Error> {
        self.run(Command::Account(AccountCommand::SetAccountData {
            event_type,
            content,
        }))
        .await
    }

    pub async fn set_room_account_data(
        &self,
        room_id: OwnedRoomId,
        event_type: RoomAccountDataEventType,
        content: Raw<AnyRoomAccountDataEventContent>,
    ) -> Result<(), Error> {
        self.run(Command::Room(RoomCommand::SetAccountData {
            room_id,
            event_type,
            content,
        }))
        .await
    }

    pub async fn set_room_notification_mode(
        &self,
        room_id: OwnedRoomId,
        mode: RoomNotificationMode,
    ) -> Result<(), Error> {
        self.run(Command::Notification(NotificationCommand::SetRoomMode {
            room_id,
            mode,
        }))
        .await
    }

    pub async fn reset_room_notification_mode(&self, room_id: OwnedRoomId) -> Result<(), Error> {
        self.run(Command::Notification(NotificationCommand::ResetRoomMode(
            room_id,
        )))
        .await
    }

    pub async fn set_default_notification_mode(
        &self,
        encrypted: bool,
        one_to_one: bool,
        mode: RoomNotificationMode,
    ) -> Result<(), Error> {
        self.run(Command::Notification(NotificationCommand::SetDefaultMode {
            encrypted,
            one_to_one,
            mode,
        }))
        .await
    }

    pub async fn add_notification_keyword(&self, keyword: impl Into<String>) -> Result<(), Error> {
        self.run(Command::Notification(NotificationCommand::AddKeyword(
            keyword.into(),
        )))
        .await
    }

    pub async fn remove_notification_keyword(
        &self,
        keyword: impl Into<String>,
    ) -> Result<(), Error> {
        self.run(Command::Notification(NotificationCommand::RemoveKeyword(
            keyword.into(),
        )))
        .await
    }

    pub async fn open_room(&self, room_id: OwnedRoomId) -> Result<(), Error> {
        self.run(Command::Room(RoomCommand::Open(room_id))).await
    }

    pub async fn close_room(&self, room_id: OwnedRoomId) -> Result<(), Error> {
        self.run(Command::Room(RoomCommand::Close(room_id))).await
    }

    pub async fn open_thread(
        &self,
        room_id: OwnedRoomId,
        root_event_id: OwnedEventId,
    ) -> Result<(), Error> {
        self.run(Command::Timeline(TimelineCommand::OpenThread {
            room_id,
            root_event_id,
        }))
        .await
    }

    pub async fn close_thread(
        &self,
        room_id: OwnedRoomId,
        root_event_id: OwnedEventId,
    ) -> Result<(), Error> {
        self.run(Command::Timeline(TimelineCommand::CloseThread {
            room_id,
            root_event_id,
        }))
        .await
    }

    pub async fn resync(&self, timeline_id: TimelineId) -> Result<(), Error> {
        self.run(Command::Timeline(TimelineCommand::Resync { timeline_id }))
            .await
    }

    pub async fn paginate(&self, timeline_id: TimelineId, limit: u16) -> Result<(), Error> {
        self.run(Command::Timeline(TimelineCommand::Paginate {
            timeline_id,
            limit,
        }))
        .await
    }

    pub async fn send_message(
        &self,
        timeline_id: TimelineId,
        content: MessageContent,
    ) -> Result<(), Error> {
        self.run(Command::Timeline(TimelineCommand::Send {
            timeline_id,
            content,
        }))
        .await
    }

    pub async fn edit_message(
        &self,
        timeline_id: TimelineId,
        target: impl Into<EventRef>,
        content: MessageEditContent,
    ) -> Result<(), Error> {
        self.run(Command::Timeline(TimelineCommand::Edit {
            timeline_id,
            target: target.into(),
            content,
        }))
        .await
    }

    pub async fn redact(
        &self,
        timeline_id: TimelineId,
        target: impl Into<EventRef>,
        reason: Option<String>,
    ) -> Result<(), Error> {
        self.run(Command::Timeline(TimelineCommand::Redact {
            timeline_id,
            target: target.into(),
            reason,
        }))
        .await
    }

    pub async fn toggle_reaction(
        &self,
        timeline_id: TimelineId,
        target: impl Into<EventRef>,
        key: impl Into<String>,
        shortcode: Option<String>,
    ) -> Result<(), Error> {
        self.run(Command::Timeline(TimelineCommand::ToggleReaction {
            timeline_id,
            target: target.into(),
            key: key.into(),
            shortcode,
        }))
        .await
    }

    pub async fn mark_read(
        &self,
        timeline_id: TimelineId,
        event_id: OwnedEventId,
    ) -> Result<(), Error> {
        self.run(Command::Timeline(TimelineCommand::MarkRead {
            timeline_id,
            event_id,
        }))
        .await
    }

    pub async fn set_typing(&self, room_id: OwnedRoomId, typing: bool) -> Result<(), Error> {
        self.run(Command::Room(RoomCommand::SetTyping { room_id, typing }))
            .await
    }

    pub async fn prepare_to_send(&self, room_id: OwnedRoomId) -> Result<(), Error> {
        self.run(Command::Room(RoomCommand::PrepareToSend(room_id)))
            .await
    }

    pub async fn join_room(
        &self,
        room: OwnedRoomOrAliasId,
        via: Vec<OwnedServerName>,
    ) -> Result<OwnedRoomId, Error> {
        self.run_for_room(RoomCommand::Join { room, via }).await
    }

    pub async fn leave_room(&self, room_id: OwnedRoomId) -> Result<(), Error> {
        self.run(Command::Room(RoomCommand::Leave(room_id))).await
    }

    pub async fn forget_room(&self, room_id: OwnedRoomId) -> Result<(), Error> {
        self.run(Command::Room(RoomCommand::Forget(room_id))).await
    }

    pub async fn accept_invite(&self, room_id: OwnedRoomId) -> Result<(), Error> {
        self.run(Command::Room(RoomCommand::AcceptInvite(room_id)))
            .await
    }

    pub async fn reject_invite(&self, room_id: OwnedRoomId) -> Result<(), Error> {
        self.run(Command::Room(RoomCommand::RejectInvite(room_id)))
            .await
    }

    pub async fn invite_user(
        &self,
        room_id: OwnedRoomId,
        user_id: OwnedUserId,
    ) -> Result<(), Error> {
        self.run(Command::Room(RoomCommand::InviteUser { room_id, user_id }))
            .await
    }

    pub async fn kick_user(
        &self,
        room_id: OwnedRoomId,
        user_id: OwnedUserId,
        reason: Option<String>,
    ) -> Result<(), Error> {
        self.run(Command::Room(RoomCommand::KickUser {
            room_id,
            user_id,
            reason,
        }))
        .await
    }

    pub async fn ban_user(
        &self,
        room_id: OwnedRoomId,
        user_id: OwnedUserId,
        reason: Option<String>,
    ) -> Result<(), Error> {
        self.run(Command::Room(RoomCommand::BanUser {
            room_id,
            user_id,
            reason,
        }))
        .await
    }

    pub async fn unban_user(
        &self,
        room_id: OwnedRoomId,
        user_id: OwnedUserId,
        reason: Option<String>,
    ) -> Result<(), Error> {
        self.run(Command::Room(RoomCommand::UnbanUser {
            room_id,
            user_id,
            reason,
        }))
        .await
    }

    pub async fn set_power_levels(
        &self,
        room_id: OwnedRoomId,
        changes: PowerLevelChanges,
    ) -> Result<(), Error> {
        self.run(Command::Room(RoomCommand::SetPowerLevels {
            room_id,
            changes,
        }))
        .await
    }

    pub async fn set_user_power_level(
        &self,
        room_id: OwnedRoomId,
        user_id: OwnedUserId,
        level: i64,
    ) -> Result<(), Error> {
        self.run(Command::Room(RoomCommand::SetUserPowerLevel {
            room_id,
            user_id,
            level,
        }))
        .await
    }

    pub async fn create_room(&self, options: CreateRoomOptions) -> Result<OwnedRoomId, Error> {
        self.run_for_room(RoomCommand::Create(options)).await
    }

    /// hands back the existing dm with them if there is one
    pub async fn create_direct(&self, user_id: OwnedUserId) -> Result<OwnedRoomId, Error> {
        self.run_for_room(RoomCommand::CreateDirect(user_id)).await
    }

    pub async fn cancel_send(
        &self,
        room_id: OwnedRoomId,
        transaction_id: OwnedTransactionId,
    ) -> Result<(), Error> {
        self.run(Command::Room(RoomCommand::CancelSend {
            room_id,
            transaction_id,
        }))
        .await
    }

    pub async fn retry_send(
        &self,
        room_id: OwnedRoomId,
        transaction_id: OwnedTransactionId,
    ) -> Result<(), Error> {
        self.run(Command::Room(RoomCommand::RetrySend {
            room_id,
            transaction_id,
        }))
        .await
    }

    pub async fn upload(&self, upload: AttachmentUpload) -> Result<(), Error> {
        self.run(Command::Media(MediaCommand::Upload(upload))).await
    }

    pub async fn recover(&self, recovery_key: impl Into<String>) -> Result<(), Error> {
        self.run(Command::Encryption(EncryptionCommand::Recover(
            recovery_key.into(),
        )))
        .await
    }

    pub async fn create_key_backup(&self) -> Result<(), Error> {
        self.run(Command::Encryption(EncryptionCommand::CreateKeyBackup))
            .await
    }

    pub async fn enable_key_backup(&self) -> Result<(), Error> {
        self.run(Command::Encryption(EncryptionCommand::EnableKeyBackup))
            .await
    }

    pub async fn disable_key_backup(&self, delete_from_server: bool) -> Result<(), Error> {
        self.run(Command::Encryption(EncryptionCommand::DisableKeyBackup {
            delete_from_server,
        }))
        .await
    }

    pub async fn recover_key_backup(&self, recovery_key: impl Into<String>) -> Result<(), Error> {
        self.run(Command::Encryption(EncryptionCommand::RecoverKeyBackup(
            recovery_key.into(),
        )))
        .await
    }

    pub async fn start_verification(&self, target: VerificationTarget) -> Result<(), Error> {
        self.run(Command::Encryption(EncryptionCommand::StartVerification(
            target,
        )))
        .await
    }

    pub async fn verify(&self, action: VerificationAction) -> Result<(), Error> {
        self.run(Command::Encryption(EncryptionCommand::Verify(action)))
            .await
    }

    /// bytes for an avatar, or for media a retained timeline showed you. concurrent asks share
    /// one download
    pub async fn media(&self, media_id: &MediaId, size: MediaSize) -> Result<Media, Error> {
        self.media.timeline_media(media_id, size).await
    }

    /// where `media` would fetch from, for a host streaming big files itself. the byte cap
    /// doesn't apply since nothing is buffered
    pub async fn media_source(
        &self,
        media_id: &MediaId,
        size: MediaSize,
    ) -> Result<MediaLocation, Error> {
        self.media.media_location(media_id, size).await
    }

    /// the space graph builds on first use and then keeps itself current
    async fn spaces(&self) -> &SpaceService {
        self.spaces
            .get_or_init(|| SpaceService::new(self.client.clone()))
            .await
    }

    pub async fn rooms(&self) -> Vec<RoomSummary> {
        room_summaries(&self.client, self.spaces().await).await
    }

    pub async fn image_packs(&self) -> Result<ImagePacks, Error> {
        image_packs(&self.client).await
    }

    /// the homeserver's preview of a page, `None` when it has none. the homeserver sees every
    /// url asked about, so a host decides whether encrypted rooms ask at all
    pub async fn url_preview(&self, url: &str) -> Result<Option<LinkPreview>, Error> {
        self.media.url_preview(url).await
    }

    /// whether the room and the user's settings let link previews show here
    pub async fn url_previews_allowed(&self, room_id: &RoomId) -> Result<bool, Error> {
        url_previews_allowed(&self.room(room_id)?).await
    }

    /// browse a space the way the server sees it, rooms you haven't joined included
    pub async fn space_hierarchy(&self, space_id: OwnedRoomId) -> SpaceHierarchy {
        SpaceHierarchy(SpaceRoomList::new(self.client.clone(), space_id).await)
    }

    pub async fn own_profile(&self) -> Result<Profile, Error> {
        let user_id = self
            .client
            .user_id()
            .ok_or_else(|| Error::new(ErrorKind::Unauthenticated, "the client is not logged in"))?
            .to_owned();
        let response = self
            .client
            .account()
            .fetch_user_profile()
            .await
            .map_err(|source| Error::from_sdk("fetching the profile failed", source))?;
        profile(user_id, &response)
    }

    pub async fn user_profile(&self, user_id: &UserId) -> Result<Profile, Error> {
        let response = self
            .client
            .account()
            .fetch_user_profile_of(user_id)
            .await
            .map_err(|source| Error::from_sdk("fetching the profile failed", source))?;
        profile(user_id.to_owned(), &response)
    }

    pub async fn set_display_name(&self, name: Option<String>) -> Result<(), Error> {
        self.run(Command::Account(AccountCommand::SetDisplayName(name)))
            .await
    }

    pub async fn set_avatar(&self, avatar: Option<OwnedMxcUri>) -> Result<(), Error> {
        self.run(Command::Account(AccountCommand::SetAvatar(avatar)))
            .await
    }

    pub async fn upload_avatar(
        &self,
        mimetype: impl Into<String>,
        bytes: Vec<u8>,
    ) -> Result<(), Error> {
        self.run(Command::Account(AccountCommand::UploadAvatar {
            mimetype: mimetype.into(),
            bytes,
        }))
        .await
    }

    pub async fn rooms_filtered(&self, filter: &RoomListFilter) -> Vec<RoomSummary> {
        let matcher = filter.matcher();
        self.rooms()
            .await
            .into_iter()
            .filter(|room| matcher.matches(room))
            .collect()
    }

    pub async fn devices(&self) -> Result<Vec<OwnDevice>, Error> {
        let current_device_id = self.client.device_id();
        Ok(self
            .client
            .devices()
            .await
            .map_err(|source| Error::from_http("listing devices failed", source))?
            .devices
            .into_iter()
            .map(|device| OwnDevice {
                current: current_device_id == Some(device.device_id.as_ref()),
                device_id: device.device_id,
                display_name: device.display_name,
                last_seen_ip: device.last_seen_ip,
                last_seen_timestamp_ms: device.last_seen_ts.map(|timestamp| timestamp.get().into()),
            })
            .collect())
    }

    fn room(&self, room_id: &RoomId) -> Result<matrix_sdk::Room, Error> {
        self.client
            .get_room(room_id)
            .ok_or_else(Error::room_unknown)
    }

    pub async fn room_notification_settings(
        &self,
        room_id: &RoomId,
    ) -> Result<Option<RoomNotificationSettings>, Error> {
        let room = self.room(room_id)?;
        let custom = room.user_defined_notification_mode().await;
        let mode = match custom {
            Some(mode) => mode,
            None => match room.notification_mode().await {
                Some(mode) => mode,
                None => return Ok(None),
            },
        };
        Ok(Some(RoomNotificationSettings {
            mode: crate::normalise::notification_mode(mode),
            custom: custom.is_some(),
        }))
    }

    pub async fn notification_keywords(&self) -> Vec<String> {
        self.client
            .notification_settings()
            .await
            .enabled_keywords()
            .await
            .into_iter()
            .collect()
    }

    pub async fn resolve_push_notification(
        &self,
        room_id: &RoomId,
        event_id: &EventId,
    ) -> Result<PushNotificationStatus, Error> {
        let notification_failed = |source| {
            Error::with_source(
                ErrorKind::Other,
                "resolving the push notification failed",
                source,
            )
        };
        let client = self
            .notification_client
            .get_or_try_init(|| async {
                NotificationClient::new(
                    self.client.clone(),
                    NotificationProcessSetup::SingleProcess {
                        sync_service: self.sync_service.clone(),
                    },
                )
                .await
            })
            .await
            .map_err(notification_failed)?;
        let status = client
            .get_notification(room_id, event_id)
            .await
            .map_err(notification_failed)?;
        Ok(normalise_push_notification(room_id, event_id, status))
    }

    pub async fn members(
        &self,
        room_id: &RoomId,
        filter: MemberFilter,
    ) -> Result<Vec<MemberSummary>, Error> {
        let room = self.room(room_id)?;
        // stores have been seen with members_synced persisted over a partial
        // roster, and the sdk never asks /members again once it's set
        if self.members_checked.lock().unwrap().insert(room_id.to_owned()) {
            room.mark_members_missing();
        }
        list_members(&room, filter).await
    }

    /// drop the cached roster for a room and fetch /members again
    pub async fn refresh_members(
        &self,
        room_id: &RoomId,
        filter: MemberFilter,
    ) -> Result<Vec<MemberSummary>, Error> {
        let room = self.room(room_id)?;
        room.mark_members_missing();
        self.members_checked.lock().unwrap().insert(room_id.to_owned());
        list_members(&room, filter).await
    }

    pub async fn room_power_levels(&self, room_id: &RoomId) -> Result<RoomPowerLevels, Error> {
        let levels = self
            .room(room_id)?
            .power_levels()
            .await
            .map_err(|source| Error::from_sdk("reading power levels failed", source.into()))?;
        Ok(RoomPowerLevels {
            ban: levels.ban.into(),
            invite: levels.invite.into(),
            kick: levels.kick.into(),
            redact: levels.redact.into(),
            events_default: levels.events_default.into(),
            state_default: levels.state_default.into(),
            users_default: levels.users_default.into(),
            room_notification: levels.notifications.room.into(),
            users: levels
                .users
                .into_iter()
                .map(|(user_id, level)| (user_id, level.into()))
                .collect(),
            events: levels
                .events
                .into_iter()
                .map(|(event_type, level)| (event_type.to_string(), level.into()))
                .collect(),
            creators: levels
                .rules
                .privileged_creators
                .into_iter()
                .flatten()
                .collect(),
        })
    }

    pub async fn pending_send_ids(
        &self,
        room_id: &RoomId,
    ) -> Result<Vec<OwnedTransactionId>, Error> {
        let (echoes, _) = self
            .room(room_id)?
            .send_queue()
            .subscribe()
            .await
            .map_err(|source| {
                Error::with_source(ErrorKind::Store, "reading the send queue failed", source)
            })?;
        Ok(echoes
            .into_iter()
            .filter(|echo| {
                matches!(
                    echo.content,
                    matrix_sdk::send_queue::LocalEchoContent::Event { .. }
                )
            })
            .map(|echo| echo.transaction_id)
            .collect())
    }

    pub async fn global_account_data(
        &self,
        event_type: GlobalAccountDataEventType,
    ) -> Result<Option<Raw<AnyGlobalAccountDataEventContent>>, Error> {
        self.client
            .account()
            .account_data_raw(event_type)
            .await
            .map_err(|source| Error::from_sdk("reading account data failed", source))
    }

    pub async fn room_account_data(
        &self,
        room_id: &RoomId,
        event_type: RoomAccountDataEventType,
    ) -> Result<Option<Raw<AnyRoomAccountDataEvent>>, Error> {
        self.room(room_id)?
            .account_data(event_type)
            .await
            .map_err(|source| Error::from_sdk("reading room account data failed", source))
    }

    #[must_use]
    pub fn connection_state(&self) -> watch::Receiver<ConnectionState> {
        self.connection.clone()
    }

    #[must_use]
    pub fn encryption_state(&self) -> Option<EncryptionState> {
        let device_id = self.client.device_id()?.to_owned();
        let encryption = self.client.encryption();
        Some(encryption_state(
            device_id,
            encryption.recovery().state(),
            encryption.verification_state().get(),
            encryption.backups().state(),
        ))
    }
}

/// one space's hierarchy, a page at a time. hold on to it while you browse, it remembers where it got to
pub struct SpaceHierarchy(SpaceRoomList);

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

impl SpaceHierarchy {
    /// fetches the next page. a no-op once the end is reached or while a page is in flight
    pub async fn paginate(&self) -> Result<(), Error> {
        self.0.paginate().await.map_err(|source| {
            Error::with_source(
                ErrorKind::Server,
                "fetching the space hierarchy failed",
                source,
            )
        })
    }

    /// everything fetched so far, the space itself excluded
    pub async fn rooms(&self) -> Vec<SpaceChild> {
        self.0.rooms().await.into_iter().map(space_child).collect()
    }

    #[must_use]
    pub fn end_reached(&self) -> bool {
        matches!(
            self.0.pagination_state(),
            SpaceRoomListPaginationState::Idle { end_reached: true }
        )
    }
}

fn profile(
    user_id: OwnedUserId,
    response: &ruma::api::client::profile::get_profile::v3::Response,
) -> Result<Profile, Error> {
    use ruma::api::client::profile::{AvatarUrl, DisplayName};
    let malformed = |source| {
        Error::with_source(
            ErrorKind::Server,
            "the server sent a malformed profile",
            source,
        )
    };
    Ok(Profile {
        user_id,
        display_name: response.get_static::<DisplayName>().map_err(malformed)?,
        avatar: response
            .get_static::<AvatarUrl>()
            .map_err(malformed)?
            .as_deref()
            .map(MediaId::plain),
    })
}

#[derive(Debug)]
pub(crate) struct Request {
    pub(crate) command: Command,
    pub(crate) reply: Responder,
}

impl Request {
    fn new(command: Command) -> (Self, Pending) {
        let (reply, pending) = oneshot::channel();
        (
            Self {
                command,
                reply: Responder(reply),
            },
            Pending(pending),
        )
    }
}

/// whoever holds this owes the caller an answer. dropping it answers `Interrupted`
#[derive(Debug)]
pub(crate) struct Responder(oneshot::Sender<Result<CommandOutput, Error>>);

impl Responder {
    pub(crate) fn send(self, result: Result<CommandOutput, Error>) {
        if let Err(error) = &result {
            tracing::debug!(%error, kind = ?error.kind, "command failed");
        }
        let _ = self.0.send(result);
    }

    pub(crate) fn done(self, result: Result<(), Error>) {
        self.send(result.map(|()| CommandOutput::Done));
    }
}

#[derive(Debug)]
pub struct EventStream {
    events: mpsc::Receiver<Event>,
}

impl EventStream {
    pub async fn recv(&mut self) -> Option<Event> {
        self.events.recv().await
    }

    pub fn try_recv(&mut self) -> Result<Event, mpsc::error::TryRecvError> {
        self.events.try_recv()
    }
}

impl futures_util::Stream for EventStream {
    type Item = Event;

    fn poll_next(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        self.events.poll_recv(context)
    }
}

#[derive(Debug)]
pub(crate) struct DriverPort {
    requests: mpsc::Receiver<Request>,
    events: mpsc::Sender<Event>,
    connection: watch::Sender<ConnectionState>,
}

#[derive(Clone, Debug)]
pub(crate) struct EventSender(mpsc::Sender<Event>);

impl EventSender {
    pub(crate) async fn send(&self, event: Event) -> Result<(), SendEventError> {
        self.0.send(event).await.map_err(|_| SendEventError)
    }

    pub(crate) async fn reserve(&self) -> Result<mpsc::Permit<'_, Event>, SendEventError> {
        self.0.reserve().await.map_err(|_| SendEventError)
    }

    /// resolves once every `EventStream` is gone
    pub(crate) async fn closed(&self) {
        self.0.closed().await;
    }
}

impl DriverPort {
    pub(crate) fn event_sender(&self) -> EventSender {
        EventSender(self.events.clone())
    }

    pub(crate) fn connection_sender(&self) -> watch::Sender<ConnectionState> {
        self.connection.clone()
    }

    pub(crate) async fn recv(&mut self) -> Option<Request> {
        self.requests.recv().await
    }

    pub(crate) fn try_recv(&mut self) -> Option<Request> {
        self.requests.try_recv().ok()
    }

    pub(crate) fn close(&mut self) {
        self.requests.close();
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, thiserror::Error)]
#[error("the event receiver has closed")]
pub(crate) struct SendEventError;

#[must_use]
pub(crate) fn transport_pair(
    client: Client,
    sync_service: Arc<SyncService>,
    media: Arc<MediaService>,
    store_lock: Option<Arc<StoreLock>>,
    command_capacity: usize,
    event_capacity: usize,
) -> (ClientHandle, DriverPort, EventStream) {
    let (request_tx, request_rx) = mpsc::channel(command_capacity);
    let (event_tx, event_rx) = mpsc::channel(event_capacity);
    let (connection_tx, connection_rx) = watch::channel(ConnectionState::Idle);
    (
        ClientHandle {
            requests: request_tx,
            client,
            connection: connection_rx,
            notification_client: Arc::new(OnceCell::new()),
            sync_service,
            media,
            spaces: Arc::new(OnceCell::new()),
            members_checked: Arc::default(),
            _store_lock: store_lock,
        },
        DriverPort {
            requests: request_rx,
            events: event_tx,
            connection: connection_tx,
        },
        EventStream { events: event_rx },
    )
}

async fn list_members(
    room: &matrix_sdk::Room,
    filter: MemberFilter,
) -> Result<Vec<MemberSummary>, Error> {
    let memberships = match filter {
        MemberFilter::Active => RoomMemberships::ACTIVE,
        MemberFilter::Joined => RoomMemberships::JOIN,
        MemberFilter::Invited => RoomMemberships::INVITE,
        MemberFilter::Banned => RoomMemberships::BAN,
        MemberFilter::Left => RoomMemberships::LEAVE,
        // An empty mask is the SDK's unfiltered wildcard: it matches every
        // membership, including custom states that `all()` would exclude.
        MemberFilter::All => RoomMemberships::empty(),
    };
    Ok(room
        .members(memberships)
        .await
        .map_err(|source| Error::from_sdk("listing members failed", source))?
        .into_iter()
        .map(member_summary)
        .collect())
}

fn normalise_push_notification(
    room_id: &RoomId,
    event_id: &EventId,
    status: NotificationStatus,
) -> PushNotificationStatus {
    match status {
        NotificationStatus::Event(item) => {
            let event = match &item.raw_event {
                RawNotificationEvent::Timeline(event) => {
                    PushNotificationEvent::Timeline(event.clone())
                }
                RawNotificationEvent::Invite(event) => PushNotificationEvent::Invite(event.clone()),
            };
            PushNotificationStatus::Event(Box::new(PushNotification {
                event_id: event_id.to_owned(),
                event,
                sender: PushNotificationSender {
                    user_id: item.event.sender().to_owned(),
                    display_name: item.sender_display_name,
                    avatar_url: item.sender_avatar_url.as_deref().map(MediaId::plain),
                    name_ambiguous: item.is_sender_name_ambiguous,
                },
                room: PushNotificationRoom {
                    room_id: room_id.to_owned(),
                    name: item.room_computed_display_name,
                    avatar_url: item.room_avatar_url.as_deref().map(MediaId::plain),
                    canonical_alias: item
                        .room_canonical_alias
                        .as_deref()
                        .and_then(|alias| alias.parse().ok()),
                    topic: item.room_topic,
                    encrypted: item.is_room_encrypted,
                    direct: item.room_is_dm,
                    joined_members_count: item.joined_members_count,
                    service_members: item
                        .service_members
                        .into_iter()
                        .filter_map(|user_id| user_id.parse().ok())
                        .collect(),
                    active_service_members_count: item.active_service_members_count,
                    space: item.is_space,
                },
                noisy: item.is_noisy,
                has_mention: item.has_mention,
                thread_id: item.thread_id,
            }))
        }
        NotificationStatus::EventNotFound => PushNotificationStatus::NotFound,
        NotificationStatus::EventFilteredOut => PushNotificationStatus::FilteredOut,
        NotificationStatus::EventRedacted => PushNotificationStatus::Redacted,
    }
}