macltok

public

Star0

Watch0

78 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, NotificationSettings, RoomNotificationMode as SdkMode,
    },
    ruma::{
        OwnedDeviceId, OwnedEventId, OwnedMxcUri, RoomId, TransactionId,
        api::client::{push::set_pushrule_actions, receipt::create_receipt::v3::ReceiptType},
        events::receipt::ReceiptThread,
        push::{Action, HighlightTweakValue, RuleKind, SoundTweakValue, Tweak},
    },
    send_queue::{LocalEchoContent, SendHandle},
};
use tokio::sync::mpsc;

use crate::{
    EncryptionCommand, Error, ErrorKind, InteractiveAuth, InteractiveAuthChallenge,
    NotificationCommand, 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>,
) {
    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, command);
                reply.done(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,
    command: EncryptionCommand,
) -> Result<(), Error> {
    let encryption = client.encryption();
    match command {
        EncryptionCommand::Recover(key) => encryption
            .recovery()
            .recover(&key)
            .await
            .map_err(|source| Error::from_recovery("account recovery failed", source)),
        EncryptionCommand::RecoverKeyBackup(key) => encryption
            .recovery()
            .recover_and_fix_backup(&key)
            .await
            .map_err(|source| Error::from_recovery("recovering key backup failed", source)),
        EncryptionCommand::CreateKeyBackup => encryption
            .backups()
            .create()
            .await
            .map_err(|source| Error::from_sdk("creating key backup failed", source)),
        EncryptionCommand::EnableKeyBackup => encryption
            .recovery()
            .enable_backup()
            .await
            .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_err(|source| Error::from_sdk("disabling key backup failed", source))
        }
        EncryptionCommand::StartVerification(target) => {
            verification.start(client, events.clone(), target).await
        }
        EncryptionCommand::Verify(action) => verification.apply(events.clone(), action).await,
    }
}

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, &settings, &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
async fn highlight_keyword(
    client: &Client,
    settings: &NotificationSettings,
    keyword: &str,
) -> Result<(), Error> {
    let ruleset = settings.ruleset().await;
    for rule in ruleset
        .content
        .iter()
        .filter(|rule| !rule.default && rule.pattern == keyword)
        .filter(|rule| !rule.actions.iter().any(Action::is_highlight))
    {
        client
            .send(set_pushrule_actions::v3::Request::new(
                RuleKind::Content,
                rule.rule_id.clone(),
                keyword_actions(&rule.actions),
            ))
            .await
            .map_err(|source| {
                Error::from_sdk("highlighting a notification keyword failed", source.into())
            })?;
    }
    Ok(())
}

fn keyword_actions(existing: &[Action]) -> Vec<Action> {
    let mut actions: Vec<_> = existing
        .iter()
        .filter(|action| !matches!(action, Action::SetTweak(Tweak::Highlight(_))))
        .cloned()
        .collect();
    if !actions
        .iter()
        .any(|action| matches!(action, Action::Notify))
    {
        actions.insert(0, Action::Notify);
    }
    if !actions
        .iter()
        .any(|action| matches!(action, Action::SetTweak(Tweak::Sound(_))))
    {
        actions.push(Action::SetTweak(Tweak::Sound(SoundTweakValue::Default)));
    }
    actions.push(Action::SetTweak(Tweak::Highlight(HighlightTweakValue::Yes)));
    actions
}

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