macltok

public
Star0
Watch0
110 commits 3 branches 0 tags 0 releases3.0 MiB
use std::{
    collections::HashSet,
    fmt,
    future::Future,
    pin::Pin,
    sync::{Arc, Mutex, Weak},
    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, EventContextCommand,
    EventContextPage, EventContextSession, EventRef, HistoryRepairCommand, HttpPusher,
    ImagePackSource, ImagePacks, InteractiveAuth, InteractiveAuthOutcome, LifecycleCommand,
    LinkPreview, Media, MediaCommand, MediaId, MediaLocation, MediaSize, MemberFilter,
    MemberSummary, MentionKey, MentionsBoundary, MentionsCommand, MentionsPage, MessageContent,
    MessageEditContent, NotificationCommand, OwnDevice, PackCommand, PackImageInfo, PackUsage,
    Patch, PowerLevelChanges, Profile, PushFormat, PushNotification, PushNotificationEvent,
    PushNotificationStatus, PusherKey, RecoveryKey, RoomCommand, RoomListFilter,
    RoomNotificationMode, RoomNotificationSettings, RoomPowerLevels, RoomSummary,
    SavedMentionContext, SearchCommand, SearchCoverage, SearchPage, SearchScope, SearchSession,
    SpaceChild, ThreadListCommand, 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>>>,
    publication: Arc<PublicationRegistry>,
    _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 {
    pub(crate) fn register_publication_mirror(&self) -> PublicationRegistration {
        self.publication.register_mirror()
    }

    /// 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, remove_pushers: Vec<PusherKey>) -> Result<(), Error> {
        self.run(Command::Account(AccountCommand::Logout { remove_pushers }))
            .await
    }

    /// replaces any pusher with the same app id and pushkey
    pub async fn set_pusher(&self, pusher: HttpPusher) -> Result<(), Error> {
        use ruma::{
            api::client::push::{PusherIds, PusherInit, PusherKind},
            push::{HttpPusherData, PushFormat as SdkFormat},
        };
        let mut data = HttpPusherData::new(pusher.url);
        data.format = match pusher.format {
            PushFormat::Full => None,
            PushFormat::EventIdOnly => Some(SdkFormat::EventIdOnly),
        };
        data.data = pusher.data;
        let pusher = PusherInit {
            ids: PusherIds::new(pusher.pushkey, pusher.app_id),
            kind: PusherKind::Http(data),
            app_display_name: pusher.app_display_name,
            device_display_name: pusher.device_display_name,
            profile_tag: None,
            lang: pusher.lang,
        };
        self.client
            .pusher()
            .set(pusher.into(), false)
            .await
            .map_err(|source| Error::from_sdk("setting the pusher failed", source))
    }

    pub async fn remove_pusher(&self, pusher: PusherKey) -> Result<(), Error> {
        remove_pusher(&self.client, pusher).await
    }

    /// the account's http pushers, every device's alike
    pub async fn pushers(&self) -> Result<Vec<HttpPusher>, Error> {
        use ruma::{
            api::client::push::{PusherKind, get_pushers},
            push::PushFormat as SdkFormat,
        };
        let response = self
            .client
            .send(get_pushers::v3::Request::new())
            .await
            .map_err(|source| Error::from_sdk("fetching pushers failed", source.into()))?;
        Ok(response
            .pushers
            .into_iter()
            .filter_map(|pusher| match pusher.kind {
                PusherKind::Http(data) => Some(HttpPusher {
                    app_id: pusher.ids.app_id,
                    pushkey: pusher.ids.pushkey,
                    app_display_name: pusher.app_display_name,
                    device_display_name: pusher.device_display_name,
                    lang: pusher.lang,
                    url: data.url,
                    format: match data.format {
                        Some(SdkFormat::EventIdOnly) => PushFormat::EventIdOnly,
                        _ => PushFormat::Full,
                    },
                    data: data.data,
                }),
                _ => None,
            })
            .collect())
    }

    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
    }

    /// the room's pins as `TimelineId::Pinned`, kept current as they change. the room has to be open
    pub async fn open_pinned(&self, room_id: OwnedRoomId) -> Result<(), Error> {
        self.run(Command::Timeline(TimelineCommand::OpenPinned(room_id)))
            .await
    }

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

    /// the room's threads, newest activity first, kept current as they gain replies.
    /// the room needn't be open. it starts empty, `paginate_thread_list` fills it
    pub async fn open_thread_list(&self, room_id: OwnedRoomId) -> Result<(), Error> {
        self.run(Command::ThreadList(ThreadListCommand::Open(room_id)))
            .await
    }

    pub async fn paginate_thread_list(
        &self,
        room_id: OwnedRoomId,
        limit: u16,
    ) -> Result<(), Error> {
        self.run(Command::ThreadList(ThreadListCommand::Paginate {
            room_id,
            limit,
        }))
        .await
    }

    pub async fn close_thread_list(&self, room_id: OwnedRoomId) -> Result<(), Error> {
        self.run(Command::ThreadList(ThreadListCommand::Close(room_id)))
            .await
    }

    /// reads the server's list so a pin made elsewhere just before isn't dropped
    pub async fn set_pinned(
        &self,
        room_id: OwnedRoomId,
        event_id: OwnedEventId,
        pinned: bool,
    ) -> Result<(), Error> {
        self.run(Command::Room(RoomCommand::SetPinned {
            room_id,
            event_id,
            pinned,
        }))
        .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 cancel_history_repair(&self, room_id: OwnedRoomId) -> Result<(), Error> {
        self.run(Command::HistoryRepair(HistoryRepairCommand::Cancel {
            room_id,
        }))
        .await
    }

    pub async fn fill_history_gap(
        &self,
        room_id: OwnedRoomId,
        gap_id: ruma::OwnedTransactionId,
    ) -> Result<(), Error> {
        self.run(Command::HistoryRepair(HistoryRepairCommand::FillGap {
            room_id,
            gap_id,
        }))
        .await
    }

    pub async fn retry_recent_history(&self, room_id: OwnedRoomId) -> Result<(), Error> {
        self.run(Command::HistoryRepair(HistoryRepairCommand::RetryRecent {
            room_id,
        }))
        .await
    }

    /// close the session when the view goes, or its media stay registered
    pub async fn event_context(
        &self,
        room_id: &RoomId,
        event_id: &EventId,
        before_limit: u16,
        after_limit: u16,
    ) -> Result<EventContextSession, Error> {
        match self
            .execute(Command::EventContext(EventContextCommand::Fetch {
                room_id: room_id.to_owned(),
                event_id: event_id.to_owned(),
                before_limit,
                after_limit,
            }))
            .await?
        {
            CommandOutput::EventContext(session) => Ok(session),
            output => Err(unexpected("event context", &output)),
        }
    }

    pub async fn event_context_before(
        &self,
        context_id: &str,
        token: &str,
        limit: u16,
    ) -> Result<EventContextPage, Error> {
        self.event_context_page(EventContextCommand::Before {
            context_id: context_id.to_owned(),
            limit,
            token: token.to_owned(),
        })
        .await
    }

    pub async fn event_context_after(
        &self,
        context_id: &str,
        token: &str,
        limit: u16,
    ) -> Result<EventContextPage, Error> {
        self.event_context_page(EventContextCommand::After {
            context_id: context_id.to_owned(),
            limit,
            token: token.to_owned(),
        })
        .await
    }

    pub async fn close_event_context(&self, context_id: &str) -> Result<(), Error> {
        self.run(Command::EventContext(EventContextCommand::Close {
            context_id: context_id.to_owned(),
        }))
        .await
    }

    async fn event_context_page(
        &self,
        command: EventContextCommand,
    ) -> Result<EventContextPage, Error> {
        match self.execute(Command::EventContext(command)).await? {
            CommandOutput::EventContextPage(page) => Ok(page),
            output => Err(unexpected("event context page", &output)),
        }
    }

    pub async fn mentions_page(
        &self,
        cursor: Option<String>,
        limit: u32,
    ) -> Result<MentionsPage, Error> {
        match self
            .execute(Command::Mentions(MentionsCommand::Page {
                cursor,
                limit: limit.min(50) as u16,
            }))
            .await?
        {
            CommandOutput::MentionsPage(page) => Ok(page),
            output => Err(unexpected("mentions page", &output)),
        }
    }

    pub async fn mention_context(
        &self,
        room_id: &RoomId,
        event_id: &EventId,
    ) -> Result<SavedMentionContext, Error> {
        self.mention_context_command(MentionsCommand::Context(MentionKey {
            room_id: room_id.to_owned(),
            event_id: event_id.to_owned(),
        }))
        .await
    }

    pub async fn retry_mention_context(
        &self,
        room_id: &RoomId,
        event_id: &EventId,
    ) -> Result<SavedMentionContext, Error> {
        self.mention_context_command(MentionsCommand::RetryContext(MentionKey {
            room_id: room_id.to_owned(),
            event_id: event_id.to_owned(),
        }))
        .await
    }

    async fn mention_context_command(
        &self,
        command: MentionsCommand,
    ) -> Result<SavedMentionContext, Error> {
        match self.execute(Command::Mentions(command)).await? {
            CommandOutput::MentionContext(context) => Ok(context),
            output => Err(unexpected("mention context", &output)),
        }
    }

    pub async fn set_mentions_read(&self, keys: Vec<MentionKey>, read: bool) -> Result<(), Error> {
        self.run(Command::Mentions(MentionsCommand::SetRead { keys, read }))
            .await
    }

    pub async fn mark_mentions_read_through(
        &self,
        boundary: MentionsBoundary,
    ) -> Result<(), Error> {
        self.run(Command::Mentions(MentionsCommand::ReadThrough(boundary)))
            .await
    }

    pub async fn recover_mentions(&self) -> Result<(), Error> {
        self.run(Command::Mentions(MentionsCommand::Recover)).await
    }

    /// newest first, at most 50 a page. close the search when its results go, or its media
    /// stay registered
    pub async fn search_messages(
        &self,
        query: &str,
        scope: SearchScope,
        limit: u16,
    ) -> Result<SearchSession, Error> {
        match self
            .execute(Command::Search(SearchCommand::Search {
                query: query.to_owned(),
                scope,
                limit,
            }))
            .await?
        {
            CommandOutput::SearchSession(session) => Ok(session),
            output => Err(unexpected("search session", &output)),
        }
    }

    pub async fn search_more(
        &self,
        search_id: &str,
        cursor: &str,
        limit: u16,
    ) -> Result<SearchPage, Error> {
        match self
            .execute(Command::Search(SearchCommand::More {
                search_id: search_id.to_owned(),
                cursor: cursor.to_owned(),
                limit,
            }))
            .await?
        {
            CommandOutput::SearchPage(page) => Ok(page),
            output => Err(unexpected("search page", &output)),
        }
    }

    pub async fn close_search(&self, search_id: &str) -> Result<(), Error> {
        self.run(Command::Search(SearchCommand::Close {
            search_id: search_id.to_owned(),
        }))
        .await
    }

    pub async fn search_coverage(&self, scope: SearchScope) -> Result<SearchCoverage, Error> {
        match self
            .execute(Command::Search(SearchCommand::Coverage(scope)))
            .await?
        {
            CommandOutput::SearchCoverage(coverage) => Ok(coverage),
            output => Err(unexpected("search coverage", &output)),
        }
    }

    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
    }

    /// the key is in the answer and nowhere else, so show it before dropping it
    pub async fn enable_recovery(&self, passphrase: Option<String>) -> Result<RecoveryKey, Error> {
        self.recovery_key(EncryptionCommand::EnableRecovery { passphrase })
            .await
    }

    pub async fn reset_recovery_key(
        &self,
        passphrase: Option<String>,
    ) -> Result<RecoveryKey, Error> {
        self.recovery_key(EncryptionCommand::ResetRecoveryKey { passphrase })
            .await
    }

    async fn recovery_key(&self, command: EncryptionCommand) -> Result<RecoveryKey, Error> {
        match self.execute(Command::Encryption(command)).await? {
            CommandOutput::RecoveryKey(key) => Ok(key),
            output => Err(unexpected("recovery key", &output)),
        }
    }

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

    /// checks a typed key or passphrase against secret storage without importing anything
    pub async fn verify_recovery_key(&self, recovery_key: &str) -> Result<bool, Error> {
        use matrix_sdk::encryption::secret_storage::SecretStorageError;
        match self
            .client
            .encryption()
            .secret_storage()
            .open_secret_store(recovery_key)
            .await
        {
            Ok(_) => Ok(true),
            Err(SecretStorageError::SecretStorageKey(_)) => Ok(false),
            Err(SecretStorageError::MissingKeyInfo { .. }) => {
                Err(Error::new(ErrorKind::Invalid, "recovery is not set up"))
            }
            Err(source) => Err(Error::from_recovery(
                "checking the recovery key failed",
                source.into(),
            )),
        }
    }

    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
    }

    /// unencrypted, for room avatars and anything else that only needs the uri
    pub async fn upload_media(
        &self,
        mimetype: impl Into<String>,
        bytes: Vec<u8>,
    ) -> Result<OwnedMxcUri, Error> {
        let command = Command::Account(AccountCommand::UploadMedia {
            mimetype: mimetype.into(),
            bytes,
        });
        match self.execute(command).await? {
            CommandOutput::Mxc(uri) => Ok(uri),
            output => Err(Error::new(
                ErrorKind::Other,
                format!("expected an mxc but got {output:?}"),
            )),
        }
    }

    pub async fn ignore_user(&self, user_id: OwnedUserId) -> Result<(), Error> {
        self.run(Command::Account(AccountCommand::IgnoreUser(user_id)))
            .await
    }

    pub async fn unignore_user(&self, user_id: OwnedUserId) -> Result<(), Error> {
        self.run(Command::Account(AccountCommand::UnignoreUser(user_id)))
            .await
    }

    /// writes msc3230 orders so these spaces sort in exactly this sequence, touching as few as it can
    pub async fn reorder_spaces(&self, order: Vec<OwnedRoomId>) -> Result<(), Error> {
        self.run(Command::Account(AccountCommand::ReorderSpaces(order)))
            .await
    }

    /// adding m.favourite drops m.lowpriority, the way Element does
    pub async fn set_favourite(&self, room_id: OwnedRoomId, favourite: bool) -> Result<(), Error> {
        self.run(Command::Room(RoomCommand::SetFavourite {
            room_id,
            favourite,
        }))
        .await
    }

    /// None goes back to your global name
    pub async fn set_room_display_name(
        &self,
        room_id: OwnedRoomId,
        name: Option<String>,
    ) -> Result<(), Error> {
        self.run(Command::Room(RoomCommand::SetOwnDisplayName {
            room_id,
            name,
        }))
        .await
    }

    /// None goes back to your global avatar
    pub async fn set_room_avatar(
        &self,
        room_id: OwnedRoomId,
        avatar: Option<OwnedMxcUri>,
    ) -> Result<(), Error> {
        self.run(Command::Room(RoomCommand::SetOwnAvatar { room_id, avatar }))
            .await
    }

    /// muting drops them below what sending a message takes, unmuting puts a muted user back
    /// at users_default and leaves anyone else alone. `ErrorKind::Forbidden` if you can't
    pub async fn mute_user(
        &self,
        room_id: OwnedRoomId,
        user_id: OwnedUserId,
        muted: bool,
    ) -> Result<(), Error> {
        self.run(Command::Room(RoomCommand::MuteUser {
            room_id,
            user_id,
            muted,
        }))
        .await
    }

    /// body defaults to the shortcode
    pub async fn send_sticker(
        &self,
        timeline_id: TimelineId,
        pack_id: String,
        shortcode: String,
    ) -> Result<(), Error> {
        self.run(Command::Pack(PackCommand::SendSticker {
            timeline_id,
            pack_id,
            shortcode,
        }))
        .await
    }

    /// fails if the state key already holds a pack
    pub async fn create_room_pack(
        &self,
        room_id: OwnedRoomId,
        state_key: String,
        display_name: String,
    ) -> Result<(), Error> {
        self.run(Command::Pack(PackCommand::CreateRoomPack {
            room_id,
            state_key,
            display_name,
        }))
        .await
    }

    pub async fn delete_pack(&self, pack: ImagePackSource) -> Result<(), Error> {
        self.run(Command::Pack(PackCommand::Delete(pack))).await
    }

    pub async fn set_pack_meta(
        &self,
        pack: ImagePackSource,
        display_name: Patch<String>,
        avatar: Patch<OwnedMxcUri>,
        usage: Patch<Vec<PackUsage>>,
    ) -> Result<(), Error> {
        self.run(Command::Pack(PackCommand::SetMeta {
            pack,
            display_name,
            avatar,
            usage,
        }))
        .await
    }

    /// adds or replaces by shortcode. a replaced image keeps its info unless the url changed
    pub async fn put_pack_image(
        &self,
        pack: ImagePackSource,
        shortcode: String,
        mxc: OwnedMxcUri,
        body: Option<String>,
        usage: Option<Vec<PackUsage>>,
        info: Option<PackImageInfo>,
    ) -> Result<(), Error> {
        self.run(Command::Pack(PackCommand::PutImage {
            pack,
            shortcode,
            mxc,
            body,
            usage,
            info,
        }))
        .await
    }

    pub async fn rename_pack_image(
        &self,
        pack: ImagePackSource,
        from: String,
        to: String,
    ) -> Result<(), Error> {
        self.run(Command::Pack(PackCommand::RenameImage { pack, from, to }))
            .await
    }

    pub async fn remove_pack_image(
        &self,
        pack: ImagePackSource,
        shortcode: String,
    ) -> Result<(), Error> {
        self.run(Command::Pack(PackCommand::RemoveImage { pack, shortcode }))
            .await
    }

    pub async fn set_pack_enabled_everywhere(
        &self,
        room_id: OwnedRoomId,
        state_key: String,
        enabled: bool,
    ) -> Result<(), Error> {
        self.run(Command::Pack(PackCommand::SetEnabledEverywhere {
            room_id,
            state_key,
            enabled,
        }))
        .await
    }

    /// the same uri when it already lives on your homeserver
    pub async fn rehost_media(&self, mxc: OwnedMxcUri) -> Result<OwnedMxcUri, Error> {
        match self
            .execute(Command::Pack(PackCommand::Rehost(mxc)))
            .await?
        {
            CommandOutput::Mxc(uri) => Ok(uri),
            output => Err(unexpected("an mxc", &output)),
        }
    }

    /// hands back the shortcode it ended up under. `usage` overrides what the image had
    pub async fn clone_pack_image(
        &self,
        from_pack: String,
        shortcode: String,
        to: ImagePackSource,
        usage: Option<Vec<PackUsage>>,
    ) -> Result<String, Error> {
        let (mut shortcodes, _) = self
            .run_clone(PackCommand::CloneImage {
                from_pack,
                shortcode,
                to,
                usage,
            })
            .await?;
        shortcodes
            .pop()
            .ok_or_else(|| unexpected("a shortcode", &CommandOutput::Done))
    }

    /// hands back the shortcode it ended up under
    pub async fn clone_sticker(
        &self,
        room_id: OwnedRoomId,
        event_id: OwnedEventId,
        to: ImagePackSource,
        shortcode: String,
        usage: Option<Vec<PackUsage>>,
    ) -> Result<String, Error> {
        let (mut shortcodes, _) = self
            .run_clone(PackCommand::CloneSticker {
                room_id,
                event_id,
                to,
                shortcode,
                usage,
            })
            .await?;
        shortcodes
            .pop()
            .ok_or_else(|| unexpected("a shortcode", &CommandOutput::Done))
    }

    /// hands back the shortcodes that couldn't be fetched, everything else is written
    pub async fn clone_pack(
        &self,
        from_pack: String,
        to: ImagePackSource,
        display_name: Option<String>,
        usage: Option<Vec<PackUsage>>,
    ) -> Result<Vec<String>, Error> {
        let (_, skipped) = self
            .run_clone(PackCommand::ClonePack {
                from_pack,
                to,
                display_name,
                usage,
            })
            .await?;
        Ok(skipped)
    }

    async fn run_clone(&self, command: PackCommand) -> Result<(Vec<String>, Vec<String>), Error> {
        match self.execute(Command::Pack(command)).await? {
            CommandOutput::Cloned {
                shortcodes,
                skipped,
            } => Ok((shortcodes, skipped)),
            output => Err(unexpected("a clone", &output)),
        }
    }

    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)?;
        let unread = self
            .client
            .get_room(room_id)
            .map(|room| (room.num_unread_notifications(), room.num_unread_mentions()));
        Ok(normalise_push_notification(
            room_id, event_id, status, unread,
        ))
    }

    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 channel() -> (Self, Pending) {
        let (reply, pending) = oneshot::channel();
        (Self(reply), Pending(pending))
    }

    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>,
    publication: Arc<PublicationRegistry>,
}

