tuwunel_service/sending/
device.rs

1use std::iter::once;
2
3use futures::StreamExt;
4use ruma::{DeviceId, OwnedRoomId, UserId};
5use serde::Serialize;
6use tuwunel_core::{Result, implement, utils::stream::BroadbandExt};
7
8use super::{Destination, EduBuf, SendingEvent, Service, TAG_DEVICE_LIST_CHANGED, TAG_TO_DEVICE};
9use crate::appservice::RegistrationInfo;
10
11/// One recipient device of a to-device event paired with its inbox count.
12///
13/// The count uniquifies the appservice transaction hash.
14type Delivery<'a> = (&'a DeviceId, u64);
15
16/// Wire shape of one `de.sorunome.msc2409.to_device` entry (MSC4203).
17///
18/// The stored to-device event is flattened with the recipient's identifiers.
19/// The ruma `AnyAppserviceToDeviceEvent` deliberately has no `Serialize`, so
20/// the send side writes this local struct.
21#[derive(Serialize)]
22struct AsToDeviceEvent<'a> {
23	#[serde(rename = "type")]
24	kind: &'a str,
25	sender: &'a UserId,
26	content: &'a serde_json::Value,
27	to_user_id: &'a UserId,
28	to_device_id: &'a DeviceId,
29}
30
31/// Queue stored to-device events for delivery to interested appservices
32/// (MSC4203).
33///
34/// `deliveries` are the concrete recipient devices already written to the
35/// inbox, after `AllDevices` expansion. Each delivery becomes one queue row
36/// per interested appservice, written under one cork; with no interested
37/// appservice nothing is serialized.
38#[implement(Service)]
39#[tracing::instrument(
40	skip(self, deliveries, content),
41	level = "debug",
42	fields(
43		%target_user,
44	),
45)]
46pub async fn send_to_device_appservices<'a, I>(
47	&self,
48	sender: &'a UserId,
49	target_user: &'a UserId,
50	deliveries: I,
51	event_type: &'a str,
52	content: &'a serde_json::Value,
53) -> Result
54where
55	I: Iterator<Item = Delivery<'a>> + Send + 'a,
56{
57	let registrations = self.services.appservice.read().await;
58	let interested = || {
59		registrations
60			.values()
61			.filter(|info| info.is_user_match(target_user))
62	};
63
64	// Serialize once, and only when an appservice will receive the events.
65	if interested().next().is_none() {
66		return Ok(());
67	}
68
69	let _cork = self.db.db.cork();
70
71	to_device_payloads(sender, target_user, deliveries, event_type, content)
72		.flat_map(|buf| interested().map(move |info| (info, buf.clone())))
73		.try_for_each(|(info, buf)| {
74			let dest = Destination::Appservice(info.registration.id.clone());
75
76			self.queue_and_dispatch(dest, SendingEvent::ToDevice(buf))
77		})
78}
79
80/// Queue a `device_lists.changed` marker (MSC3202) for delivery to
81/// appservices that opted into transaction extensions and are interested
82/// in `user_id`.
83///
84/// The caller passes the count it already allocated so the marker uniquifies
85/// the transaction hash.
86#[implement(Service)]
87#[tracing::instrument(
88	skip(self),
89	level = "debug",
90	fields(
91		%user_id,
92	),
93)]
94pub async fn send_device_list_appservices(&self, user_id: &UserId, count: u64) -> Result {
95	let registrations = self.services.appservice.read().await;
96	let extended = || {
97		registrations
98			.values()
99			.filter(|info| info.registration.msc3202_transaction_extensions)
100	};
101
102	if extended().next().is_none() {
103		return Ok(());
104	}
105
106	let payload = device_list_payload(user_id, count);
107	let _cork = self.db.db.cork();
108
109	for info in extended() {
110		if !info.is_user_match(user_id) && !self.shares_device_list_room(user_id, info).await {
111			continue;
112		}
113
114		let dest = Destination::Appservice(info.registration.id.clone());
115
116		self.queue_and_dispatch(dest, SendingEvent::DeviceListChanged(payload.clone()))?;
117	}
118
119	Ok(())
120}
121
122/// Whether `user_id` shares a device-list-interesting room with `info`.
123///
124/// A joined room the appservice participates in counts when it is encrypted,
125/// or unconditionally when `device_key_update_encrypted_rooms_only` is off.
126#[implement(Service)]
127async fn shares_device_list_room(&self, user_id: &UserId, info: &RegistrationInfo) -> bool {
128	let update_all_rooms = !self
129		.services
130		.config
131		.device_key_update_encrypted_rooms_only;
132
133	self.services
134		.state_cache
135		.rooms_joined(user_id)
136		.map(ToOwned::to_owned)
137		.broad_any(async |room_id: OwnedRoomId| {
138			if !update_all_rooms
139				&& !self
140					.services
141					.state_accessor
142					.is_encrypted_room(&room_id)
143					.await
144			{
145				return false;
146			}
147
148			self.services
149				.state_cache
150				.appservice_in_room(&room_id, info)
151				.await
152		})
153		.await
154}
155
156fn to_device_payloads<'a>(
157	sender: &'a UserId,
158	target_user: &'a UserId,
159	deliveries: impl Iterator<Item = Delivery<'a>> + 'a,
160	event_type: &'a str,
161	content: &'a serde_json::Value,
162) -> impl Iterator<Item = EduBuf> + 'a {
163	deliveries.map(move |(to_device_id, count)| {
164		let event = AsToDeviceEvent {
165			kind: event_type,
166			sender,
167			content,
168			to_user_id: target_user,
169			to_device_id,
170		};
171
172		tagged_json(TAG_TO_DEVICE, count, &event)
173	})
174}
175
176/// A tagged queue row value whose body is `value` serialized as JSON.
177///
178/// The body is written straight into the inline buffer after the prefix.
179fn tagged_json(tag: u8, count: u64, value: &impl Serialize) -> EduBuf {
180	let mut buf = tag_prefix(tag, count); // serde_json::to_writer out-param
181
182	serde_json::to_writer(&mut buf, value).expect("tagged queue row value serializes");
183	buf
184}
185
186fn device_list_payload(user_id: &UserId, count: u64) -> EduBuf {
187	let mut buf = tag_prefix(TAG_DEVICE_LIST_CHANGED, count);
188
189	buf.extend_from_slice(user_id.as_bytes());
190	buf
191}
192
193/// The `[tag][count]` head every tagged queue row value starts with.
194///
195/// The count is written big-endian so the prefix is `TAG_PREFIX_LEN` bytes.
196fn tag_prefix(tag: u8, count: u64) -> EduBuf { once(tag).chain(count.to_be_bytes()).collect() }
tuwunel_service/sending/device.rs

tuwunel_service/sending/
device.rs
