tuwunel_api_mas/
sync_devices.rs

1use axum::extract::State;
2use futures::{StreamExt, TryStreamExt};
3use ruma::OwnedDeviceId;
4use synapse_admin_api::mas::sync_devices::{Request, Response};
5use tuwunel_core::{
6	Result,
7	utils::stream::{IterStream, TryBroadbandExt, automatic_width},
8};
9
10use super::{Mas, existing_user};
11use crate::Ruma;
12
13/// # `POST /_synapse/mas/sync_devices`
14pub(crate) async fn sync_devices_route(
15	_mas: Mas,
16	State(services): State<crate::State>,
17	body: Ruma<Request>,
18) -> Result<Response> {
19	let user_id = existing_user(services, &body.localpart).await?;
20
21	let current: Vec<OwnedDeviceId> = services
22		.users
23		.all_device_ids(&user_id)
24		.map(ToOwned::to_owned)
25		.collect()
26		.await;
27
28	let added = || {
29		body.devices
30			.iter()
31			.filter(|device_id| !current.contains(*device_id))
32	};
33
34	// Checked before any removal, so a refused addition leaves every device.
35	added()
36		.try_stream()
37		.broad_and_then(async |device_id| {
38			services
39				.users
40				.check_device_id(&user_id, Some(device_id.as_ref()))
41				.await
42		})
43		.try_collect::<()>()
44		.await?;
45
46	current
47		.iter()
48		.filter(|device_id| !body.devices.contains(*device_id))
49		.stream()
50		.for_each_concurrent(automatic_width(), async |device_id| {
51			services
52				.users
53				.remove_device(&user_id, device_id)
54				.await;
55		})
56		.await;
57
58	added()
59		.try_stream()
60		.broad_and_then(async |device_id| {
61			services
62				.users
63				.create_device(&user_id, Some(device_id.as_ref()), (None, None), None, None, None)
64				.await
65				.map(|_| ())
66		})
67		.try_collect::<()>()
68		.await?;
69
70	Ok(Response::new())
71}
tuwunel_api_mas/sync_devices.rs

tuwunel_api_mas/
sync_devices.rs
