tuwunel_api/client/
to_device.rs

1use std::{collections::BTreeMap, iter::once};
2
3use axum::extract::State;
4use futures::StreamExt;
5use ruma::{
6	OwnedDeviceId,
7	api::{
8		client::to_device::send_event_to_device,
9		error::ErrorKind,
10		federation::{self, transactions::edu::DirectDeviceContent},
11	},
12	to_device::DeviceIdOrAllDevices,
13};
14use tuwunel_core::{
15	Error, Result,
16	smallvec::SmallVec,
17	utils::{ReadyExt, result::LogErr},
18};
19use tuwunel_service::sending::EduBuf;
20
21use crate::Ruma;
22
23/// Recipient devices of one `AllDevices` to-device send paired with their
24/// inbox counts.
25type Deliveries = SmallVec<[(OwnedDeviceId, u64); 1]>;
26
27/// # `PUT /_matrix/client/r0/sendToDevice/{eventType}/{txnId}`
28///
29/// Send a to-device event to a set of client devices.
30pub(crate) async fn send_event_to_device_route(
31	State(services): State<crate::State>,
32	body: Ruma<send_event_to_device::v3::Request>,
33) -> Result<send_event_to_device::v3::Response> {
34	let sender_user = body.sender_user();
35	let sender_device = body.sender_device.as_deref();
36
37	// Check if this is a new transaction id
38	if services
39		.transaction_ids
40		.existing_txnid(sender_user, sender_device, &body.txn_id)
41		.await
42		.is_ok()
43	{
44		return Ok(send_event_to_device::v3::Response {});
45	}
46
47	for (target_user_id, map) in &body.messages {
48		for (target_device_id_maybe, event) in map {
49			if !services.globals.user_is_local(target_user_id) {
50				let mut map = BTreeMap::new();
51				map.insert(target_device_id_maybe.clone(), event.clone());
52				let mut messages = BTreeMap::new();
53				messages.insert(target_user_id.clone(), map);
54
55				let mut buf = EduBuf::new();
56				serde_json::to_writer(
57					&mut buf,
58					&federation::transactions::edu::Edu::DirectToDevice(DirectDeviceContent {
59						sender: sender_user.to_owned(),
60						ev_type: body.event_type.clone(),
61						message_id: services.globals.next_count().to_string().into(),
62						messages,
63					}),
64				)
65				.expect("DirectToDevice EDU can be serialized");
66
67				services
68					.sending
69					.send_edu_server(target_user_id.server_name(), buf)?;
70
71				continue;
72			}
73
74			let event_type = &body.event_type.to_string();
75
76			let event = event
77				.deserialize_as()
78				.map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "Event is invalid"))?;
79
80			match target_device_id_maybe {
81				| DeviceIdOrAllDevices::DeviceId(target_device_id) => {
82					let count = services.users.add_to_device_event(
83						sender_user,
84						target_user_id,
85						target_device_id,
86						event_type,
87						&event,
88					);
89
90					services
91						.sending
92						.send_to_device_appservices(
93							sender_user,
94							target_user_id,
95							once((&**target_device_id, count)),
96							event_type,
97							&event,
98						)
99						.await
100						.log_err()
101						.ok();
102				},
103
104				| DeviceIdOrAllDevices::AllDevices => {
105					let interested = services
106						.appservice
107						.is_interested_in_user(target_user_id)
108						.await;
109
110					let deliveries: Deliveries = services
111						.users
112						.all_device_ids(target_user_id)
113						.map(|target_device_id| {
114							let count = services.users.add_to_device_event(
115								sender_user,
116								target_user_id,
117								target_device_id,
118								event_type,
119								&event,
120							);
121
122							(target_device_id, count)
123						})
124						.ready_filter_map(|(target_device_id, count)| {
125							interested.then(|| (target_device_id.to_owned(), count))
126						})
127						.collect()
128						.await;
129
130					if !deliveries.is_empty() {
131						services
132							.sending
133							.send_to_device_appservices(
134								sender_user,
135								target_user_id,
136								deliveries
137									.iter()
138									.map(|(device_id, count)| (&**device_id, *count)),
139								event_type,
140								&event,
141							)
142							.await
143							.log_err()
144							.ok();
145					}
146				},
147			}
148		}
149	}
150
151	// Save transaction id with empty data
152	services
153		.transaction_ids
154		.add_txnid(sender_user, sender_device, &body.txn_id, &[]);
155
156	Ok(send_event_to_device::v3::Response {})
157}
tuwunel_api/client/to_device.rs

tuwunel_api/client/
to_device.rs
