tuwunel_service/rooms/state_cache/
via.rs

1use std::cmp::Reverse;
2
3use futures::{Stream, StreamExt, stream::iter};
4use ruma::{
5	OwnedServerName, RoomId, ServerName,
6	events::{StateEventType, room::power_levels::RoomPowerLevelsEventContent},
7	int,
8};
9use tuwunel_core::{
10	Result, implement,
11	itertools::Itertools,
12	utils::{StreamTools, stream::TryIgnore},
13	warn,
14};
15use tuwunel_database::{Ignore, Txn};
16
17#[implement(super::Service)]
18#[tracing::instrument(level = "debug", skip(self, txn, servers))]
19pub(crate) async fn add_servers_invite_via(
20	&self,
21	txn: &mut Txn,
22	room_id: &RoomId,
23	servers: Vec<OwnedServerName>,
24) {
25	let mut servers: Vec<_> = self
26		.servers_invite_via(room_id)
27		.map(ToOwned::to_owned)
28		.chain(iter(servers.into_iter()))
29		.collect()
30		.await;
31
32	servers.sort_unstable();
33	servers.dedup();
34
35	let servers = servers
36		.iter()
37		.map(|server| server.as_bytes())
38		.collect_vec()
39		.join(&[0xFF][..]);
40
41	txn.insert_raw(&self.db.roomid_inviteviaservers, room_id.as_bytes(), &servers);
42}
43
44/// Gets up to five servers that are likely to be in the room in the
45/// distant future.
46///
47/// See <https://spec.matrix.org/latest/appendices/#routing>
48#[implement(super::Service)]
49#[tracing::instrument(skip(self), level = "trace")]
50pub async fn servers_route_via(&self, room_id: &RoomId) -> Result<Vec<OwnedServerName>> {
51	let most_powerful = self.most_powerful_user_server(room_id).await;
52
53	Ok(most_powerful
54		.into_iter()
55		.chain(self.popular_servers(room_id).await)
56		.take(5)
57		.collect())
58}
59
60/// The room's highest power-level user's server, provided that user holds at
61/// least power level 50.
62#[implement(super::Service)]
63#[tracing::instrument(skip(self), level = "trace")]
64pub async fn most_powerful_user_server(&self, room_id: &RoomId) -> Option<OwnedServerName> {
65	self.services
66		.state_accessor
67		.room_state_get_content(room_id, &StateEventType::RoomPowerLevels, "")
68		.await
69		.ok()
70		.and_then(|content: RoomPowerLevelsEventContent| {
71			content
72				.users
73				.into_iter()
74				.max_by_key(|(_, power)| *power)
75				.filter(|(_, power)| *power >= int!(50))
76				.map(|(user, _)| user.server_name().to_owned())
77		})
78}
79
80/// Servers participating in the room, ordered by descending resident user
81/// count. Counting members per server is an aggregation, so the result is
82/// materialized rather than streamed.
83#[implement(super::Service)]
84#[tracing::instrument(skip(self), level = "trace")]
85pub async fn popular_servers(&self, room_id: &RoomId) -> Vec<OwnedServerName> {
86	self.room_members(room_id)
87		.counts_by(|user| user.server_name().to_owned())
88		.await
89		.into_iter()
90		.sorted_by_key(|(_, users)| Reverse(*users))
91		.map(|(server, _)| server)
92		.collect()
93}
94
95#[implement(super::Service)]
96#[tracing::instrument(skip(self), level = "debug")]
97pub fn servers_invite_via<'a>(
98	&'a self,
99	room_id: &'a RoomId,
100) -> impl Stream<Item = &ServerName> + Send + 'a {
101	type KeyVal<'a> = (Ignore, Vec<&'a ServerName>);
102
103	self.db
104		.roomid_inviteviaservers
105		.stream_raw_prefix(room_id)
106		.ignore_err()
107		.map(|(_, servers): KeyVal<'_>| *servers.last().expect("at least one server"))
108}
tuwunel_service/rooms/state_cache/via.rs

tuwunel_service/rooms/state_cache/
via.rs
