macltok

public

Star0

Watch0

125 commits 3 branches 0 tags 0 releases3.0 MiB

use std::{future::Future, panic::AssertUnwindSafe};

use futures_util::FutureExt;
use matrix_sdk::encryption::verification::VerificationRequest;
use matrix_sdk::{
    Client,
    notification_settings::{IsEncrypted, IsOneToOne, RoomNotificationMode as SdkMode},
    ruma::{
        OwnedDeviceId, OwnedEventId, OwnedMxcUri, RoomId, TransactionId,
        api::client::{
            push::{get_pushrule, set_pushrule_actions},
            receipt::create_receipt::v3::ReceiptType,
        },
        events::receipt::ReceiptThread,
        push::{Action, HighlightTweakValue, RuleKind, Tweak},
    },
    send_queue::{LocalEchoContent, SendHandle},
};
use tokio::sync::{mpsc, watch};

use crate::{
    CommandOutput, EncryptionCommand, Error, ErrorKind, InteractiveAuth, InteractiveAuthChallenge,
    NotificationCommand, RecoveryKey, RoomNotificationMode, TimelineId,
    encryption::VerificationController,
    transport::{EventSender, Responder},
};

pub(super) struct Job<T> {
    pub(super) command: T,
    pub(super) reply: Responder,
}

/// a panicking command answers with an error instead of taking its worker down with it
pub(super) async fn guarded<T>(
    label: &'static str,
    work: impl Future<Output = Result<T, Error>>,
) -> Result<T, Error> {
    AssertUnwindSafe(work)
        .catch_unwind()
        .await
        .unwrap_or_else(|_| Err(Error::new(ErrorKind::Other, format!("{label} panicked"))))
}

pub(super) async fn run_notifications(
    client: Client,
    mut jobs: mpsc::Receiver<Job<NotificationCommand>>,
) {
    while let Some(Job { command, reply }) = jobs.recv().await {
        reply.done(guarded("notification command", notification(&client, command)).await);
    }
}

pub(super) async fn run_encryption(
    client: Client,
    events: EventSender,
    mut jobs: mpsc::Receiver<Job<EncryptionCommand>>,
    mut incoming: mpsc::UnboundedReceiver<VerificationRequest>,
    enabling: watch::Sender<bool>,
) {
    let mut verification = VerificationController::default();
    loop {
        tokio::select! {
            job = jobs.recv() => {
                let Some(Job { command, reply }) = job else { return };
                let work = encryption(&client, &events, &mut verification, &enabling, command);
                reply.send(guarded("encryption command", work).await);
            }
            Some(request) = incoming.recv() => {
                verification.adopt_incoming(events.clone(), request).await;
            }
        }
    }
}

async fn encryption(
    client: &Client,
    events: &EventSender,
    verification: &mut VerificationController,
    enabling: &watch::Sender<bool>,
    command: EncryptionCommand,
) -> Result<CommandOutput, Error> {
    let encryption = client.encryption();
    let recovery = encryption.recovery();
    let key = |key: String| CommandOutput::RecoveryKey(RecoveryKey::new(key));
    let done = CommandOutput::Done;
    match command {
        EncryptionCommand::EnableRecovery { passphrase } => {
            let _enabling = Enabling::start(enabling);
            let enable = recovery.enable();
            match passphrase.as_deref() {
                Some(passphrase) => enable.with_passphrase(passphrase).await,
                None => enable.await,
            }
            .map(key)
            .map_err(|source| Error::from_recovery("enabling recovery failed", source))
        }
        EncryptionCommand::ResetRecoveryKey { passphrase } => {
            let reset = recovery.reset_key();
            match passphrase.as_deref() {
                Some(passphrase) => reset.with_passphrase(passphrase).await,
                None => reset.await,
            }
            .map(key)
            .map_err(|source| Error::from_recovery("resetting the recovery key failed", source))
        }
        EncryptionCommand::DisableRecovery => recovery
            .disable()
            .await
            .map(|()| done)
            .map_err(|source| Error::from_recovery("disabling recovery failed", source)),
        EncryptionCommand::Recover(key) => recovery
            .recover(&key)
            .await
            .map(|()| done)
            .map_err(|source| Error::from_recovery("account recovery failed", source)),
        EncryptionCommand::RecoverKeyBackup(key) => recovery
            .recover_and_fix_backup(&key)
            .await
            .map(|()| done)
            .map_err(|source| Error::from_recovery("recovering key backup failed", source)),
        EncryptionCommand::CreateKeyBackup => encryption
            .backups()
            .create()
            .await
            .map(|()| done)
            .map_err(|source| Error::from_sdk("creating key backup failed", source)),
        EncryptionCommand::EnableKeyBackup => recovery
            .enable_backup()
            .await
            .map(|()| done)
            .map_err(|source| Error::from_recovery("enabling key backup failed", source)),
        EncryptionCommand::DisableKeyBackup { delete_from_server } => {
            let backups = encryption.backups();
            if delete_from_server {
                backups.disable_and_delete().await
            } else {
                backups.disable().await
            }
            .map(|()| done)
            .map_err(|source| Error::from_sdk("disabling key backup failed", source))
        }
        EncryptionCommand::StartVerification(target) => verification
            .start(client, events.clone(), target)
            .await
            .map(|()| done),
        EncryptionCommand::Verify(action) => verification
            .apply(events.clone(), action)
            .await
            .map(|()| done),
    }
}