#[doc(hidden)]
#[derive(Clone, Debug)]
pub struct PublicationAck(Arc<PublicationAckInner>);

#[derive(Debug)]
struct PublicationAckInner {
    sender: Mutex<Option<oneshot::Sender<bool>>>,
}

impl PartialEq for PublicationAck {
    fn eq(&self, other: &Self) -> bool {
        Arc::ptr_eq(&self.0, &other.0)
    }
}

impl Eq for PublicationAck {}

impl PublicationAck {
    pub(crate) fn complete(&self, published: bool) {
        let sender = self
            .0
            .sender
            .lock()
            .unwrap_or_else(|error| error.into_inner())
            .take();
        if let Some(sender) = sender {
            let _ = sender.send(published);
        }
    }
}

#[derive(Debug, Default)]
struct PublicationRegistry {
    state: Mutex<PublicationRegistryState>,
}

#[derive(Debug, Default)]
struct PublicationRegistryState {
    mirrors: usize,
    pending: Vec<Weak<PublicationAckInner>>,
}

impl PublicationRegistry {
    fn register_mirror(self: &Arc<Self>) -> PublicationRegistration {
        self.state
            .lock()
            .unwrap_or_else(|error| error.into_inner())
            .mirrors += 1;
        PublicationRegistration {
            registry: self.clone(),
        }
    }

