macltok

public
Star0
Watch0
1 commit 3 branches 0 tags 0 releases3.0 MiB
use std::sync::{
    Arc,
    atomic::{AtomicU64, Ordering},
};
use std::{
    fmt,
    pin::Pin,
    task::{Context, Poll},
};

use matrix_sdk::{Client, RoomMemberships};
use matrix_sdk_ui::{
    notification_client::{
        NotificationClient, NotificationProcessSetup, NotificationStatus, RawNotificationEvent,
    },
    sync_service::SyncService,
};
use ruma::{
    EventId, RoomId,
    events::{
        AnyGlobalAccountDataEventContent, AnyRoomAccountDataEvent, GlobalAccountDataEventType,
        RoomAccountDataEventType,
    },
    serde::Raw,
};
use thiserror::Error;
use tokio::sync::{OnceCell, mpsc, oneshot, watch};

use crate::{
    AccountCommand, Command, CommandId, ConnectionState, EncryptionState, Event, InteractiveAuth,
    InteractiveAuthOutcome, MemberFilter, MemberSummary, NotificationCommand, OwnDevice,
    PowerLevelChanges, PushNotification, PushNotificationEvent, PushNotificationStatus,
    RoomListFilter, RoomNotificationMode, RoomNotificationSettings, RoomPowerLevels, RoomSummary,
    encryption::encryption_state,
    model::{PushNotificationRoom, PushNotificationSender},
    normalise::{member_summary, room_summaries},
};

#[derive(Clone)]
pub struct ClientHandle {
    commands: mpsc::Sender<DriverRequest>,
    next_id: Arc<AtomicU64>,
    client: Client,
    connection: watch::Receiver<ConnectionState>,
    notification_client: Arc<OnceCell<NotificationClient>>,
    sync_service: Arc<SyncService>,
}

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

impl ClientHandle {
    /// Current credentials after a refresh. Persist this snapshot atomically
    /// when `Event::SessionTokensChanged` arrives; it contains secrets.
    #[cfg(feature = "sqlite")]
    pub fn session_record(&self) -> Result<crate::SessionRecord, crate::AccountError> {
        crate::account::session_record(&self.client)
    }

    /// Queue a command and return its correlation ID, not its result. Drain
    /// `EventStream` concurrently and match one terminal result event by ID.
    /// A successful timeline send only means the SDK queue accepted it.
    pub async fn send(&self, command: Command) -> Result<CommandId, SendCommandError> {
        let id = self.next_command_id();
        self.commands
            .send(DriverRequest::Command(CommandEnvelope { id, command }))
            .await
            .map_err(|_| SendCommandError::Closed)?;
        Ok(id)
    }

    pub async fn delete_devices(
        &self,
        device_ids: Vec<ruma::OwnedDeviceId>,
        auth: Option<InteractiveAuth>,
    ) -> Result<InteractiveAuthOutcome, CommandResultError> {
        let (reply, result) = oneshot::channel();
        self.commands
            .send(DriverRequest::DeleteDevices {
                id: self.next_command_id(),
                device_ids,
                auth,
                reply,
            })
            .await
            .map_err(|_| SendCommandError::Closed)?;
        result.await.map_err(|_| CommandResultError::Closed)?
    }

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

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

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

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

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

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

    pub async fn open_room(
        &self,
        room_id: ruma::OwnedRoomId,
    ) -> Result<CommandId, SendCommandError> {
        self.send(Command::Room(crate::RoomCommand::Open(room_id)))
            .await
    }

    pub async fn close_room(
        &self,
        room_id: ruma::OwnedRoomId,
    ) -> Result<CommandId, SendCommandError> {
        self.send(Command::Room(crate::RoomCommand::Close(room_id)))
            .await
    }

    pub async fn open_thread(
        &self,
        room_id: ruma::OwnedRoomId,
        root_event_id: ruma::OwnedEventId,
    ) -> Result<CommandId, SendCommandError> {
        self.send(Command::Timeline(crate::TimelineCommand::OpenThread {
            room_id,
            root_event_id,
        }))
        .await
    }

    pub async fn close_thread(
        &self,
        room_id: ruma::OwnedRoomId,
        root_event_id: ruma::OwnedEventId,
    ) -> Result<CommandId, SendCommandError> {
        self.send(Command::Timeline(crate::TimelineCommand::CloseThread {
            room_id,
            root_event_id,
        }))
        .await
    }

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

