arminius_lib_db_stripped_state/
lib.rs

1// SPDX-FileCopyrightText: 2025 BWI GmbH
2//
3// SPDX-License-Identifier: Apache-2.0
4
5//! Queries related to the `rooms` table.
6
7use arminius_lib_db::{Database, DatabaseError, ResultExt as _};
8use arminius_lib_model::types::StreamId;
9use ruma::{
10	OwnedRoomId, OwnedUserId, UserId,
11	api::federation::membership::RawStrippedState,
12	events::{AnyStrippedStateEvent, StateEventType},
13	serde::Raw,
14};
15
16/// A stripped state event stored in the database
17#[derive(Debug, serde::Deserialize)]
18pub struct StrippedStateEvent {
19	/// The `content` field of the event
20	pub content: serde_json::Value,
21	/// The `sender` field of the event
22	pub sender: OwnedUserId,
23	/// The `state_key` field of the event
24	pub state_key: String,
25	/// The `type` field of the event
26	#[serde(rename = "type")]
27	pub event_type: StateEventType,
28}
29
30impl TryFrom<RawStrippedState> for StrippedStateEvent {
31	type Error = serde_json::Error;
32
33	fn try_from(event: RawStrippedState) -> Result<Self, Self::Error> {
34		match event {
35			#[expect(
36				deprecated,
37				reason = "The deprecation specifically states that we should not send, but still accept stripped state events received over federation"
38			)]
39			RawStrippedState::Stripped(raw) => {
40				let event = raw.deserialize()?;
41				Ok(StrippedStateEvent {
42					content: serde_json::to_value(event.content())?,
43					sender: event.sender().to_owned(),
44					state_key: event.state_key().to_owned(),
45					event_type: event.event_type(),
46				})
47			}
48			RawStrippedState::Pdu(raw_value) => Ok(serde_json::from_str(raw_value.get())?),
49			_ => {
50				unreachable!("There are only two variants of this enum")
51			}
52		}
53	}
54}
55
56impl From<StrippedStateEvent> for Raw<AnyStrippedStateEvent> {
57	#[expect(
58		clippy::expect_used,
59		reason = "Because we define this type ourself, and it consists of values and strings, we don't expect serialization failures"
60	)]
61	fn from(event: StrippedStateEvent) -> Self {
62		Raw::new(&serde_json::json!({
63			"content": event.content,
64			"sender": event.sender,
65			"state_key": event.state_key,
66			"type": event.event_type,
67		}))
68		.expect("We know that these types cannot fail to serialize")
69		.cast_unchecked()
70	}
71}
72
73/// An invite received over federation, to a room in which we are not
74/// participating
75#[derive(Debug)]
76pub struct FederatedInvite {
77	/// The user who was invited
78	pub invited_user: OwnedUserId,
79	/// The ID of the room we were invited to
80	pub room_id: OwnedRoomId,
81	/// The stripped state of the room
82	pub state_events: Vec<StrippedStateEvent>,
83}
84
85/// Queries related to the `federated_invites` table.
86// Queries are usually used as extension traits so suggest `import as _`.
87#[rust_analyzer::prefer_underscore_import]
88pub trait StrippedStateQueries: Database {
89	/// Insert a federated invite into the database
90	#[tracing::instrument(skip(self, invite))]
91	async fn insert_federated_invite(
92		&mut self,
93		invite: &FederatedInvite,
94	) -> Result<(), DatabaseError> {
95		let mut nids = Vec::new();
96		for event in &invite.state_events {
97			let nid = sqlx::query_scalar!(
98				"INSERT INTO stripped_state_events (content, sender, state_key, type) VALUES ($1, $2, $3, $4) RETURNING nid;",
99				event.content,
100				event.sender.as_str(),
101				event.state_key.as_str(),
102				&event.event_type.to_string(),
103			).fetch_one(self.connection().await?).await?;
104			nids.push(nid);
105		}
106		sqlx::query!(
107			"INSERT INTO federated_invites (user_nid, room_id, state_events) VALUES ((SELECT nid FROM users WHERE user_id = $1), $2, $3)",
108			invite.invited_user.as_str(),
109			invite.room_id.as_str(),
110			&nids
111		)
112		.execute(self.connection().await?)
113		.await?;
114		Ok(())
115	}
116
117	/// Fetch a list of federated items
118	#[tracing::instrument(skip(self))]
119	async fn fetch_federated_invites(
120		&mut self,
121		user: &UserId,
122		from_stream_id: StreamId,
123	) -> Result<Vec<(StreamId, FederatedInvite)>, DatabaseError> {
124		let rooms = sqlx::query!(
125			r#"
126			SELECT
127				stream_id,
128				room_id,
129				state_events
130			FROM
131				federated_invites
132				JOIN users ON federated_invites.user_nid = users.nid
133			WHERE
134				user_id = $1
135				AND stream_id >= $2
136			"#,
137			user.as_str(),
138			from_stream_id,
139		)
140		.fetch_all(self.connection().await?)
141		.await?;
142		let mut invites = Vec::new();
143		for room in rooms {
144			let state_events = sqlx::query!(
145				"SELECT * FROM stripped_state_events WHERE nid = ANY($1)",
146				room.state_events.as_deref()
147			)
148			.fetch_all(self.connection().await?)
149			.await?
150			.into_iter()
151			.map(|row| -> Result<StrippedStateEvent, DatabaseError> {
152				Ok(StrippedStateEvent {
153					content: row.content,
154					sender: row.sender.parse().invalid_data("Invalid user id in database")?,
155					state_key: row.state_key,
156					event_type: StateEventType::from(row.r#type),
157				})
158			})
159			.collect::<Result<Vec<_>, _>>()?;
160			invites.push((
161				room.stream_id,
162				FederatedInvite {
163					invited_user: user.to_owned(),
164					room_id: room.room_id.parse().invalid_data("Invalid UserID in database")?,
165					state_events,
166				},
167			));
168		}
169		Ok(invites)
170	}
171
172	/// Insert a federated invite into the database
173	#[tracing::instrument(skip(self, invite))]
174	async fn insert_federated_knock(
175		&mut self,
176		invite: &FederatedInvite,
177	) -> Result<(), DatabaseError> {
178		let mut nids = Vec::new();
179		for event in &invite.state_events {
180			let nid = sqlx::query_scalar!(
181				"INSERT INTO stripped_state_events (content, sender, state_key, type) VALUES ($1, $2, $3, $4) RETURNING nid;",
182				event.content,
183				event.sender.as_str(),
184				event.state_key.as_str(),
185				&event.event_type.to_string(),
186			).fetch_one(self.connection().await?).await?;
187			nids.push(nid);
188		}
189		sqlx::query!(
190			"INSERT INTO federated_knocks (user_nid, room_id, state_events) VALUES ((SELECT nid FROM users WHERE user_id = $1), $2, $3)",
191			invite.invited_user.as_str(),
192			invite.room_id.as_str(),
193			&nids
194		)
195		.execute(self.connection().await?)
196		.await?;
197		Ok(())
198	}
199
200	/// Fetch a list of federated items
201	#[tracing::instrument(skip(self))]
202	async fn fetch_federated_knocks(
203		&mut self,
204		user: &UserId,
205		from_stream_id: StreamId,
206	) -> Result<Vec<(StreamId, FederatedInvite)>, DatabaseError> {
207		let rooms = sqlx::query!(
208			r#"
209			SELECT
210				stream_id,
211				room_id,
212				state_events
213			FROM
214				federated_knocks
215				JOIN users ON federated_knocks.user_nid = users.nid
216			WHERE
217				user_id = $1
218				AND stream_id >= $2
219			"#,
220			user.as_str(),
221			from_stream_id,
222		)
223		.fetch_all(self.connection().await?)
224		.await?;
225		let mut knocks = Vec::new();
226		for room in rooms {
227			let state_events = sqlx::query!(
228				"SELECT * FROM stripped_state_events WHERE nid = ANY($1)",
229				room.state_events.as_deref()
230			)
231			.fetch_all(self.connection().await?)
232			.await?
233			.into_iter()
234			.map(|row| -> Result<StrippedStateEvent, DatabaseError> {
235				Ok(StrippedStateEvent {
236					content: row.content,
237					sender: row.sender.parse().invalid_data("Invalid user id in database")?,
238					state_key: row.state_key,
239					event_type: StateEventType::from(row.r#type),
240				})
241			})
242			.collect::<Result<Vec<_>, _>>()?;
243			knocks.push((
244				room.stream_id,
245				FederatedInvite {
246					invited_user: user.to_owned(),
247					room_id: room.room_id.parse().invalid_data("Invalid UserID in database")?,
248					state_events,
249				},
250			));
251		}
252		Ok(knocks)
253	}
254}
255
256/// Make the query available on types implementing the
257/// Database trait.
258impl<T> StrippedStateQueries for T where T: Database {}
arminius_lib_db_stripped_state/lib.rs

arminius_lib_db_stripped_state/
lib.rs