    fn acknowledgement(&self) -> Option<(PublicationAck, oneshot::Receiver<bool>)> {
        let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner());
        if state.mirrors == 0 {
            return None;
        }
        state.pending.retain(|pending| pending.strong_count() > 0);
        let (sender, receiver) = oneshot::channel();
        let acknowledgement = PublicationAck(Arc::new(PublicationAckInner {
            sender: Mutex::new(Some(sender)),
        }));
        state.pending.push(Arc::downgrade(&acknowledgement.0));
        Some((acknowledgement, receiver))
    }
}

pub(crate) struct PublicationRegistration {
    registry: Arc<PublicationRegistry>,
}

impl Drop for PublicationRegistration {
    fn drop(&mut self) {
        let pending = {
            let mut state = self
                .registry
                .state
                .lock()
                .unwrap_or_else(|error| error.into_inner());
            state.mirrors -= 1;
            if state.mirrors == 0 {
                std::mem::take(&mut state.pending)
            } else {
                Vec::new()
            }
        };
        for acknowledgement in pending.into_iter().filter_map(|ack| ack.upgrade()) {
            PublicationAck(acknowledgement).complete(false);
        }
    }
}

#[derive(Clone, Debug)]
pub(crate) struct EventSender {
    events: mpsc::Sender<Event>,
    publication: Arc<PublicationRegistry>,
}