    pub async fn leave_room(
        &self,
        room_id: ruma::OwnedRoomId,
    ) -> Result<CommandId, SendCommandError> {
        self.send(Command::Room(crate::RoomCommand::Leave(room_id)))
            .await
    }

    pub async fn forget_room(
        &self,
        room_id: ruma::OwnedRoomId,
    ) -> Result<CommandId, SendCommandError> {
        self.send(Command::Room(crate::RoomCommand::Forget(room_id)))
            .await
    }

    pub async fn accept_invite(
        &self,
        room_id: ruma::OwnedRoomId,
    ) -> Result<CommandId, SendCommandError> {
        self.send(Command::Room(crate::RoomCommand::AcceptInvite(room_id)))
            .await
    }

    pub async fn reject_invite(
        &self,
        room_id: ruma::OwnedRoomId,
    ) -> Result<CommandId, SendCommandError> {
        self.send(Command::Room(crate::RoomCommand::RejectInvite(room_id)))
            .await
    }

    pub async fn invite_user(
        &self,
        room_id: ruma::OwnedRoomId,
        user_id: ruma::OwnedUserId,
    ) -> Result<CommandId, SendCommandError> {
        self.send(Command::Room(crate::RoomCommand::InviteUser {
            room_id,
            user_id,
        }))
        .await
    }

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

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

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

    pub async fn set_power_levels(
        &self,
        room_id: ruma::OwnedRoomId,
        changes: PowerLevelChanges,
    ) -> Result<CommandId, SendCommandError> {
        self.send(Command::Room(crate::RoomCommand::SetPowerLevels {
            room_id,
            changes,
        }))
        .await
    }

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

    pub async fn create_room(
        &self,
        options: crate::CreateRoomOptions,
    ) -> Result<ruma::OwnedRoomId, CommandResultError> {
        self.room_result(crate::RoomCommand::Create(options)).await
    }

    pub async fn create_direct(
        &self,
        user_id: ruma::OwnedUserId,
    ) -> Result<ruma::OwnedRoomId, CommandResultError> {
        self.room_result(crate::RoomCommand::CreateDirect(user_id))
            .await
    }

    async fn room_result(
        &self,
        command: crate::RoomCommand,
    ) -> Result<ruma::OwnedRoomId, CommandResultError> {
        let (reply, result) = oneshot::channel();
        self.commands
            .send(DriverRequest::RoomResult {
                id: self.next_command_id(),
                command,
                reply,
            })
            .await
            .map_err(|_| SendCommandError::Closed)?;
        result.await.map_err(|_| CommandResultError::Closed)?
    }

    pub async fn cancel_send(
        &self,
        room_id: ruma::OwnedRoomId,
        transaction_id: ruma::OwnedTransactionId,
    ) -> Result<CommandId, SendCommandError> {
        self.send(Command::Room(crate::RoomCommand::CancelSend {
            room_id,
            transaction_id,
        }))
        .await
    }

    pub async fn retry_send(
        &self,
        room_id: ruma::OwnedRoomId,
        transaction_id: ruma::OwnedTransactionId,
    ) -> Result<CommandId, SendCommandError> {
        self.send(Command::Room(crate::RoomCommand::RetrySend {
            room_id,
            transaction_id,
        }))
        .await
    }

    pub async fn upload_media(
        &self,
        timeline_id: crate::TimelineId,
        filename: String,
        mimetype: String,
        bytes: Vec<u8>,
        caption: Option<String>,
    ) -> Result<CommandId, SendCommandError> {
        self.send(Command::Media(crate::MediaCommand::Upload {
            timeline_id,
            filename,
            mimetype,
            bytes,
            caption,
        }))
        .await
    }

    pub async fn upload_attachment(
        &self,
        upload: crate::AttachmentUpload,
    ) -> Result<CommandId, SendCommandError> {
        self.send(Command::Media(crate::MediaCommand::UploadAttachment(
            upload,
        )))
        .await
    }

    pub async fn create_key_backup(&self) -> Result<CommandId, SendCommandError> {
        self.send(Command::Encryption(
            crate::EncryptionCommand::CreateKeyBackup,
        ))
        .await
    }

