tuwunel_api_client/sync/v5/extensions/
to_device.rs

1use futures::StreamExt;
2use ruma::{
3	DeviceId, UserId,
4	api::client::sync::sync_events::v5::{request::ToDevice, response},
5};
6use tuwunel_core::{
7	Err, Result, err,
8	utils::{BoolExt, math::usize_from_ruma_bounded, stream::ReadyExt},
9};
10use tuwunel_service::Services;
11
12use super::{Connection, SyncInfo};
13
14const DEFAULT_LIMIT: usize = 100;
15
16pub(in crate::sync::v5) fn check(request: &ToDevice) -> Result {
17	if request
18		.limit
19		.is_some_and(|limit| u64::from(limit) == 0)
20	{
21		return Err!(Request(BadJson("To-device limit must be positive.")));
22	}
23
24	since(request).map(|_| ())
25}
26
27#[tracing::instrument(level = "trace", skip_all)]
28pub(in crate::sync::v5) async fn acknowledge(
29	services: &Services,
30	sender_user: &UserId,
31	sender_device: Option<&DeviceId>,
32	conn: &Connection,
33) -> Result {
34	let Some(sender_device) =
35		sender_device.filter(|_| conn.extensions.to_device.enabled.unwrap_or(false))
36	else {
37		return Ok(());
38	};
39
40	// MSC4538 acknowledgements use the extension's token independently of pos.
41	let Some(since) = since(&conn.extensions.to_device)? else {
42		return Ok(());
43	};
44
45	// MSC4538:59-60 deviation: an unissued token returns and deletes nothing;
46	// clients treat 400 as permanent.
47	if since > conn.next_batch {
48		return Ok(());
49	}
50
51	services
52		.users
53		.remove_to_device_events(sender_user, sender_device, since)
54		.await;
55
56	Ok(())
57}
58
59#[tracing::instrument(name = "to_device", level = "trace", skip_all, ret)]
60pub(super) async fn collect(
61	SyncInfo { services, sender_user, sender_device, .. }: SyncInfo<'_>,
62	conn: &Connection,
63) -> Result<Option<response::ToDevice>> {
64	let Some(sender_device) = sender_device else {
65		return Ok(None);
66	};
67
68	let since = since(&conn.extensions.to_device)?;
69
70	let (next_batch, events) = services
71		.users
72		.get_to_device_events(sender_user, sender_device, since, Some(conn.next_batch))
73		.take(limit(&conn.extensions.to_device))
74		.ready_fold((0, Vec::new()), |(_, mut events), (count, event)| {
75			events.push(event);
76			(count, events)
77		})
78		.await;
79
80	let to_device = events
81		.is_empty()
82		.is_false()
83		.then(|| response::ToDevice {
84			next_batch: next_batch.to_string().into(),
85			events,
86		});
87
88	Ok(to_device)
89}
90
91pub(super) fn limit(config: &ToDevice) -> usize {
92	config
93		.limit
94		.map_or(DEFAULT_LIMIT, |limit| usize_from_ruma_bounded(limit, usize::MAX, usize::MAX))
95}
96
97#[tracing::instrument(level = "trace", skip_all)]
98pub(in crate::sync::v5) async fn pending(
99	services: &Services,
100	sender_user: &UserId,
101	sender_device: Option<&DeviceId>,
102	conn: &Connection,
103) -> Result<bool> {
104	let Some(sender_device) =
105		sender_device.filter(|_| conn.extensions.to_device.enabled.unwrap_or(false))
106	else {
107		return Ok(false);
108	};
109
110	let pending = services
111		.users
112		.get_to_device_events(
113			sender_user,
114			sender_device,
115			since(&conn.extensions.to_device)?,
116			Some(conn.next_batch),
117		)
118		.ready_any(|_| true)
119		.await;
120
121	Ok(pending)
122}
123
124fn since(config: &ToDevice) -> Result<Option<u64>> {
125	config
126		.since
127		.as_deref()
128		.map(str::parse)
129		.transpose()
130		.map_err(|_| err!(Request(InvalidParam("Unparsable to-device since."))))
131}
tuwunel_api_client/sync/v5/extensions/to_device.rs

tuwunel_api_client/sync/v5/extensions/
to_device.rs