impl EventSender {
    #[cfg(all(test, feature = "sqlite"))]
    pub(crate) fn detached(capacity: usize) -> (Self, mpsc::Receiver<Event>) {
        let (events, receiver) = mpsc::channel(capacity);
        let sender = Self {
            events,
            publication: Arc::new(PublicationRegistry::default()),
        };
        (sender, receiver)
    }

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

    pub(crate) async fn publish_barrier(&self) -> Result<(), SendEventError> {
        let Some((acknowledgement, published)) = self.publication.acknowledgement() else {
            return Ok(());
        };
        if let Err(error) = self
            .send(Event::PublicationBarrier {
                acknowledgement: Some(acknowledgement.clone()),
            })
            .await
        {
            acknowledgement.complete(false);
            return Err(error);
        }
        drop(acknowledgement);
        match published.await {
            Ok(true) => Ok(()),
            Ok(false) | Err(_) => Err(SendEventError),
        }
    }

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

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

impl DriverPort {
    pub(crate) fn event_sender(&self) -> EventSender {
        EventSender {
            events: self.events.clone(),
            publication: self.publication.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);
    let publication = Arc::new(PublicationRegistry::default());
    (
        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(),
            publication: publication.clone(),
            _store_lock: store_lock,
        },
        DriverPort {
            requests: request_rx,
            events: event_tx,
            connection: connection_tx,
            publication,
        },
        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,
    unread: Option<(u64, u64)>,
) -> 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(),
                    unread_notifications: unread.map(|(notifications, _)| notifications),
                    unread_mentions: unread.map(|(_, mentions)| mentions),
                    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,
    }
}