    pub async fn enable_key_backup(&self) -> Result<CommandId, SendCommandError> {
        self.send(Command::Encryption(
            crate::EncryptionCommand::EnableKeyBackup,
        ))
        .await
    }

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

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

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

    pub fn try_send(&self, command: Command) -> Result<CommandId, SendCommandError> {
        let id = self.next_command_id();
        self.commands
            .try_send(DriverRequest::Command(CommandEnvelope { id, command }))
            .map_err(|error| match error {
                mpsc::error::TrySendError::Full(_) => SendCommandError::Full,
                mpsc::error::TrySendError::Closed(_) => SendCommandError::Closed,
            })?;
        Ok(id)
    }

    fn next_command_id(&self) -> CommandId {
        CommandId::new(self.next_id.fetch_add(1, Ordering::Relaxed))
    }

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

    pub async fn rooms_filtered(&self, filter: &RoomListFilter) -> Vec<RoomSummary> {
        let mut room_filter = filter.clone();
        let search = room_filter
            .search
            .take()
            .map(|search| search.to_lowercase());

        self.rooms()
            .await
            .into_iter()
            .filter(|room| {
                room_filter.matches(room)
                    && search.as_ref().is_none_or(|search| {
                        room.name.to_lowercase().contains(search)
                            || room.room_id.as_str().to_lowercase().contains(search)
                    })
            })
            .collect()
    }

    pub async fn devices(&self) -> Result<Vec<OwnDevice>, ClientQueryError> {
        let current_device_id = self.client.device_id();
        Ok(self
            .client
            .devices()
            .await
            .map_err(|source| ClientQueryError::Sdk(source.into()))?
            .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())
    }

    pub async fn room_notification_settings(
        &self,
        room_id: &RoomId,
    ) -> Result<Option<RoomNotificationSettings>, ClientQueryError> {
        let room = self
            .client
            .get_room(room_id)
            .ok_or_else(|| ClientQueryError::RoomUnknown(room_id.to_owned()))?;
        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, ClientQueryError> {
        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(ClientQueryError::Notification)?;
        let status = client
            .get_notification(room_id, event_id)
            .await
            .map_err(ClientQueryError::Notification)?;
        Ok(normalise_push_notification(room_id, event_id, status))
    }

    pub async fn members(
        &self,
        room_id: &RoomId,
        filter: MemberFilter,
    ) -> Result<Vec<MemberSummary>, ClientQueryError> {
        let room = self
            .client
            .get_room(room_id)
            .ok_or_else(|| ClientQueryError::RoomUnknown(room_id.to_owned()))?;
        let memberships = match filter {
            MemberFilter::Active => RoomMemberships::ACTIVE,
            MemberFilter::Joined => RoomMemberships::JOIN,
            MemberFilter::Invited => RoomMemberships::INVITE,
            // 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(ClientQueryError::Sdk)?
            .into_iter()
            .map(member_summary)
            .collect())
    }

    pub async fn room_power_levels(
        &self,
        room_id: &RoomId,
    ) -> Result<RoomPowerLevels, ClientQueryError> {
        let room = self
            .client
            .get_room(room_id)
            .ok_or_else(|| ClientQueryError::RoomUnknown(room_id.to_owned()))?;
        let levels = room
            .power_levels()
            .await
            .map_err(|source| ClientQueryError::Sdk(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(),
        })
    }

    pub async fn pending_send_ids(
        &self,
        room_id: &RoomId,
    ) -> Result<Vec<ruma::OwnedTransactionId>, ClientQueryError> {
        let room = self
            .client
            .get_room(room_id)
            .ok_or_else(|| ClientQueryError::RoomUnknown(room_id.to_owned()))?;
        let (echoes, _) = room
            .send_queue()
            .subscribe()
            .await
            .map_err(ClientQueryError::SendQueue)?;
        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>>, ClientQueryError> {
        self.client
            .account()
            .account_data_raw(event_type)
            .await
            .map_err(ClientQueryError::Sdk)
    }

    pub async fn set_global_account_data(
        &self,
        event_type: GlobalAccountDataEventType,
        content: Raw<AnyGlobalAccountDataEventContent>,
    ) -> Result<(), ClientQueryError> {
        let (reply, result) = oneshot::channel();
        self.commands
            .send(DriverRequest::SetGlobalAccountData {
                event_type,
                content,
                reply,
            })
            .await
            .map_err(|_| ClientQueryError::EngineStopped)?;
        result.await.map_err(|_| ClientQueryError::EngineStopped)?
    }

    pub async fn room_account_data(
        &self,
        room_id: &RoomId,
        event_type: RoomAccountDataEventType,
    ) -> Result<Option<Raw<AnyRoomAccountDataEvent>>, ClientQueryError> {
        let room = self
            .client
            .get_room(room_id)
            .ok_or_else(|| ClientQueryError::RoomUnknown(room_id.to_owned()))?;
        room.account_data(event_type)
            .await
            .map_err(ClientQueryError::Sdk)
    }

    pub async fn set_room_account_data(
        &self,
        room_id: &RoomId,
        event_type: RoomAccountDataEventType,
        content: Raw<ruma::events::AnyRoomAccountDataEventContent>,
    ) -> Result<(), ClientQueryError> {
        let (reply, result) = oneshot::channel();
        self.commands
            .send(DriverRequest::SetRoomAccountData {
                room_id: room_id.to_owned(),
                event_type,
                content,
                reply,
            })
            .await
            .map_err(|_| ClientQueryError::EngineStopped)?;
        result.await.map_err(|_| ClientQueryError::EngineStopped)?
    }

    #[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(),
        ))
    }
}

