macltok

public

Star0

Watch0

4 commits 3 branches 0 tags 0 releases3.0 MiB

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

use futures_util::FutureExt;
use matrix_sdk::{
    Client,
    notification_settings::{IsEncrypted, IsOneToOne, RoomNotificationMode as SdkMode},
    ruma::{
        OwnedDeviceId, OwnedEventId, RoomId, TransactionId,
        api::client::receipt::create_receipt::v3::ReceiptType, events::receipt::ReceiptThread,
    },
    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>>,
) {
    let mut verification = VerificationController::default();
    while let Some(Job { command, reply }) = jobs.recv().await {
        let work = encryption(&client, &events, &mut verification, command);
        reply.done(guarded("encryption command", work).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) => settings.add_keyword(keyword).await,
        NotificationCommand::RemoveKeyword(keyword) => settings.remove_keyword(&keyword).await,
    }
    .map_err(|source| {
        Error::with_source(
            ErrorKind::Server,
            "updating notification settings failed",
            source,
        )
    })
}

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 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::{
5	Client,
6	notification_settings::{IsEncrypted, IsOneToOne, RoomNotificationMode as SdkMode},
7	ruma::{
8	OwnedDeviceId, OwnedEventId, RoomId, TransactionId,
9	api::client::receipt::create_receipt::v3::ReceiptType, events::receipt::ReceiptThread,
10	},
11	send_queue::{LocalEchoContent, SendHandle},
12	};
13	use tokio::sync::mpsc;
14
15	use crate::{
16	EncryptionCommand, Error, ErrorKind, InteractiveAuth, InteractiveAuthChallenge,
17	NotificationCommand, RoomNotificationMode, TimelineId,
18	encryption::VerificationController,
19	transport::{EventSender, Responder},
20	};
21
22	pub(super) struct Job<T> {
23	pub(super) command: T,
24	pub(super) reply: Responder,
25	}
26
27	/// a panicking command answers with an error instead of taking its worker down with it
28	pub(super) async fn guarded<T>(
29	label: &'static str,
30	work: impl Future<Output = Result<T, Error>>,
31	) -> Result<T, Error> {
32	AssertUnwindSafe(work)
33	.catch_unwind()
34	.await
35	.unwrap_or_else(\|_\| Err(Error::new(ErrorKind::Other, format!("{label} panicked"))))
36	}
37
38	pub(super) async fn run_notifications(
39	client: Client,
40	mut jobs: mpsc::Receiver<Job<NotificationCommand>>,
41	) {
42	while let Some(Job { command, reply }) = jobs.recv().await {
43	reply.done(guarded("notification command", notification(&client, command)).await);
44	}
45	}
46
47	pub(super) async fn run_encryption(
48	client: Client,
49	events: EventSender,
50	mut jobs: mpsc::Receiver<Job<EncryptionCommand>>,
51	) {
52	let mut verification = VerificationController::default();
53	while let Some(Job { command, reply }) = jobs.recv().await {
54	let work = encryption(&client, &events, &mut verification, command);
55	reply.done(guarded("encryption command", work).await);
56	}
57	}
58
59	async fn encryption(
60	client: &Client,
61	events: &EventSender,
62	verification: &mut VerificationController,
63	command: EncryptionCommand,
64	) -> Result<(), Error> {
65	let encryption = client.encryption();
66	match command {
67	EncryptionCommand::Recover(key) => encryption
68	.recovery()
69	.recover(&key)
70	.await
71	.map_err(\|source\| Error::from_recovery("account recovery failed", source)),
72	EncryptionCommand::RecoverKeyBackup(key) => encryption
73	.recovery()
74	.recover_and_fix_backup(&key)
75	.await
76	.map_err(\|source\| Error::from_recovery("recovering key backup failed", source)),
77	EncryptionCommand::CreateKeyBackup => encryption
78	.backups()
79	.create()
80	.await
81	.map_err(\|source\| Error::from_sdk("creating key backup failed", source)),
82	EncryptionCommand::EnableKeyBackup => encryption
83	.recovery()
84	.enable_backup()
85	.await
86	.map_err(\|source\| Error::from_recovery("enabling key backup failed", source)),
87	EncryptionCommand::DisableKeyBackup { delete_from_server } => {
88	let backups = encryption.backups();
89	if delete_from_server {
90	backups.disable_and_delete().await
91	} else {
92	backups.disable().await
93	}
94	.map_err(\|source\| Error::from_sdk("disabling key backup failed", source))
95	}
96	EncryptionCommand::StartVerification(target) => {
97	verification.start(client, events.clone(), target).await
98	}
99	EncryptionCommand::Verify(action) => verification.apply(events.clone(), action).await,
100	}
101	}
102
103	async fn notification(client: &Client, command: NotificationCommand) -> Result<(), Error> {
104	let settings = client.notification_settings().await;
105	match command {
106	NotificationCommand::SetRoomMode { room_id, mode } => {
107	settings
108	.set_room_notification_mode(&room_id, sdk_mode(mode))
109	.await
110	}
111	NotificationCommand::ResetRoomMode(room_id) => {
112	settings.delete_user_defined_room_rules(&room_id).await
113	}
114	NotificationCommand::SetDefaultMode {
115	encrypted,
116	one_to_one,
117	mode,
118	} => {
119	settings
120	.set_default_room_notification_mode(
121	IsEncrypted::from(encrypted),
122	IsOneToOne::from(one_to_one),
123	sdk_mode(mode),
124	)
125	.await
126	}
127	NotificationCommand::AddKeyword(keyword) => settings.add_keyword(keyword).await,
128	NotificationCommand::RemoveKeyword(keyword) => settings.remove_keyword(&keyword).await,
129	}
130	.map_err(\|source\| {
131	Error::with_source(
132	ErrorKind::Server,
133	"updating notification settings failed",
134	source,
135	)
136	})
137	}
138
139	const fn sdk_mode(mode: RoomNotificationMode) -> SdkMode {
140	match mode {
141	RoomNotificationMode::AllMessages => SdkMode::AllMessages,
142	RoomNotificationMode::MentionsAndKeywordsOnly => SdkMode::MentionsAndKeywordsOnly,
143	RoomNotificationMode::Mute => SdkMode::Mute,
144	}
145	}
146
147	fn room(client: &Client, room_id: &RoomId) -> Result<matrix_sdk::Room, Error> {
148	client.get_room(room_id).ok_or_else(Error::room_unknown)
149	}
150
151	/// the closed-room path. without a timeline nothing stops the marker going backwards
152	pub(super) async fn mark_read(
153	client: &Client,
154	timeline_id: &TimelineId,
155	event_id: OwnedEventId,
156	) -> Result<(), Error> {
157	let room = room(client, timeline_id.room_id())?;
158	let receipt_failed = \|source\| Error::from_sdk("sending read receipt failed", source);
159	let thread = match timeline_id.thread_root() {
160	None => ReceiptThread::Main,
161	Some(root) => ReceiptThread::Thread(root.clone()),
162	};
163	room.send_single_receipt(ReceiptType::Read, thread, event_id.clone())
164	.await
165	.map_err(receipt_failed)?;
166	if timeline_id.thread_root().is_none() {
167	room.send_single_receipt(ReceiptType::FullyRead, ReceiptThread::Unthreaded, event_id)
168	.await
169	.map_err(receipt_failed)?;
170	}
171	Ok(())
172	}
173
174	pub(super) async fn set_typing(
175	client: &Client,
176	room_id: &RoomId,
177	typing: bool,
178	) -> Result<(), Error> {
179	room(client, room_id)?
180	.typing_notice(typing)
181	.await
182	.map_err(\|source\| Error::from_sdk("typing update failed", source))
183	}
184
185	async fn send_handle(
186	client: &Client,
187	room_id: &RoomId,
188	transaction_id: &TransactionId,
189	) -> Result<SendHandle, Error> {
190	let (echoes, _) = room(client, room_id)?
191	.send_queue()
192	.subscribe()
193	.await
194	.map_err(\|source\| {
195	Error::with_source(ErrorKind::Store, "reading send queue failed", source)
196	})?;
197	echoes
198	.into_iter()
199	.find_map(\|echo\| match echo.content {
200	LocalEchoContent::Event { send_handle, .. }
201	if echo.transaction_id.as_str() == transaction_id.as_str() =>
202	{
203	Some(send_handle)
204	}
205	_ => None,
206	})
207	.ok_or_else(\|\| {
208	Error::new(
209	ErrorKind::Invalid,
210	"send is no longer pending or is not a message",
211	)
212	})
213	}
214
215	pub(super) async fn cancel_send(
216	client: &Client,
217	room_id: &RoomId,
218	transaction_id: &TransactionId,
219	) -> Result<(), Error> {
220	match send_handle(client, room_id, transaction_id)
221	.await?
222	.abort()
223	.await
224	{
225	Ok(true) => Ok(()),
226	Ok(false) => Err(Error::new(
227	ErrorKind::Invalid,
228	"send could not be cancelled because it has already been sent",
229	)),
230	Err(source) => Err(Error::with_source(
231	ErrorKind::Store,
232	"cancelling send failed",
233	source,
234	)),
235	}
236	}
237
238	pub(super) async fn retry_send(
239	client: &Client,
240	room_id: &RoomId,
241	transaction_id: &TransactionId,
242	) -> Result<(), Error> {
243	send_handle(client, room_id, transaction_id)
244	.await?
245	.unwedge()
246	.await
247	.map_err(\|source\| Error::with_source(ErrorKind::Store, "retrying send failed", source))?;
248	room(client, room_id)?.send_queue().set_enabled(true);
249	Ok(())
250	}
251
252	pub(super) async fn upload_avatar(
253	client: &Client,
254	mimetype: &str,
255	bytes: Vec<u8>,
256	max_bytes: usize,
257	) -> Result<(), Error> {
258	if bytes.len() > max_bytes {
259	return Err(Error::new(
260	ErrorKind::MediaTooLarge,
261	"avatar exceeds the configured byte limit",
262	));
263	}
264	let mimetype: mime::Mime = mimetype
265	.parse()
266	.map_err(\|_\| Error::new(ErrorKind::Invalid, "avatar MIME type is invalid"))?;
267	if mimetype.type_() != mime::IMAGE {
268	return Err(Error::new(
269	ErrorKind::Invalid,
270	"an avatar has to be an image",
271	));
272	}
273	client
274	.account()
275	.upload_avatar(&mimetype, bytes)
276	.await
277	.map(drop)
278	.map_err(\|source\| Error::from_sdk("uploading avatar failed", source))
279	}
280
281	pub(super) async fn rename_device(
282	client: &Client,
283	device_id: &OwnedDeviceId,
284	display_name: &str,
285	) -> Result<(), Error> {
286	client
287	.rename_device(device_id, display_name)
288	.await
289	.map(drop)
290	.map_err(\|source\| Error::from_http("renaming device failed", source))
291	}
292
293	pub(super) async fn delete_devices(
294	client: &Client,
295	device_ids: &[OwnedDeviceId],
296	auth: Option<InteractiveAuth>,
297	) -> Result<(), Error> {
298	let user_id = client
299	.user_id()
300	.ok_or_else(\|\| Error::new(ErrorKind::Unauthenticated, "the client is not logged in"))?;
301	let auth = auth
302	.map(\|auth\| auth.into_sdk(user_id))
303	.transpose()
304	.map_err(\|message\| {
305	Error::new(
306	ErrorKind::Invalid,
307	format!("interactive authentication data is invalid, {message}"),
308	)
309	})?;
310	match client.delete_devices(device_ids, auth).await {
311	Ok(_) => Ok(()),
312	Err(source) => match source.as_uiaa_response() {
313	Some(info) => Err(Error::interactive_auth(InteractiveAuthChallenge::from(
314	info,
315	))),
316	None => Err(Error::from_http("deleting devices failed", source)),
317	},
318	}
319	}