pub(crate) async fn remove_pusher(client: &Client, pusher: PusherKey) -> Result<(), Error> {
    use ruma::api::client::push::PusherIds;
    client
        .pusher()
        .delete(PusherIds::new(pusher.pushkey, pusher.app_id))
        .await
        .map_err(|source| Error::from_sdk("removing the pusher failed", source))
}

fn unexpected(wanted: &str, output: &CommandOutput) -> Error {
    Error::new(
        ErrorKind::Other,
        format!("expected {wanted} but got {output:?}"),
    )
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn publish_barrier_waits_for_mirror_publication() {
        let publication = Arc::new(PublicationRegistry::default());
        let _registration = publication.register_mirror();
        let (events, mut received) = mpsc::channel(1);
        let sender = EventSender {
            events,
            publication,
        };
        let publisher = tokio::spawn(async move { sender.publish_barrier().await });

        let Some(Event::PublicationBarrier {
            acknowledgement: Some(acknowledgement),
        }) = received.recv().await
        else {
            panic!("barrier was not published");
        };
        tokio::task::yield_now().await;
        assert!(!publisher.is_finished());

        acknowledgement.complete(true);
        assert!(matches!(publisher.await.unwrap(), Ok(())));
    }

    #[tokio::test]
    async fn publish_barrier_fails_when_last_mirror_drops() {
        let publication = Arc::new(PublicationRegistry::default());
        let registration = publication.register_mirror();
        let (events, mut received) = mpsc::channel(1);
        let sender = EventSender {
            events,
            publication,
        };
        let publisher = tokio::spawn(async move { sender.publish_barrier().await });

        assert!(matches!(
            received.recv().await,
            Some(Event::PublicationBarrier {
                acknowledgement: Some(_),
            })
        ));
        drop(registration);

        assert!(matches!(publisher.await.unwrap(), Err(SendEventError)));
    }
}