tuwunel_api_client/
to_device.rs

1use std::{collections::BTreeMap, iter::once};
2
3use axum::extract::State;
4use futures::StreamExt;
5use ruma::{
6	DeviceId, OwnedDeviceId,
7	api::{
8		client::to_device::send_event_to_device,
9		error::ErrorKind,
10		federation::{self, transactions::edu::DirectDeviceContent},
11	},
12	events::AnyToDeviceEventContent,
13	serde::Raw,
14	to_device::DeviceIdOrAllDevices,
15};
16use tuwunel_core::{
17	Err, Error, Result,
18	smallvec::SmallVec,
19	utils::{ReadyExt, result::LogErr},
20};
21use tuwunel_service::{sending::EduBuf, users::device::MAX_DEVICE_ID_LENGTH};
22
23use crate::Ruma;
24
25/// Recipient devices of one `AllDevices` to-device send paired with their
26/// inbox counts.
27type Deliveries = SmallVec<[(OwnedDeviceId, u64); 1]>;
28
29/// Largest serialized content of one to-device message.
30///
31/// Synapse refuses larger content with the same limit, measured the same way on
32/// canonical JSON. Together with the cap on recipient device IDs it keeps the
33/// EDU carrying a message for a remote device far inside a peer's request limit.
34const MAX_CONTENT_BYTES: usize = 65_536;
35
36/// # `PUT /_matrix/client/r0/sendToDevice/{eventType}/{txnId}`
37///
38/// Send a to-device event to a set of client devices.
39pub(crate) async fn send_event_to_device_route(
40	State(services): State<crate::State>,
41	body: Ruma<send_event_to_device::v3::Request>,
42) -> Result<send_event_to_device::v3::Response> {
43	let sender_user = body.sender_user();
44	let sender_device = body.sender_device.as_deref();
45
46	// Check if this is a new transaction id
47	if services
48		.transaction_ids
49		.existing_txnid(sender_user, sender_device, &body.txn_id)
50		.await
51		.is_ok()
52	{
53		return Ok(send_event_to_device::v3::Response {});
54	}
55
56	// Checked before any delivery, so a refused request queues nothing.
57	if body
58		.messages
59		.values()
60		.flat_map(BTreeMap::iter)
61		.any(too_large)
62	{
63		return Err!(Request(TooLarge("To-device message is too large.")));
64	}
65
66	for (target_user_id, map) in &body.messages {
67		for (target_device_id_maybe, event) in map {
68			if !services.globals.user_is_local(target_user_id) {
69				let mut map = BTreeMap::new();
70				map.insert(target_device_id_maybe.clone(), event.clone());
71				let mut messages = BTreeMap::new();
72				messages.insert(target_user_id.clone(), map);
73
74				let mut buf = EduBuf::new();
75				serde_json::to_writer(
76					&mut buf,
77					&federation::transactions::edu::Edu::DirectToDevice(DirectDeviceContent {
78						sender: sender_user.to_owned(),
79						ev_type: body.event_type.clone(),
80						message_id: services.globals.next_count().to_string().into(),
81						messages,
82					}),
83				)
84				.expect("DirectToDevice EDU can be serialized");
85
86				services
87					.sending
88					.send_edu_server(target_user_id.server_name(), buf)?;
89
90				continue;
91			}
92
93			let event_type = &body.event_type.to_string();
94
95			let event = event
96				.deserialize_as()
97				.map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "Event is invalid"))?;
98
99			match target_device_id_maybe {
100				| DeviceIdOrAllDevices::DeviceId(target_device_id) => {
101					let count = services.users.add_to_device_event(
102						sender_user,
103						target_user_id,
104						target_device_id,
105						event_type,
106						&event,
107					);
108
109					services
110						.sending
111						.send_to_device_appservices(
112							sender_user,
113							target_user_id,
114							once((&**target_device_id, count)),
115							event_type,
116							&event,
117						)
118						.await
119						.log_err()
120						.ok();
121				},
122
123				| DeviceIdOrAllDevices::AllDevices => {
124					let interested = services
125						.appservice
126						.is_interested_in_user(target_user_id)
127						.await;
128
129					let deliveries: Deliveries = services
130						.users
131						.all_device_ids(target_user_id)
132						.map(|target_device_id| {
133							let count = services.users.add_to_device_event(
134								sender_user,
135								target_user_id,
136								target_device_id,
137								event_type,
138								&event,
139							);
140
141							(target_device_id, count)
142						})
143						.ready_filter_map(|(target_device_id, count)| {
144							interested.then(|| (target_device_id.to_owned(), count))
145						})
146						.collect()
147						.await;
148
149					if !deliveries.is_empty() {
150						services
151							.sending
152							.send_to_device_appservices(
153								sender_user,
154								target_user_id,
155								deliveries
156									.iter()
157									.map(|(device_id, count)| (&**device_id, *count)),
158								event_type,
159								&event,
160							)
161							.await
162							.log_err()
163							.ok();
164					}
165				},
166			}
167		}
168	}
169
170	// Save transaction id with empty data
171	services
172		.transaction_ids
173		.add_txnid(sender_user, sender_device, &body.txn_id, &[]);
174
175	Ok(send_event_to_device::v3::Response {})
176}
177
178/// Whether one message is too large to queue.
179///
180/// Its content may not pass `MAX_CONTENT_BYTES`. No device can have an ID past
181/// `MAX_DEVICE_ID_LENGTH`, so neither may its recipient.
182fn too_large((target, event): (&DeviceIdOrAllDevices, &Raw<AnyToDeviceEventContent>)) -> bool {
183	let long_id = |device_id: &DeviceId| device_id.as_str().len() > MAX_DEVICE_ID_LENGTH;
184
185	event.json().get().len() > MAX_CONTENT_BYTES
186		|| matches!(target, DeviceIdOrAllDevices::DeviceId(device_id) if long_id(device_id))
187}
tuwunel_api_client/to_device.rs

tuwunel_api_client/
to_device.rs