/// a panicking enable must not leave the state saying it is still running
struct Enabling<'a>(&'a watch::Sender<bool>);

impl<'a> Enabling<'a> {
    fn start(sender: &'a watch::Sender<bool>) -> Self {
        sender.send_replace(true);
        Self(sender)
    }
}

impl Drop for Enabling<'_> {
    fn drop(&mut self) {
        self.0.send_replace(false);
    }
}

async fn notification(client: &Client, command: NotificationCommand) -> Result<(), Error> {
    let settings = client.notification_settings().await;
    match command {
        NotificationCommand::SetRoomMode { room_id, mode } => {
            settings
                .set_room_notification_mode(&room_id, sdk_mode(mode))
                .await
        }
        NotificationCommand::ResetRoomMode(room_id) => {
            settings.delete_user_defined_room_rules(&room_id).await
        }
        NotificationCommand::SetDefaultMode {
            encrypted,
            one_to_one,
            mode,
        } => {
            settings
                .set_default_room_notification_mode(
                    IsEncrypted::from(encrypted),
                    IsOneToOne::from(one_to_one),
                    sdk_mode(mode),
                )
                .await
        }
        NotificationCommand::AddKeyword(keyword) => {
            match settings.add_keyword(keyword.clone()).await {
                Ok(()) => return highlight_keyword(client, &keyword).await,
                Err(error) => Err(error),
            }
        }
        NotificationCommand::RemoveKeyword(keyword) => settings.remove_keyword(&keyword).await,
    }
    .map_err(|source| {
        Error::with_source(
            ErrorKind::Server,
            "updating notification settings failed",
            source,
        )
    })
}

/// sdk keyword rules carry only notify and a sound, while Element's loud keywords also
/// highlight, which is what lets the inbox keep them as mentions. The sdk names a
/// keyword's content rule after the keyword itself, a rule under another id is left
/// alone.
async fn highlight_keyword(client: &Client, keyword: &str) -> Result<(), Error> {
    let failed = |source| Error::from_http("highlighting a notification keyword failed", source);
    let rule = match client
        .send(get_pushrule::v3::Request::new(
            RuleKind::Content,
            keyword.to_owned(),
        ))
        .await
    {
        Ok(response) => response.rule,
        Err(error)
            if error
                .as_client_api_error()
                .is_some_and(|error| error.status_code.as_u16() == 404) =>
        {
            return Ok(());
        }
        Err(error) => return Err(failed(error)),
    };
    if rule.actions.iter().any(Action::is_highlight) {
        return Ok(());
    }
    let mut actions: Vec<_> = rule
        .actions
        .into_iter()
        .filter(|action| !matches!(action, Action::SetTweak(Tweak::Highlight(_))))
        .collect();
    actions.push(Action::SetTweak(Tweak::Highlight(HighlightTweakValue::Yes)));
    client
        .send(set_pushrule_actions::v3::Request::new(
            RuleKind::Content,
            keyword.to_owned(),
            actions,
        ))
        .await
        .map_err(failed)?;
    Ok(())
}

const fn sdk_mode(mode: RoomNotificationMode) -> SdkMode {
    match mode {
        RoomNotificationMode::AllMessages => SdkMode::AllMessages,
        RoomNotificationMode::MentionsAndKeywordsOnly => SdkMode::MentionsAndKeywordsOnly,
        RoomNotificationMode::Mute => SdkMode::Mute,
    }
}

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

