arminius_server_client/handlers/
to_device.rs

1// SPDX-FileCopyrightText: 2025 BWI GmbH
2//
3// SPDX-License-Identifier: Apache-2.0
4
5//! Implements sending to-device messages
6
7#![allow(
8	clippy::allow_attributes,
9	reason = "This source file contains enums which allow `clippy::missing_docs_in_private_items`. But an exception to this lint will mistakenly report that it's unfulfilled when used with expect, even though it is not."
10)]
11
12use std::collections::{BTreeMap, HashMap};
13
14use arminius_lib_component::{
15	Component, ServerState,
16	error::{InputJsonError, InternalServerError, MatrixApiError, ResultExt as _},
17	ruma_requests::ArminiusResponse,
18};
19use arminius_lib_db::Database as _;
20use arminius_lib_db_todevice::{ToDeviceMessage, ToDeviceQueries as _};
21use arminius_lib_macros::FromDbError;
22use arminius_server_federation_output::{clients::FederationNatsClient, types::EduRequest};
23use axum::{Router, extract::State, routing::put};
24use itertools::Itertools as _;
25use ruma::{
26	OwnedServerName, TransactionId,
27	api::{
28		client::to_device::send_event_to_device,
29		federation::transactions::edu::{DirectDeviceContent, Edu},
30	},
31	serde::Raw,
32};
33
34use crate::client_request::AuthenticatedClientRequest;
35
36#[allow(
37	missing_docs,
38	clippy::missing_docs_in_private_items,
39	reason = "TODO(#2003): This exception predates us enabling `clippy::allow_attributes_without_reason`, so we don't know why it was enabled"
40)]
41#[derive(thiserror::Error, Debug, MatrixApiError, FromDbError)]
42enum SendEventToDeviceError {
43	#[error(transparent)]
44	InternalServerError(#[from] InternalServerError),
45	#[error(transparent)]
46	InputJsonError(#[from] InputJsonError),
47}
48
49/// Handler for `PUT /_matrix/client/v3/sendToDevice/{eventType}/{txnId}` (See
50/// <https://spec.matrix.org/v1.15/client-server-api/#put_matrixclientv3sendtodeviceeventtypetxnid>)
51///
52/// Endpoint for sending events to one or more devices directly.
53async fn send_event_to_device_handler(
54	mut state: State<ServerState>,
55	AuthenticatedClientRequest(claims, request): AuthenticatedClientRequest<
56		send_event_to_device::v3::Request,
57	>,
58) -> Result<ArminiusResponse<send_event_to_device::v3::Response>, SendEventToDeviceError> {
59	// FIXME(#706): Check whether another request with the same `txnId` has already
60	// been made.
61	let (local_messages, remote_messages): (Vec<ToDeviceMessage>, Vec<ToDeviceMessage>) = request
62		.messages
63		.into_iter()
64		.flat_map(|(user, messages)| {
65			messages
66				.into_iter()
67				.map(|(device, message)| {
68					let message: serde_json::Value =
69						message.deserialize_as().map_err(InputJsonError::from)?;
70					Ok(ToDeviceMessage { user: user.clone(), device, message })
71				})
72				.collect::<Vec<Result<ToDeviceMessage, InputJsonError>>>()
73		})
74		.collect::<Result<Vec<_>, InputJsonError>>()?
75		.into_iter()
76		.partition(|el| el.user.server_name() == state.config.server_name);
77
78	let mut transaction = state.database.begin_transaction().await?;
79
80	let local_insert_result = transaction
81		.insert_to_device_messages(request.event_type.clone(), &claims.user_id, local_messages)
82		.await?;
83
84	transaction.commit().await?;
85
86	if local_insert_result.rows_affected() > 0 {
87		state
88			.nats_connection
89			.notify_new_sync_data()
90			.await
91			.unrecoverable_error("Failed to notify clients about potential new sync data")?;
92	}
93
94	let client = FederationNatsClient::new(state.nats_connection.client.clone());
95	let remote_messages: HashMap<OwnedServerName, Vec<_>> = remote_messages
96		.into_iter()
97		.map(|message| (message.user.server_name().to_owned(), message))
98		.into_group_map();
99	for (destination, messages) in remote_messages {
100		// FIXME(#2305): When the FOS supports proper to-device chunking logic, we should stop
101		// splitting up our EDUs like this
102		for message in messages {
103			let mut direct_device_content = DirectDeviceContent::new(
104				claims.user_id.clone(),
105				request.event_type.clone(),
106				TransactionId::new(),
107			);
108			direct_device_content.messages = BTreeMap::from([(
109				message.user,
110				BTreeMap::from([(
111					message.device,
112					Raw::new(&message.message).map_err(InputJsonError::from)?.cast_unchecked(),
113				)]),
114			)]);
115			let edu = Edu::DirectToDevice(direct_device_content);
116			let edu = Raw::new(&edu).unrecoverable_error("Could not serialize EDU")?;
117
118			client
119				.send_edu_message(&EduRequest { destination: destination.clone(), edu })
120				.await
121				.unrecoverable_error("Failed sending to-device message over NATS")?;
122		}
123	}
124
125	Ok(ArminiusResponse(send_event_to_device::v3::Response::new()))
126}
127
128/// Component for to-device paths
129#[derive(Debug)]
130pub struct ToDeviceComponent;
131
132impl Component for ToDeviceComponent {
133	fn router() -> Router<ServerState> {
134		Router::new().route(
135			"/_matrix/client/v3/sendToDevice/{eventType}/{txnId}",
136			put(send_event_to_device_handler),
137		)
138	}
139}
arminius_server_client/handlers/to_device.rs

arminius_server_client/handlers/
to_device.rs