#[derive(Debug, Eq, PartialEq)]
pub(crate) struct CommandEnvelope {
    pub(crate) id: CommandId,
    pub(crate) command: Command,
}

#[derive(Debug)]
pub(crate) enum DriverRequest {
    Command(CommandEnvelope),
    DeleteDevices {
        id: CommandId,
        device_ids: Vec<ruma::OwnedDeviceId>,
        auth: Option<InteractiveAuth>,
        reply: oneshot::Sender<Result<InteractiveAuthOutcome, CommandResultError>>,
    },
    RoomResult {
        id: CommandId,
        command: crate::RoomCommand,
        reply: oneshot::Sender<Result<ruma::OwnedRoomId, CommandResultError>>,
    },
    SetGlobalAccountData {
        event_type: GlobalAccountDataEventType,
        content: Raw<AnyGlobalAccountDataEventContent>,
        reply: oneshot::Sender<Result<(), ClientQueryError>>,
    },
    SetRoomAccountData {
        room_id: ruma::OwnedRoomId,
        event_type: RoomAccountDataEventType,
        content: Raw<ruma::events::AnyRoomAccountDataEventContent>,
        reply: oneshot::Sender<Result<(), ClientQueryError>>,
    },
}

#[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 {
    commands: mpsc::Receiver<DriverRequest>,
    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::Closed)
    }

    /// Resolves once every [`EventStream`] receiver has been dropped, so the
    /// engine can notice a lost event consumer inside a `select!`.
    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 async fn recv(&mut self) -> Option<DriverRequest> {
        self.commands.recv().await
    }

    pub async fn send(&self, event: Event) -> Result<(), SendEventError> {
        self.event_sender().send(event).await
    }
}

#[derive(Clone, Copy, Debug, Eq, Error, PartialEq)]
pub enum SendCommandError {
    #[error("the command channel is full")]
    Full,
    #[error("the engine has stopped")]
    Closed,
}

#[derive(Debug, Error)]
pub enum CommandResultError {
    #[error(transparent)]
    Send(#[from] SendCommandError),
    #[error("the engine stopped before the operation completed")]
    Closed,
    #[error("{0:?}")]
    Failed(crate::CommandFailure),
}

#[derive(Debug, Error)]
pub enum ClientQueryError {
    #[error("room {0} is not available in the local store")]
    RoomUnknown(ruma::OwnedRoomId),
    #[error("Matrix query failed")]
    Sdk(#[source] matrix_sdk::Error),
    #[error("reading the send queue failed")]
    SendQueue(#[source] matrix_sdk::send_queue::RoomSendQueueError),
    #[error("resolving the push notification failed")]
    Notification(#[source] matrix_sdk_ui::notification_client::Error),
    #[error("the engine stopped before the operation completed")]
    EngineStopped,
}

#[derive(Clone, Copy, Debug, Eq, Error, PartialEq)]
pub(crate) enum SendEventError {
    #[error("the event receiver has closed")]
    Closed,
}

#[must_use]
pub(crate) fn transport_pair(
    client: Client,
    sync_service: Arc<SyncService>,
    command_capacity: usize,
    event_capacity: usize,
) -> (ClientHandle, DriverPort, EventStream) {
    let (command_tx, command_rx) = mpsc::channel(command_capacity);
    let (event_tx, event_rx) = mpsc::channel(event_capacity);
    let (connection_tx, connection_rx) = watch::channel(ConnectionState::Idle);
    (
        ClientHandle {
            commands: command_tx,
            next_id: Arc::new(AtomicU64::new(1)),
            client,
            connection: connection_rx,
            notification_client: Arc::new(OnceCell::new()),
            sync_service,
        },
        DriverPort {
            commands: command_rx,
            events: event_tx,
            connection: connection_tx,
        },
        EventStream { events: event_rx },
    )
}

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.map(Into::into),
                    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.map(Into::into),
                    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,
    }
}

#[cfg(test)]
mod tests {
    use matrix_sdk::{Client, SessionMeta, SessionTokens, authentication::matrix::MatrixSession};

