tuwunel_service/account_data/
mod.rs

1mod direct;
2mod push_rules;
3mod room_tags;
4
5use std::sync::Arc;
6
7use futures::{Stream, StreamExt, TryFutureExt, pin_mut};
8use ruma::{
9	RoomId, UserId,
10	events::{
11		AnyGlobalAccountDataEvent, AnyRawAccountDataEvent, AnyRoomAccountDataEvent,
12		GlobalAccountDataEventType, RoomAccountDataEventType,
13	},
14	push::{RuleKind, Ruleset},
15	serde::Raw,
16};
17use serde::Deserialize;
18use serde_json::json;
19use tuwunel_core::{
20	Err, Result, at, err, implement,
21	utils::{ReadyExt, TryReadyExt, result::LogErr, stream::TryIgnore},
22};
23use tuwunel_database::{Deserialized, Handle, Ignore, Interfix, Json, Map};
24
25/// Maximum number of push rules one account may hold.
26///
27/// The ruleset is a single account-data blob rewritten in full on every
28/// mutation and matched against every event for every local recipient, so the
29/// count bounds the write cost and the per-event matching work alike.
30pub const MAX_RULES: usize = 10_000;
31
32/// Longest rule ID stored, in bytes.
33///
34/// Room IDs reach 255 bytes and are routinely used as rule IDs, so the ceiling
35/// sits above that rather than at it.
36pub const MAX_RULE_ID_BYTES: usize = 300;
37
38/// Largest match and action data stored for one rule, in bytes.
39///
40/// Rule IDs are excluded and bounded separately by [`MAX_RULE_ID_BYTES`].
41pub const MAX_RULE_BYTES: usize = 1024;
42
43pub struct Service {
44	services: Arc<crate::services::OnceServices>,
45	db: Data,
46}
47
48struct Data {
49	roomuserdataid_accountdata: Arc<Map>,
50	roomusertype_roomuserdataid: Arc<Map>,
51}
52
53impl crate::Service for Service {
54	fn build(args: &crate::Args<'_>) -> Result<Arc<Self>> {
55		Ok(Arc::new(Self {
56			services: args.services.clone(),
57			db: Data {
58				roomuserdataid_accountdata: args.db["roomuserdataid_accountdata"].clone(),
59				roomusertype_roomuserdataid: args.db["roomusertype_roomuserdataid"].clone(),
60			},
61		}))
62	}
63
64	fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
65}
66
67/// Whether a ruleset has room for the rule with the given kind and ID.
68///
69/// A rule replacing one already present adds nothing, so only an unseen ID is
70/// held against [`MAX_RULES`]. The ID is bounded here rather than at each call
71/// site because a room ID serves as the rule ID for room rules and carries no
72/// length of its own once arbitrary-length identifiers are accepted.
73#[must_use]
74pub fn admits_rule(ruleset: &Ruleset, kind: RuleKind, rule_id: &str) -> bool {
75	rule_id.len() <= MAX_RULE_ID_BYTES
76		&& (ruleset.get(kind, rule_id).is_some()
77			|| ruleset.iter().take(MAX_RULES).count() < MAX_RULES)
78}
79
80/// Places one event in the account data of the user and removes the
81/// previous entry.
82#[implement(Service)]
83pub async fn update(
84	&self,
85	room_id: Option<&RoomId>,
86	user_id: &UserId,
87	event_type: RoomAccountDataEventType,
88	data: &serde_json::Value,
89) -> Result {
90	if data.get("type").is_none() || data.get("content").is_none() {
91		return Err!(Request(InvalidParam("Account data doesn't have all required fields.")));
92	}
93
94	let count = self.services.globals.next_count();
95	let roomuserdataid = (room_id, user_id, *count, &event_type);
96	let key = (room_id, user_id, &event_type);
97	let prev = self
98		.db
99		.roomusertype_roomuserdataid
100		.qry(&key)
101		.await;
102
103	let mut txn = self.services.db.txn();
104
105	txn.put(&self.db.roomuserdataid_accountdata, roomuserdataid, Json(data));
106	txn.put(&self.db.roomusertype_roomuserdataid, key, roomuserdataid);
107
108	if let Ok(prev) = prev {
109		txn.del_raw(&self.db.roomuserdataid_accountdata, prev);
110	}
111
112	txn.execute();
113
114	Ok(())
115}
116
117/// MSC3391: replace the stored event with a tombstone whose content is
118/// `{}`. Delta sync surfaces the empty content so clients can apply the
119/// deletion; initial sync and GET treat the tombstone as not-present.
120#[implement(Service)]
121pub async fn delete(
122	&self,
123	room_id: Option<&RoomId>,
124	user_id: &UserId,
125	event_type: RoomAccountDataEventType,
126) -> Result {
127	let tombstone = json!({
128		"type": event_type.to_string(),
129		"content": {},
130	});
131
132	self.update(room_id, user_id, event_type, &tombstone)
133		.await
134}
135
136/// Searches the room account data for a specific kind.
137#[implement(Service)]
138pub async fn get_global<T>(&self, user_id: &UserId, kind: GlobalAccountDataEventType) -> Result<T>
139where
140	T: for<'de> Deserialize<'de>,
141{
142	self.get_raw(None, user_id, &kind.to_string())
143		.await
144		.deserialized()
145}
146
147/// Searches the global account data for a specific kind.
148#[implement(Service)]
149pub async fn get_room<T>(
150	&self,
151	room_id: &RoomId,
152	user_id: &UserId,
153	kind: RoomAccountDataEventType,
154) -> Result<T>
155where
156	T: for<'de> Deserialize<'de>,
157{
158	self.get_raw(Some(room_id), user_id, &kind.to_string())
159		.await
160		.deserialized()
161}
162
163#[implement(Service)]
164pub async fn get_raw(
165	&self,
166	room_id: Option<&RoomId>,
167	user_id: &UserId,
168	kind: &str,
169) -> Result<Handle<'_>> {
170	let key = (room_id, user_id, kind.to_owned());
171	self.db
172		.roomusertype_roomuserdataid
173		.qry(&key)
174		.and_then(|roomuserdataid| {
175			self.db
176				.roomuserdataid_accountdata
177				.get(&roomuserdataid)
178		})
179		.await
180}
181
182/// Returns all changes to the account data that happened after `since`.
183#[implement(Service)]
184pub fn changes_since<'a>(
185	&'a self,
186	room_id: Option<&'a RoomId>,
187	user_id: &'a UserId,
188	since: u64,
189	to: Option<u64>,
190) -> impl Stream<Item = AnyRawAccountDataEvent> + Send + 'a {
191	self.changes_since_fallible(room_id, user_id, since, to)
192		.map(LogErr::log_err)
193		.ignore_err()
194}
195
196/// Returns bounded account-data changes without suppressing failures.
197///
198/// The lower bound is exclusive and the optional upper bound is inclusive.
199/// Cursor, decode, and deserialization failures remain in the stream for an
200/// atomic caller to handle.
201#[implement(Service)]
202pub fn changes_since_fallible<'a>(
203	&'a self,
204	room_id: Option<&'a RoomId>,
205	user_id: &'a UserId,
206	since: u64,
207	to: Option<u64>,
208) -> impl Stream<Item = Result<AnyRawAccountDataEvent>> + Send + 'a {
209	type Key<'a> = (Option<&'a RoomId>, &'a UserId, u64, Ignore);
210
211	// Skip the data that's exactly at since, because we sent that last time
212	let first_possible = (room_id, user_id, since.saturating_add(1));
213
214	self.db
215		.roomuserdataid_accountdata
216		.stream_from(&first_possible)
217		.ready_try_take_while(move |((room_id_, user_id_, count, _), _): &(Key<'_>, _)| {
218			Ok(room_id == *room_id_ && user_id == *user_id_ && to.is_none_or(|to| *count <= to))
219		})
220		.ready_and_then(move |(_, v)| {
221			match room_id {
222				| Some(_) => serde_json::from_slice::<Raw<AnyRoomAccountDataEvent>>(v)
223					.map(AnyRawAccountDataEvent::Room),
224				| None => serde_json::from_slice::<Raw<AnyGlobalAccountDataEvent>>(v)
225					.map(AnyRawAccountDataEvent::Global),
226			}
227			.map_err(|e| err!(Database("Database contains invalid account data: {e}")))
228		})
229}
230
231/// MSC4025: erase all account data for a user in the given namespace
232/// (global if `room_id` is `None`, otherwise a single room). Mirrors
233/// `threads::delete_all_rooms_threads`: prefix-scan the keys and
234/// remove each.
235#[implement(Service)]
236pub async fn erase_user(&self, user_id: &UserId, room_id: Option<&RoomId>) {
237	let prefix = (room_id, user_id, Interfix);
238	let mut txn = self.services.db.txn();
239
240	self.db
241		.roomuserdataid_accountdata
242		.keys_prefix_raw(&prefix)
243		.ignore_err()
244		.ready_for_each(|key| txn.del_raw(&self.db.roomuserdataid_accountdata, key))
245		.await;
246
247	self.db
248		.roomusertype_roomuserdataid
249		.keys_prefix_raw(&prefix)
250		.ignore_err()
251		.ready_for_each(|key| txn.del_raw(&self.db.roomusertype_roomuserdataid, key))
252		.await;
253
254	txn.execute();
255}
256
257/// Returns all changes to the account data that happened after `since`.
258#[implement(Service)]
259pub async fn last_count<'a>(
260	&'a self,
261	room_id: Option<&'a RoomId>,
262	user_id: &'a UserId,
263	upper: Option<u64>,
264) -> Result<u64> {
265	type Key<'a> = (Option<&'a RoomId>, &'a UserId, u64, Ignore);
266
267	let upper = upper.unwrap_or(u64::MAX);
268	let key = (room_id, user_id, upper, Interfix);
269	let keys = self
270		.db
271		.roomuserdataid_accountdata
272		.rev_keys_from(&key)
273		.ignore_err()
274		.ready_take_while(move |(room_id_, user_id_, ..): &Key<'_>| {
275			room_id == *room_id_ && user_id == *user_id_
276		})
277		.map(at!(2));
278
279	pin_mut!(keys);
280	keys.next()
281		.await
282		.ok_or_else(|| err!(Request(NotFound("No account data found."))))
283}
tuwunel_service/account_data/mod.rs

tuwunel_service/account_data/
mod.rs