/// the closed-room path. without a timeline nothing stops the marker going backwards
pub(super) async fn mark_read(
    client: &Client,
    timeline_id: &TimelineId,
    event_id: OwnedEventId,
) -> Result<(), Error> {
    let room = room(client, timeline_id.room_id())?;
    let receipt_failed = |source| Error::from_sdk("sending read receipt failed", source);
    let thread = match timeline_id.thread_root() {
        None => ReceiptThread::Main,
        Some(root) => ReceiptThread::Thread(root.clone()),
    };
    room.send_single_receipt(ReceiptType::Read, thread, event_id.clone())
        .await
        .map_err(receipt_failed)?;
    if timeline_id.thread_root().is_none() {
        room.send_single_receipt(ReceiptType::FullyRead, ReceiptThread::Unthreaded, event_id)
            .await
            .map_err(receipt_failed)?;
    }
    Ok(())
}

pub(super) async fn set_typing(
    client: &Client,
    room_id: &RoomId,
    typing: bool,
) -> Result<(), Error> {
    room(client, room_id)?
        .typing_notice(typing)
        .await
        .map_err(|source| Error::from_sdk("typing update failed", source))
}

async fn send_handle(
    client: &Client,
    room_id: &RoomId,
    transaction_id: &TransactionId,
) -> Result<SendHandle, Error> {
    let (echoes, _) = room(client, room_id)?
        .send_queue()
        .subscribe()
        .await
        .map_err(|source| {
            Error::with_source(ErrorKind::Store, "reading send queue failed", source)
        })?;
    echoes
        .into_iter()
        .find_map(|echo| match echo.content {
            LocalEchoContent::Event { send_handle, .. }
                if echo.transaction_id.as_str() == transaction_id.as_str() =>
            {
                Some(send_handle)
            }
            _ => None,
        })
        .ok_or_else(|| {
            Error::new(
                ErrorKind::Invalid,
                "send is no longer pending or is not a message",
            )
        })
}

pub(super) async fn cancel_send(
    client: &Client,
    room_id: &RoomId,
    transaction_id: &TransactionId,
) -> Result<(), Error> {
    match send_handle(client, room_id, transaction_id)
        .await?
        .abort()
        .await
    {
        Ok(true) => Ok(()),
        Ok(false) => Err(Error::new(
            ErrorKind::Invalid,
            "send could not be cancelled because it has already been sent",
        )),
        Err(source) => Err(Error::with_source(
            ErrorKind::Store,
            "cancelling send failed",
            source,
        )),
    }
}

pub(super) async fn retry_send(
    client: &Client,
    room_id: &RoomId,
    transaction_id: &TransactionId,
) -> Result<(), Error> {
    send_handle(client, room_id, transaction_id)
        .await?
        .unwedge()
        .await
        .map_err(|source| Error::with_source(ErrorKind::Store, "retrying send failed", source))?;
    room(client, room_id)?.send_queue().set_enabled(true);
    Ok(())
}

pub(super) async fn upload_avatar(
    client: &Client,
    mimetype: &str,
    bytes: Vec<u8>,
    max_bytes: usize,
) -> Result<(), Error> {
    if bytes.len() > max_bytes {
        return Err(Error::new(
            ErrorKind::MediaTooLarge,
            "avatar exceeds the configured byte limit",
        ));
    }
    let mimetype: mime::Mime = mimetype
        .parse()
        .map_err(|_| Error::new(ErrorKind::Invalid, "avatar MIME type is invalid"))?;
    if mimetype.type_() != mime::IMAGE {
        return Err(Error::new(
            ErrorKind::Invalid,
            "an avatar has to be an image",
        ));
    }
    client
        .account()
        .upload_avatar(&mimetype, bytes)
        .await
        .map(drop)
        .map_err(|source| Error::from_sdk("uploading avatar failed", source))
}

pub(super) async fn upload_media(
    client: &Client,
    mimetype: &str,
    bytes: Vec<u8>,
    max_bytes: usize,
) -> Result<OwnedMxcUri, Error> {
    if bytes.len() > max_bytes {
        return Err(Error::new(
            ErrorKind::MediaTooLarge,
            "media exceeds the configured byte limit",
        ));
    }
    let mimetype: mime::Mime = mimetype
        .parse()
        .map_err(|_| Error::new(ErrorKind::Invalid, "media MIME type is invalid"))?;
    client
        .media()
        .upload(&mimetype, bytes, None)
        .await
        .map(|response| response.content_uri)
        .map_err(|source| Error::from_sdk("uploading media failed", source))
}