    use crate::{ConnectionState, LifecycleCommand};

    use super::*;

    async fn sync_service(client: Client) -> Arc<SyncService> {
        Arc::new(SyncService::builder(client).build().await.unwrap())
    }

    async fn test_client() -> Client {
        let client = Client::builder()
            .homeserver_url("http://localhost")
            .build()
            .await
            .unwrap();
        client
            .restore_session(MatrixSession {
                meta: SessionMeta {
                    user_id: "@test:localhost".parse().unwrap(),
                    device_id: "DEVICE".into(),
                },
                tokens: SessionTokens {
                    access_token: "access_token".to_owned(),
                    refresh_token: None,
                },
            })
            .await
            .unwrap();
        client
    }

    #[tokio::test]
    async fn transports_are_isolated() {
        let first_client = test_client().await;
        let second_client = test_client().await;
        let (first_handle, mut first_driver, mut first_events) =
            transport_pair(first_client.clone(), sync_service(first_client).await, 2, 2);
        let (_second_handle, mut second_driver, mut second_events) = transport_pair(
            second_client.clone(),
            sync_service(second_client).await,
            2,
            2,
        );

        let id = first_handle
            .send(Command::Lifecycle(LifecycleCommand::Shutdown))
            .await
            .unwrap();
        let request = first_driver.recv().await.unwrap();
        assert!(matches!(
            request,
            DriverRequest::Command(CommandEnvelope {
                id: received_id,
                command: Command::Lifecycle(LifecycleCommand::Shutdown),
            }) if received_id == id
        ));
        assert!(second_driver.commands.try_recv().is_err());

        first_driver
            .send(Event::ConnectionChanged(ConnectionState::Online))
            .await
            .unwrap();
        assert_eq!(
            first_events.recv().await,
            Some(Event::ConnectionChanged(ConnectionState::Online))
        );
        assert!(second_events.try_recv().is_err());
    }

    #[tokio::test]
    async fn bounded_command_channel_reports_pressure() {
        let client = test_client().await;
        let (handle, _driver, _events) =
            transport_pair(client.clone(), sync_service(client).await, 1, 1);

        handle
            .try_send(Command::Lifecycle(LifecycleCommand::Shutdown))
            .unwrap();
        assert_eq!(
            handle.try_send(Command::Lifecycle(LifecycleCommand::Shutdown)),
            Err(SendCommandError::Full)
        );
    }

    #[tokio::test]
    async fn cloned_handles_share_command_id_sequence() {
        let client = test_client().await;
        let (handle, mut driver, _events) =
            transport_pair(client.clone(), sync_service(client).await, 1, 1);
        let cloned = handle.clone();
        let first = handle
            .try_send(Command::Lifecycle(LifecycleCommand::Shutdown))
            .unwrap();
        assert_eq!(first.get(), 1);
        assert_eq!(
            cloned.try_send(Command::Lifecycle(LifecycleCommand::Shutdown)),
            Err(SendCommandError::Full)
        );
        driver.recv().await.unwrap();

        let third = cloned
            .send(Command::Lifecycle(LifecycleCommand::Shutdown))
            .await
            .unwrap();
        assert_eq!(third.get(), 3);
        driver.recv().await.unwrap();

        let fourth = handle
            .try_send(Command::Lifecycle(LifecycleCommand::Shutdown))
            .unwrap();
        assert_eq!(fourth.get(), 4);
    }
}