pub(super) async fn rename_device(
    client: &Client,
    device_id: &OwnedDeviceId,
    display_name: &str,
) -> Result<(), Error> {
    client
        .rename_device(device_id, display_name)
        .await
        .map(drop)
        .map_err(|source| Error::from_http("renaming device failed", source))
}

pub(super) async fn delete_devices(
    client: &Client,
    device_ids: &[OwnedDeviceId],
    auth: Option<InteractiveAuth>,
) -> Result<(), Error> {
    let user_id = client
        .user_id()
        .ok_or_else(|| Error::new(ErrorKind::Unauthenticated, "the client is not logged in"))?;
    let auth = auth
        .map(|auth| auth.into_sdk(user_id))
        .transpose()
        .map_err(|message| {
            Error::new(
                ErrorKind::Invalid,
                format!("interactive authentication data is invalid, {message}"),
            )
        })?;
    match client.delete_devices(device_ids, auth).await {
        Ok(_) => Ok(()),
        Err(source) => match source.as_uiaa_response() {
            Some(info) => Err(Error::interactive_auth(InteractiveAuthChallenge::from(
                info,
            ))),
            None => Err(Error::from_http("deleting devices failed", source)),
        },
    }
}

1	use std::{future::Future, panic::AssertUnwindSafe};
2
3	use futures_util::FutureExt;
4	use matrix_sdk::encryption::verification::VerificationRequest;
5	use matrix_sdk::{
6	Client,
7	notification_settings::{IsEncrypted, IsOneToOne, RoomNotificationMode as SdkMode},
8	ruma::{
9	OwnedDeviceId, OwnedEventId, OwnedMxcUri, RoomId, TransactionId,
10	api::client::{
11	push::{get_pushrule, set_pushrule_actions},
12	receipt::create_receipt::v3::ReceiptType,
13	},
14	events::receipt::ReceiptThread,
15	push::{Action, HighlightTweakValue, RuleKind, Tweak},
16	},
17	send_queue::{LocalEchoContent, SendHandle},
18	};
19	use tokio::sync::{mpsc, watch};
20
21	use crate::{
22	CommandOutput, EncryptionCommand, Error, ErrorKind, InteractiveAuth, InteractiveAuthChallenge,
23	NotificationCommand, RecoveryKey, RoomNotificationMode, TimelineId,
24	encryption::VerificationController,
25	transport::{EventSender, Responder},
26	};
27
28	pub(super) struct Job<T> {
29	pub(super) command: T,
30	pub(super) reply: Responder,
31	}
32
33	/// a panicking command answers with an error instead of taking its worker down with it
34	pub(super) async fn guarded<T>(
35	label: &'static str,
36	work: impl Future<Output = Result<T, Error>>,
37	) -> Result<T, Error> {
38	AssertUnwindSafe(work)
39	.catch_unwind()
40	.await
41	.unwrap_or_else(\|_\| Err(Error::new(ErrorKind::Other, format!("{label} panicked"))))
42	}
43
44	pub(super) async fn run_notifications(
45	client: Client,
46	mut jobs: mpsc::Receiver<Job<NotificationCommand>>,
47	) {
48	while let Some(Job { command, reply }) = jobs.recv().await {
49	reply.done(guarded("notification command", notification(&client, command)).await);
50	}
51	}
52
53	pub(super) async fn run_encryption(
54	client: Client,
55	events: EventSender,
56	mut jobs: mpsc::Receiver<Job<EncryptionCommand>>,
57	mut incoming: mpsc::UnboundedReceiver<VerificationRequest>,
58	enabling: watch::Sender<bool>,
59	) {
60	let mut verification = VerificationController::default();
61	loop {
62	tokio::select! {
63	job = jobs.recv() => {
64	let Some(Job { command, reply }) = job else { return };
65	let work = encryption(&client, &events, &mut verification, &enabling, command);
66	reply.send(guarded("encryption command", work).await);
67	}
68	Some(request) = incoming.recv() => {
69	verification.adopt_incoming(events.clone(), request).await;
70	}
71	}
72	}
73	}
74
75	async fn encryption(
76	client: &Client,
77	events: &EventSender,
78	verification: &mut VerificationController,
79	enabling: &watch::Sender<bool>,
80	command: EncryptionCommand,
81	) -> Result<CommandOutput, Error> {
82	let encryption = client.encryption();
83	let recovery = encryption.recovery();
84	let key = \|key: String\| CommandOutput::RecoveryKey(RecoveryKey::new(key));
85	let done = CommandOutput::Done;
86	match command {
87	EncryptionCommand::EnableRecovery { passphrase } => {
88	let _enabling = Enabling::start(enabling);
89	let enable = recovery.enable();
90	match passphrase.as_deref() {
91	Some(passphrase) => enable.with_passphrase(passphrase).await,
92	None => enable.await,
93	}
94	.map(key)
95	.map_err(\|source\| Error::from_recovery("enabling recovery failed", source))
96	}
97	EncryptionCommand::ResetRecoveryKey { passphrase } => {
98	let reset = recovery.reset_key();
99	match passphrase.as_deref() {
100	Some(passphrase) => reset.with_passphrase(passphrase).await,
101	None => reset.await,
102	}
103	.map(key)
104	.map_err(\|source\| Error::from_recovery("resetting the recovery key failed", source))
105	}
106	EncryptionCommand::DisableRecovery => recovery
107	.disable()
108	.await
109	.map(\|()\| done)
110	.map_err(\|source\| Error::from_recovery("disabling recovery failed", source)),
111	EncryptionCommand::Recover(key) => recovery
112	.recover(&key)
113	.await
114	.map(\|()\| done)
115	.map_err(\|source\| Error::from_recovery("account recovery failed", source)),
116	EncryptionCommand::RecoverKeyBackup(key) => recovery
117	.recover_and_fix_backup(&key)
118	.await
119	.map(\|()\| done)
120	.map_err(\|source\| Error::from_recovery("recovering key backup failed", source)),
121	EncryptionCommand::CreateKeyBackup => encryption
122	.backups()
123	.create()
124	.await
125	.map(\|()\| done)
126	.map_err(\|source\| Error::from_sdk("creating key backup failed", source)),
127	EncryptionCommand::EnableKeyBackup => recovery
128	.enable_backup()
129	.await
130	.map(\|()\| done)
131	.map_err(\|source\| Error::from_recovery("enabling key backup failed", source)),
132	EncryptionCommand::DisableKeyBackup { delete_from_server } => {
133	let backups = encryption.backups();
134	if delete_from_server {
135	backups.disable_and_delete().await
136	} else {
137	backups.disable().await
138	}
139	.map(\|()\| done)
140	.map_err(\|source\| Error::from_sdk("disabling key backup failed", source))
141	}
142	EncryptionCommand::StartVerification(target) => verification
143	.start(client, events.clone(), target)
144	.await
145	.map(\|()\| done),
146	EncryptionCommand::Verify(action) => verification
147	.apply(events.clone(), action)
148	.await
149	.map(\|()\| done),
150	}
151	}
152
153	/// a panicking enable must not leave the state saying it is still running
154	struct Enabling<'a>(&'a watch::Sender<bool>);
155
156	impl<'a> Enabling<'a> {
157	fn start(sender: &'a watch::Sender<bool>) -> Self {
158	sender.send_replace(true);
159	Self(sender)
160	}
161	}
162
163	impl Drop for Enabling<'_> {
164	fn drop(&mut self) {
165	self.0.send_replace(false);
166	}
167	}
168
169	async fn notification(client: &Client, command: NotificationCommand) -> Result<(), Error> {
170	let settings = client.notification_settings().await;
171	match command {
172	NotificationCommand::SetRoomMode { room_id, mode } => {
173	settings
174	.set_room_notification_mode(&room_id, sdk_mode(mode))
175	.await
176	}
177	NotificationCommand::ResetRoomMode(room_id) => {
178	settings.delete_user_defined_room_rules(&room_id).await
179	}
180	NotificationCommand::SetDefaultMode {
181	encrypted,
182	one_to_one,
183	mode,
184	} => {
185	settings
186	.set_default_room_notification_mode(
187	IsEncrypted::from(encrypted),
188	IsOneToOne::from(one_to_one),
189	sdk_mode(mode),
190	)
191	.await
192	}
193	NotificationCommand::AddKeyword(keyword) => {
194	match settings.add_keyword(keyword.clone()).await {
195	Ok(()) => return highlight_keyword(client, &keyword).await,
196	Err(error) => Err(error),
197	}
198	}
199	NotificationCommand::RemoveKeyword(keyword) => settings.remove_keyword(&keyword).await,
200	}
201	.map_err(\|source\| {
202	Error::with_source(
203	ErrorKind::Server,
204	"updating notification settings failed",
205	source,
206	)
207	})
208	}
209
210	/// sdk keyword rules carry only notify and a sound, while Element's loud keywords also
211	/// highlight, which is what lets the inbox keep them as mentions. The sdk names a
212	/// keyword's content rule after the keyword itself, a rule under another id is left
213	/// alone.
214	async fn highlight_keyword(client: &Client, keyword: &str) -> Result<(), Error> {
215	let failed = \|source\| Error::from_http("highlighting a notification keyword failed", source);
216	let rule = match client
217	.send(get_pushrule::v3::Request::new(
218	RuleKind::Content,
219	keyword.to_owned(),
220	))
221	.await
222	{
223	Ok(response) => response.rule,
224	Err(error)
225	if error
226	.as_client_api_error()
227	.is_some_and(\|error\| error.status_code.as_u16() == 404) =>
228	{
229	return Ok(());
230	}
231	Err(error) => return Err(failed(error)),
232	};
233	if rule.actions.iter().any(Action::is_highlight) {
234	return Ok(());
235	}
236	let mut actions: Vec<_> = rule
237	.actions
238	.into_iter()
239	.filter(\|action\| !matches!(action, Action::SetTweak(Tweak::Highlight(_))))
240	.collect();
241	actions.push(Action::SetTweak(Tweak::Highlight(HighlightTweakValue::Yes)));
242	client
243	.send(set_pushrule_actions::v3::Request::new(
244	RuleKind::Content,
245	keyword.to_owned(),
246	actions,
247	))
248	.await
249	.map_err(failed)?;
250	Ok(())
251	}
252
253	const fn sdk_mode(mode: RoomNotificationMode) -> SdkMode {
254	match mode {
255	RoomNotificationMode::AllMessages => SdkMode::AllMessages,
256	RoomNotificationMode::MentionsAndKeywordsOnly => SdkMode::MentionsAndKeywordsOnly,
257	RoomNotificationMode::Mute => SdkMode::Mute,
258	}
259	}
260
261	fn room(client: &Client, room_id: &RoomId) -> Result<matrix_sdk::Room, Error> {
262	client.get_room(room_id).ok_or_else(Error::room_unknown)
263	}
264
265	/// the closed-room path. without a timeline nothing stops the marker going backwards
266	pub(super) async fn mark_read(
267	client: &Client,
268	timeline_id: &TimelineId,
269	event_id: OwnedEventId,
270	) -> Result<(), Error> {
271	let room = room(client, timeline_id.room_id())?;
272	let receipt_failed = \|source\| Error::from_sdk("sending read receipt failed", source);
273	let thread = match timeline_id.thread_root() {
274	None => ReceiptThread::Main,
275	Some(root) => ReceiptThread::Thread(root.clone()),
276	};
277	room.send_single_receipt(ReceiptType::Read, thread, event_id.clone())
278	.await
279	.map_err(receipt_failed)?;
280	if timeline_id.thread_root().is_none() {
281	room.send_single_receipt(ReceiptType::FullyRead, ReceiptThread::Unthreaded, event_id)
282	.await
283	.map_err(receipt_failed)?;
284	}
285	Ok(())
286	}
287
288	pub(super) async fn set_typing(
289	client: &Client,
290	room_id: &RoomId,
291	typing: bool,
292	) -> Result<(), Error> {
293	room(client, room_id)?
294	.typing_notice(typing)
295	.await
296	.map_err(\|source\| Error::from_sdk("typing update failed", source))
297	}
298
299	async fn send_handle(
300	client: &Client,
301	room_id: &RoomId,
302	transaction_id: &TransactionId,
303	) -> Result<SendHandle, Error> {
304	let (echoes, _) = room(client, room_id)?
305	.send_queue()
306	.subscribe()
307	.await
308	.map_err(\|source\| {
309	Error::with_source(ErrorKind::Store, "reading send queue failed", source)
310	})?;
311	echoes
312	.into_iter()
313	.find_map(\|echo\| match echo.content {
314	LocalEchoContent::Event { send_handle, .. }
315	if echo.transaction_id.as_str() == transaction_id.as_str() =>
316	{
317	Some(send_handle)
318	}
319	_ => None,
320	})
321	.ok_or_else(\|\| {
322	Error::new(
323	ErrorKind::Invalid,
324	"send is no longer pending or is not a message",
325	)
326	})
327	}
328
329	pub(super) async fn cancel_send(
330	client: &Client,
331	room_id: &RoomId,
332	transaction_id: &TransactionId,
333	) -> Result<(), Error> {
334	match send_handle(client, room_id, transaction_id)
335	.await?
336	.abort()
337	.await
338	{
339	Ok(true) => Ok(()),
340	Ok(false) => Err(Error::new(
341	ErrorKind::Invalid,
342	"send could not be cancelled because it has already been sent",
343	)),
344	Err(source) => Err(Error::with_source(
345	ErrorKind::Store,
346	"cancelling send failed",
347	source,
348	)),
349	}
350	}
351
352	pub(super) async fn retry_send(
353	client: &Client,
354	room_id: &RoomId,
355	transaction_id: &TransactionId,
356	) -> Result<(), Error> {
357	send_handle(client, room_id, transaction_id)
358	.await?
359	.unwedge()
360	.await
361	.map_err(\|source\| Error::with_source(ErrorKind::Store, "retrying send failed", source))?;
362	room(client, room_id)?.send_queue().set_enabled(true);
363	Ok(())
364	}
365
366	pub(super) async fn upload_avatar(
367	client: &Client,
368	mimetype: &str,
369	bytes: Vec<u8>,
370	max_bytes: usize,
371	) -> Result<(), Error> {
372	if bytes.len() > max_bytes {
373	return Err(Error::new(
374	ErrorKind::MediaTooLarge,
375	"avatar exceeds the configured byte limit",
376	));
377	}
378	let mimetype: mime::Mime = mimetype
379	.parse()
380	.map_err(\|_\| Error::new(ErrorKind::Invalid, "avatar MIME type is invalid"))?;
381	if mimetype.type_() != mime::IMAGE {
382	return Err(Error::new(
383	ErrorKind::Invalid,
384	"an avatar has to be an image",
385	));
386	}
387	client
388	.account()
389	.upload_avatar(&mimetype, bytes)
390	.await
391	.map(drop)
392	.map_err(\|source\| Error::from_sdk("uploading avatar failed", source))
393	}
394
395	pub(super) async fn upload_media(
396	client: &Client,
397	mimetype: &str,
398	bytes: Vec<u8>,
399	max_bytes: usize,
400	) -> Result<OwnedMxcUri, Error> {
401	if bytes.len() > max_bytes {
402	return Err(Error::new(
403	ErrorKind::MediaTooLarge,
404	"media exceeds the configured byte limit",
405	));
406	}
407	let mimetype: mime::Mime = mimetype
408	.parse()
409	.map_err(\|_\| Error::new(ErrorKind::Invalid, "media MIME type is invalid"))?;
410	client
411	.media()
412	.upload(&mimetype, bytes, None)
413	.await
414	.map(\|response\| response.content_uri)
415	.map_err(\|source\| Error::from_sdk("uploading media failed", source))
416	}
417
418	pub(super) async fn rename_device(
419	client: &Client,
420	device_id: &OwnedDeviceId,
421	display_name: &str,
422	) -> Result<(), Error> {
423	client
424	.rename_device(device_id, display_name)
425	.await
426	.map(drop)
427	.map_err(\|source\| Error::from_http("renaming device failed", source))
428	}
429
430	pub(super) async fn delete_devices(
431	client: &Client,
432	device_ids: &[OwnedDeviceId],
433	auth: Option<InteractiveAuth>,
434	) -> Result<(), Error> {
435	let user_id = client
436	.user_id()
437	.ok_or_else(\|\| Error::new(ErrorKind::Unauthenticated, "the client is not logged in"))?;
438	let auth = auth
439	.map(\|auth\| auth.into_sdk(user_id))
440	.transpose()
441	.map_err(\|message\| {
442	Error::new(
443	ErrorKind::Invalid,
444	format!("interactive authentication data is invalid, {message}"),
445	)
446	})?;
447	match client.delete_devices(device_ids, auth).await {
448	Ok(_) => Ok(()),
449	Err(source) => match source.as_uiaa_response() {
450	Some(info) => Err(Error::interactive_auth(InteractiveAuthChallenge::from(
451	info,
452	))),
453	None => Err(Error::from_http("deleting devices failed", source)),
454	},
455	}
456	}