mirror of
https://forgejo.ellis.link/continuwuation/continuwuity/
synced 2026-10-23 21:52:23 +00:00
style: Document & refactor rooms/state_cache service
This commit is contained in:
3 files changed
+745
-792
No files matched your search
@@ -4,10 +4,9 @@
|
||||
use std::{collections::HashMap, sync::Arc};
|
||||
|
||||
use conduwuit::{
|
||||
Pdu, Result, SyncRwLock, implement,
|
||||
Pdu, Result, SyncRwLock,
|
||||
result::LogErr,
|
||||
utils::{ReadyExt, stream::TryIgnore},
|
||||
warn,
|
||||
};
|
||||
use database::{Deserialized, Ignore, Interfix, Map};
|
||||
use futures::{Stream, StreamExt, future::join5, pin_mut};
|
||||
@@ -93,448 +92,414 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
|
||||
fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
|
||||
}
|
||||
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
pub async fn appservice_in_room(&self, room_id: &RoomId, appservice: &RegistrationInfo) -> bool {
|
||||
if let Some(cached) = self
|
||||
.appservice_in_room_cache
|
||||
.read()
|
||||
.get(room_id)
|
||||
.and_then(|map| map.get(&appservice.registration.id))
|
||||
.copied()
|
||||
{
|
||||
return cached;
|
||||
impl Service {
|
||||
/// Checks if an appservice (or any user in its claimed namespace) is in the
|
||||
/// target room.
|
||||
pub async fn appservice_in_room(
|
||||
&self,
|
||||
room_id: &RoomId,
|
||||
appservice: &RegistrationInfo,
|
||||
) -> bool {
|
||||
if let Some(cached) = self
|
||||
.appservice_in_room_cache
|
||||
.read()
|
||||
.get(room_id)
|
||||
.and_then(|map| map.get(&appservice.registration.id))
|
||||
.copied()
|
||||
{
|
||||
return cached;
|
||||
}
|
||||
|
||||
let appservice_user_id = UserId::parse_with_server_name(
|
||||
appservice.registration.sender_localpart.as_str(),
|
||||
self.services.globals.server_name(),
|
||||
);
|
||||
|
||||
let Ok(bridge_user_id) = appservice_user_id.log_err() else {
|
||||
return false;
|
||||
};
|
||||
|
||||
let (as_user_joined, any_as_namespace_user_joined) = tokio::join!(
|
||||
self.is_joined(&bridge_user_id, room_id),
|
||||
self.room_members(room_id)
|
||||
.ready_any(|user_id| appservice.users.is_match(user_id.as_str()))
|
||||
);
|
||||
let in_room = as_user_joined || any_as_namespace_user_joined;
|
||||
|
||||
self.appservice_in_room_cache
|
||||
.write()
|
||||
.entry(room_id.into())
|
||||
.or_default()
|
||||
.insert(appservice.registration.id.clone(), in_room);
|
||||
|
||||
in_room
|
||||
}
|
||||
|
||||
let bridge_user_id = UserId::parse_with_server_name(
|
||||
appservice.registration.sender_localpart.as_str(),
|
||||
self.services.globals.server_name(),
|
||||
);
|
||||
/// Gets the size of the AS joined room cache and its capacity.
|
||||
pub fn get_appservice_in_room_cache_usage(&self) -> (usize, usize) {
|
||||
let cache = self.appservice_in_room_cache.read();
|
||||
|
||||
let Ok(bridge_user_id) = bridge_user_id.log_err() else {
|
||||
return false;
|
||||
};
|
||||
(cache.len(), cache.capacity())
|
||||
}
|
||||
|
||||
let in_room = self.is_joined(&bridge_user_id, room_id).await
|
||||
|| self
|
||||
.room_members(room_id)
|
||||
.ready_any(|user_id| appservice.users.is_match(user_id.as_str()))
|
||||
.await;
|
||||
/// Clears the internal cache of appservice joined rooms.
|
||||
pub fn clear_appservice_in_room_cache(&self) {
|
||||
self.appservice_in_room_cache.write().clear();
|
||||
}
|
||||
|
||||
self.appservice_in_room_cache
|
||||
.write()
|
||||
.entry(room_id.into())
|
||||
.or_default()
|
||||
.insert(appservice.registration.id.clone(), in_room);
|
||||
/// Returns an iterator of all servers participating in this room.
|
||||
pub fn room_servers<'a>(
|
||||
&'a self,
|
||||
room_id: &'a RoomId,
|
||||
) -> impl Stream<Item = OwnedServerName> + Send + 'a {
|
||||
let prefix = (room_id, Interfix);
|
||||
self.db
|
||||
.roomserverids
|
||||
.keys_prefix(&prefix)
|
||||
.ignore_err()
|
||||
.map(|(_, server): (Ignore, OwnedServerName)| server)
|
||||
}
|
||||
|
||||
in_room
|
||||
}
|
||||
/// Determines if a server name has at least one joined member in a room.
|
||||
pub async fn server_in_room<'a>(
|
||||
&'a self,
|
||||
server: &'a ServerName,
|
||||
room_id: &'a RoomId,
|
||||
) -> bool {
|
||||
let key = (server, room_id);
|
||||
self.db.serverroomids.qry(&key).await.is_ok()
|
||||
}
|
||||
|
||||
#[implement(Service)]
|
||||
pub fn get_appservice_in_room_cache_usage(&self) -> (usize, usize) {
|
||||
let cache = self.appservice_in_room_cache.read();
|
||||
/// Returns an iterator of all rooms a server participates in (as far as we
|
||||
/// know).
|
||||
pub fn server_rooms<'a>(
|
||||
&'a self,
|
||||
server: &'a ServerName,
|
||||
) -> impl Stream<Item = OwnedRoomId> + Send + 'a {
|
||||
let prefix = (server, Interfix);
|
||||
self.db
|
||||
.serverroomids
|
||||
.keys_prefix(&prefix)
|
||||
.ignore_err()
|
||||
.map(|(_, room_id): (Ignore, OwnedRoomId)| room_id)
|
||||
}
|
||||
|
||||
(cache.len(), cache.capacity())
|
||||
}
|
||||
/// Returns true if server can see user by sharing at least one room.
|
||||
pub async fn server_sees_user(&self, server: &ServerName, user_id: &UserId) -> bool {
|
||||
self.server_rooms(server)
|
||||
.any(async |room_id| self.is_joined(user_id, &room_id).await)
|
||||
.await
|
||||
}
|
||||
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(level = "debug", skip_all)]
|
||||
pub fn clear_appservice_in_room_cache(&self) { self.appservice_in_room_cache.write().clear(); }
|
||||
/// Returns true if user_a and user_b share at least one room.
|
||||
pub async fn user_sees_user(&self, user_a: &UserId, user_b: &UserId) -> bool {
|
||||
let get_shared_rooms = self.get_shared_rooms(user_a, user_b);
|
||||
|
||||
/// Returns an iterator of all servers participating in this room.
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "debug")]
|
||||
pub fn room_servers<'a>(
|
||||
&'a self,
|
||||
room_id: &'a RoomId,
|
||||
) -> impl Stream<Item = OwnedServerName> + Send + 'a {
|
||||
let prefix = (room_id, Interfix);
|
||||
self.db
|
||||
.roomserverids
|
||||
.keys_prefix(&prefix)
|
||||
.ignore_err()
|
||||
.map(|(_, server): (Ignore, OwnedServerName)| server)
|
||||
}
|
||||
pin_mut!(get_shared_rooms);
|
||||
get_shared_rooms.next().await.is_some()
|
||||
}
|
||||
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "trace")]
|
||||
pub async fn server_in_room<'a>(&'a self, server: &'a ServerName, room_id: &'a RoomId) -> bool {
|
||||
let key = (server, room_id);
|
||||
self.db.serverroomids.qry(&key).await.is_ok()
|
||||
}
|
||||
/// Lists the rooms common between two users.
|
||||
pub fn get_shared_rooms<'a>(
|
||||
&'a self,
|
||||
user_a: &'a UserId,
|
||||
user_b: &'a UserId,
|
||||
) -> impl Stream<Item = OwnedRoomId> + Send + 'a {
|
||||
use conduwuit::utils::set;
|
||||
|
||||
/// Returns an iterator of all rooms a server participates in (as far as we
|
||||
/// know).
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "debug")]
|
||||
pub fn server_rooms<'a>(
|
||||
&'a self,
|
||||
server: &'a ServerName,
|
||||
) -> impl Stream<Item = OwnedRoomId> + Send + 'a {
|
||||
let prefix = (server, Interfix);
|
||||
self.db
|
||||
.serverroomids
|
||||
.keys_prefix(&prefix)
|
||||
.ignore_err()
|
||||
.map(|(_, room_id): (Ignore, OwnedRoomId)| room_id)
|
||||
}
|
||||
let a = self.rooms_joined(user_a);
|
||||
let b = self.rooms_joined(user_b);
|
||||
set::intersection_sorted_stream2(a, b)
|
||||
}
|
||||
|
||||
/// Returns true if server can see user by sharing at least one room.
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "trace")]
|
||||
pub async fn server_sees_user(&self, server: &ServerName, user_id: &UserId) -> bool {
|
||||
self.server_rooms(server)
|
||||
.any(async |room_id| self.is_joined(user_id, &room_id).await)
|
||||
.await
|
||||
}
|
||||
/// Returns an iterator of all joined members of a room.
|
||||
pub fn room_members<'a>(
|
||||
&'a self,
|
||||
room_id: &'a RoomId,
|
||||
) -> impl Stream<Item = OwnedUserId> + Send + 'a {
|
||||
let prefix = (room_id, Interfix);
|
||||
self.db
|
||||
.roomuserid_joined
|
||||
.keys_prefix(&prefix)
|
||||
.ignore_err()
|
||||
.map(|(_, user_id): (Ignore, OwnedUserId)| user_id)
|
||||
}
|
||||
|
||||
/// Returns true if user_a and user_b share at least one room.
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "trace")]
|
||||
pub async fn user_sees_user(&self, user_a: &UserId, user_b: &UserId) -> bool {
|
||||
let get_shared_rooms = self.get_shared_rooms(user_a, user_b);
|
||||
/// Returns the number of users which are currently in a room.
|
||||
pub async fn room_joined_count(&self, room_id: &RoomId) -> Result<u64> {
|
||||
self.db.roomid_joinedcount.get(room_id).await.deserialized()
|
||||
}
|
||||
|
||||
pin_mut!(get_shared_rooms);
|
||||
get_shared_rooms.next().await.is_some()
|
||||
}
|
||||
/// Returns an iterator of all our local users in the room, even if they're
|
||||
/// deactivated.
|
||||
pub fn local_users_in_room<'a>(
|
||||
&'a self,
|
||||
room_id: &'a RoomId,
|
||||
) -> impl Stream<Item = OwnedUserId> + Send + 'a {
|
||||
self.room_members(room_id)
|
||||
.ready_filter(|user| self.services.globals.user_is_local(user))
|
||||
}
|
||||
|
||||
/// List the rooms common between two users
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "debug")]
|
||||
pub fn get_shared_rooms<'a>(
|
||||
&'a self,
|
||||
user_a: &'a UserId,
|
||||
user_b: &'a UserId,
|
||||
) -> impl Stream<Item = OwnedRoomId> + Send + 'a {
|
||||
use conduwuit::utils::set;
|
||||
/// Returns an iterator of all our local joined users in a room who are
|
||||
/// active (not deactivated)
|
||||
pub fn active_local_users_in_room<'a>(
|
||||
&'a self,
|
||||
room_id: &'a RoomId,
|
||||
) -> impl Stream<Item = OwnedUserId> + Send + 'a {
|
||||
self.local_users_in_room(room_id)
|
||||
.filter_map(async |user_id| {
|
||||
if self.services.users.is_active(&user_id).await {
|
||||
Some(user_id)
|
||||
} else {
|
||||
None
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
let a = self.rooms_joined(user_a);
|
||||
let b = self.rooms_joined(user_b);
|
||||
set::intersection_sorted_stream2(a, b)
|
||||
}
|
||||
/// Returns the number of users which are currently invited to a room
|
||||
pub async fn room_invited_count(&self, room_id: &RoomId) -> Result<u64> {
|
||||
self.db
|
||||
.roomid_invitedcount
|
||||
.get(room_id)
|
||||
.await
|
||||
.deserialized()
|
||||
}
|
||||
|
||||
/// Returns an iterator of all joined members of a room.
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "debug")]
|
||||
pub fn room_members<'a>(
|
||||
&'a self,
|
||||
room_id: &'a RoomId,
|
||||
) -> impl Stream<Item = OwnedUserId> + Send + 'a {
|
||||
let prefix = (room_id, Interfix);
|
||||
self.db
|
||||
.roomuserid_joined
|
||||
.keys_prefix(&prefix)
|
||||
.ignore_err()
|
||||
.map(|(_, user_id): (Ignore, OwnedUserId)| user_id)
|
||||
}
|
||||
/// Returns an iterator over all User IDs who ever joined a room, even if
|
||||
/// they are no longer joined.
|
||||
pub fn room_useroncejoined<'a>(
|
||||
&'a self,
|
||||
room_id: &'a RoomId,
|
||||
) -> impl Stream<Item = OwnedUserId> + Send + 'a {
|
||||
let prefix = (room_id, Interfix);
|
||||
self.db
|
||||
.roomuseroncejoinedids
|
||||
.keys_prefix(&prefix)
|
||||
.ignore_err()
|
||||
.map(|(_, user_id): (Ignore, OwnedUserId)| user_id)
|
||||
}
|
||||
|
||||
/// Returns the number of users which are currently in a room
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "trace")]
|
||||
pub async fn room_joined_count(&self, room_id: &RoomId) -> Result<u64> {
|
||||
self.db.roomid_joinedcount.get(room_id).await.deserialized()
|
||||
}
|
||||
/// Returns an iterator over all invited members of a room.
|
||||
pub fn room_members_invited<'a>(
|
||||
&'a self,
|
||||
room_id: &'a RoomId,
|
||||
) -> impl Stream<Item = OwnedUserId> + Send + 'a {
|
||||
let prefix = (room_id, Interfix);
|
||||
self.db
|
||||
.roomuserid_invitecount
|
||||
.keys_prefix(&prefix)
|
||||
.ignore_err()
|
||||
.map(|(_, user_id): (Ignore, OwnedUserId)| user_id)
|
||||
}
|
||||
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "debug")]
|
||||
/// Returns an iterator of all our local users in the room, even if they're
|
||||
/// deactivated
|
||||
pub fn local_users_in_room<'a>(
|
||||
&'a self,
|
||||
room_id: &'a RoomId,
|
||||
) -> impl Stream<Item = OwnedUserId> + Send + 'a {
|
||||
self.room_members(room_id)
|
||||
.ready_filter(|user| self.services.globals.user_is_local(user))
|
||||
}
|
||||
/// Returns an iterator over all knocking members of a room.
|
||||
pub fn room_members_knocked<'a>(
|
||||
&'a self,
|
||||
room_id: &'a RoomId,
|
||||
) -> impl Stream<Item = OwnedUserId> + Send + 'a {
|
||||
let prefix = (room_id, Interfix);
|
||||
self.db
|
||||
.roomuserid_knockedcount
|
||||
.keys_prefix(&prefix)
|
||||
.ignore_err()
|
||||
.map(|(_, user_id): (Ignore, OwnedUserId)| user_id)
|
||||
}
|
||||
|
||||
/// Returns an iterator of all our local joined users in a room who are
|
||||
/// active (not deactivated)
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "trace")]
|
||||
pub fn active_local_users_in_room<'a>(
|
||||
&'a self,
|
||||
room_id: &'a RoomId,
|
||||
) -> impl Stream<Item = OwnedUserId> + Send + 'a {
|
||||
self.local_users_in_room(room_id)
|
||||
.filter_map(async |user_id| {
|
||||
if self.services.users.is_active(&user_id).await {
|
||||
Some(user_id)
|
||||
} else {
|
||||
None
|
||||
}
|
||||
})
|
||||
}
|
||||
/// Gets the invite count value.
|
||||
pub async fn get_invite_count(&self, room_id: &RoomId, user_id: &UserId) -> Result<u64> {
|
||||
let key = (room_id, user_id);
|
||||
self.db
|
||||
.roomuserid_invitecount
|
||||
.qry(&key)
|
||||
.await
|
||||
.deserialized()
|
||||
}
|
||||
|
||||
/// Returns the number of users which are currently invited to a room
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "trace")]
|
||||
pub async fn room_invited_count(&self, room_id: &RoomId) -> Result<u64> {
|
||||
self.db
|
||||
.roomid_invitedcount
|
||||
.get(room_id)
|
||||
.await
|
||||
.deserialized()
|
||||
}
|
||||
/// Gets the knock count value.
|
||||
pub async fn get_knock_count(&self, room_id: &RoomId, user_id: &UserId) -> Result<u64> {
|
||||
let key = (room_id, user_id);
|
||||
self.db
|
||||
.roomuserid_knockedcount
|
||||
.qry(&key)
|
||||
.await
|
||||
.deserialized()
|
||||
}
|
||||
|
||||
/// Returns an iterator over all User IDs who ever joined a room.
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "debug")]
|
||||
pub fn room_useroncejoined<'a>(
|
||||
&'a self,
|
||||
room_id: &'a RoomId,
|
||||
) -> impl Stream<Item = OwnedUserId> + Send + 'a {
|
||||
let prefix = (room_id, Interfix);
|
||||
self.db
|
||||
.roomuseroncejoinedids
|
||||
.keys_prefix(&prefix)
|
||||
.ignore_err()
|
||||
.map(|(_, user_id): (Ignore, OwnedUserId)| user_id)
|
||||
}
|
||||
/// Gets the left count value.
|
||||
pub async fn get_left_count(&self, room_id: &RoomId, user_id: &UserId) -> Result<u64> {
|
||||
let key = (room_id, user_id);
|
||||
self.db.roomuserid_leftcount.qry(&key).await.deserialized()
|
||||
}
|
||||
|
||||
/// Returns an iterator over all invited members of a room.
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "debug")]
|
||||
pub fn room_members_invited<'a>(
|
||||
&'a self,
|
||||
room_id: &'a RoomId,
|
||||
) -> impl Stream<Item = OwnedUserId> + Send + 'a {
|
||||
let prefix = (room_id, Interfix);
|
||||
self.db
|
||||
.roomuserid_invitecount
|
||||
.keys_prefix(&prefix)
|
||||
.ignore_err()
|
||||
.map(|(_, user_id): (Ignore, OwnedUserId)| user_id)
|
||||
}
|
||||
/// Returns an iterator over all rooms this user joined.
|
||||
pub fn rooms_joined<'a>(
|
||||
&'a self,
|
||||
user_id: &'a UserId,
|
||||
) -> impl Stream<Item = OwnedRoomId> + Send + 'a {
|
||||
self.db
|
||||
.userroomid_joined
|
||||
.keys_raw_prefix(user_id)
|
||||
.ignore_err()
|
||||
.map(|(_, room_id): (Ignore, OwnedRoomId)| room_id)
|
||||
}
|
||||
|
||||
/// Returns an iterator over all knocked members of a room.
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "debug")]
|
||||
pub fn room_members_knocked<'a>(
|
||||
&'a self,
|
||||
room_id: &'a RoomId,
|
||||
) -> impl Stream<Item = OwnedUserId> + Send + 'a {
|
||||
let prefix = (room_id, Interfix);
|
||||
self.db
|
||||
.roomuserid_knockedcount
|
||||
.keys_prefix(&prefix)
|
||||
.ignore_err()
|
||||
.map(|(_, user_id): (Ignore, OwnedUserId)| user_id)
|
||||
}
|
||||
/// Returns an iterator over all rooms a user was invited to.
|
||||
pub fn rooms_invited<'a>(
|
||||
&'a self,
|
||||
user_id: &'a UserId,
|
||||
) -> impl Stream<Item = StrippedStateEventItem> + Send + 'a {
|
||||
type KeyVal = (Key, Raw<Vec<AnyStrippedStateEvent>>);
|
||||
type Key = (OwnedUserId, OwnedRoomId);
|
||||
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "trace")]
|
||||
pub async fn get_invite_count(&self, room_id: &RoomId, user_id: &UserId) -> Result<u64> {
|
||||
let key = (room_id, user_id);
|
||||
self.db
|
||||
.roomuserid_invitecount
|
||||
.qry(&key)
|
||||
.await
|
||||
.deserialized()
|
||||
}
|
||||
let prefix = (user_id, Interfix);
|
||||
self.db
|
||||
.userroomid_invitestate
|
||||
.stream_prefix(&prefix)
|
||||
.ignore_err()
|
||||
.map(|((_, room_id), state): KeyVal| (room_id, state))
|
||||
.map(|(room_id, state)| Ok((room_id, state.deserialize_as_unchecked()?)))
|
||||
.ignore_err()
|
||||
}
|
||||
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "trace")]
|
||||
pub async fn get_knock_count(&self, room_id: &RoomId, user_id: &UserId) -> Result<u64> {
|
||||
let key = (room_id, user_id);
|
||||
self.db
|
||||
.roomuserid_knockedcount
|
||||
.qry(&key)
|
||||
.await
|
||||
.deserialized()
|
||||
}
|
||||
/// Returns an iterator over all rooms a user is currently knocking on.
|
||||
pub fn rooms_knocked<'a>(
|
||||
&'a self,
|
||||
user_id: &'a UserId,
|
||||
) -> impl Stream<Item = StrippedStateEventItem> + Send + 'a {
|
||||
type KeyVal = (Key, Raw<Vec<AnyStrippedStateEvent>>);
|
||||
type Key = (OwnedUserId, OwnedRoomId);
|
||||
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "trace")]
|
||||
pub async fn get_left_count(&self, room_id: &RoomId, user_id: &UserId) -> Result<u64> {
|
||||
let key = (room_id, user_id);
|
||||
self.db.roomuserid_leftcount.qry(&key).await.deserialized()
|
||||
}
|
||||
let prefix = (user_id, Interfix);
|
||||
self.db
|
||||
.userroomid_knockedstate
|
||||
.stream_prefix(&prefix)
|
||||
.ignore_err()
|
||||
.map(|((_, room_id), state): KeyVal| (room_id, state))
|
||||
.map(|(room_id, state)| Ok((room_id, state.deserialize_as_unchecked()?)))
|
||||
.ignore_err()
|
||||
}
|
||||
|
||||
/// Returns an iterator over all rooms this user joined.
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "debug")]
|
||||
pub fn rooms_joined<'a>(
|
||||
&'a self,
|
||||
user_id: &'a UserId,
|
||||
) -> impl Stream<Item = OwnedRoomId> + Send + 'a {
|
||||
self.db
|
||||
.userroomid_joined
|
||||
.keys_raw_prefix(user_id)
|
||||
.ignore_err()
|
||||
.map(|(_, room_id): (Ignore, OwnedRoomId)| room_id)
|
||||
}
|
||||
/// Returns the invite state for a user in a room, if they are invited.
|
||||
pub async fn invite_state(
|
||||
&self,
|
||||
user_id: &UserId,
|
||||
room_id: &RoomId,
|
||||
) -> Result<Vec<Raw<AnyStrippedStateEvent>>> {
|
||||
let key = (user_id, room_id);
|
||||
self.db
|
||||
.userroomid_invitestate
|
||||
.qry(&key)
|
||||
.await
|
||||
.deserialized()
|
||||
.and_then(|val: Raw<Vec<AnyStrippedStateEvent>>| {
|
||||
val.deserialize_as_unchecked().map_err(Into::into)
|
||||
})
|
||||
}
|
||||
|
||||
/// Returns an iterator over all rooms a user was invited to.
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "debug")]
|
||||
pub fn rooms_invited<'a>(
|
||||
&'a self,
|
||||
user_id: &'a UserId,
|
||||
) -> impl Stream<Item = StrippedStateEventItem> + Send + 'a {
|
||||
type KeyVal = (Key, Raw<Vec<AnyStrippedStateEvent>>);
|
||||
type Key = (OwnedUserId, OwnedRoomId);
|
||||
/// Returns the knock state for a user in a room, if they are knocking.
|
||||
pub async fn knock_state(
|
||||
&self,
|
||||
user_id: &UserId,
|
||||
room_id: &RoomId,
|
||||
) -> Result<Vec<Raw<AnyStrippedStateEvent>>> {
|
||||
let key = (user_id, room_id);
|
||||
self.db
|
||||
.userroomid_knockedstate
|
||||
.qry(&key)
|
||||
.await
|
||||
.deserialized()
|
||||
.and_then(|val: Raw<Vec<AnyStrippedStateEvent>>| {
|
||||
val.deserialize_as_unchecked().map_err(Into::into)
|
||||
})
|
||||
}
|
||||
|
||||
let prefix = (user_id, Interfix);
|
||||
self.db
|
||||
.userroomid_invitestate
|
||||
.stream_prefix(&prefix)
|
||||
.ignore_err()
|
||||
.map(|((_, room_id), state): KeyVal| (room_id, state))
|
||||
.map(|(room_id, state)| Ok((room_id, state.deserialize_as_unchecked()?)))
|
||||
.ignore_err()
|
||||
}
|
||||
/// Returns the left state for a user in a room, if they have left.
|
||||
pub async fn left_state(&self, user_id: &UserId, room_id: &RoomId) -> Result<Option<Pdu>> {
|
||||
let key = (user_id, room_id);
|
||||
self.db.userroomid_leftstate.qry(&key).await.deserialized()
|
||||
}
|
||||
|
||||
/// Returns an iterator over all rooms a user is currently knocking.
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "trace")]
|
||||
pub fn rooms_knocked<'a>(
|
||||
&'a self,
|
||||
user_id: &'a UserId,
|
||||
) -> impl Stream<Item = StrippedStateEventItem> + Send + 'a {
|
||||
type KeyVal = (Key, Raw<Vec<AnyStrippedStateEvent>>);
|
||||
type Key = (OwnedUserId, OwnedRoomId);
|
||||
/// Returns an iterator over all rooms a user left.
|
||||
pub fn rooms_left<'a>(
|
||||
&'a self,
|
||||
user_id: &'a UserId,
|
||||
) -> impl Stream<Item = (OwnedRoomId, Option<Pdu>)> + Send + 'a {
|
||||
type KeyVal = (Key, Raw<Option<Pdu>>);
|
||||
type Key = (OwnedUserId, OwnedRoomId);
|
||||
|
||||
let prefix = (user_id, Interfix);
|
||||
self.db
|
||||
.userroomid_knockedstate
|
||||
.stream_prefix(&prefix)
|
||||
.ignore_err()
|
||||
.map(|((_, room_id), state): KeyVal| (room_id, state))
|
||||
.map(|(room_id, state)| Ok((room_id, state.deserialize_as_unchecked()?)))
|
||||
.ignore_err()
|
||||
}
|
||||
let prefix = (user_id, Interfix);
|
||||
self.db
|
||||
.userroomid_leftstate
|
||||
.stream_prefix(&prefix)
|
||||
.ignore_err()
|
||||
.map(|((_, room_id), state): KeyVal| (room_id, state))
|
||||
.map(|(room_id, state)| Ok((room_id, state.deserialize()?)))
|
||||
.ignore_err()
|
||||
}
|
||||
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "trace")]
|
||||
pub async fn invite_state(
|
||||
&self,
|
||||
user_id: &UserId,
|
||||
room_id: &RoomId,
|
||||
) -> Result<Vec<Raw<AnyStrippedStateEvent>>> {
|
||||
let key = (user_id, room_id);
|
||||
self.db
|
||||
.userroomid_invitestate
|
||||
.qry(&key)
|
||||
.await
|
||||
.deserialized()
|
||||
.and_then(|val: Raw<Vec<AnyStrippedStateEvent>>| {
|
||||
val.deserialize_as_unchecked().map_err(Into::into)
|
||||
})
|
||||
}
|
||||
/// Determines a user's membership in a target room. If the user has never
|
||||
/// been a member of the target room at any point in time, None is returned.
|
||||
pub async fn user_membership(
|
||||
&self,
|
||||
user_id: &UserId,
|
||||
room_id: &RoomId,
|
||||
) -> Option<MembershipState> {
|
||||
let states = join5(
|
||||
self.is_joined(user_id, room_id),
|
||||
self.is_left(user_id, room_id),
|
||||
self.is_knocked(user_id, room_id),
|
||||
self.is_invited(user_id, room_id),
|
||||
self.once_joined(user_id, room_id),
|
||||
)
|
||||
.await;
|
||||
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "trace")]
|
||||
pub async fn knock_state(
|
||||
&self,
|
||||
user_id: &UserId,
|
||||
room_id: &RoomId,
|
||||
) -> Result<Vec<Raw<AnyStrippedStateEvent>>> {
|
||||
let key = (user_id, room_id);
|
||||
self.db
|
||||
.userroomid_knockedstate
|
||||
.qry(&key)
|
||||
.await
|
||||
.deserialized()
|
||||
.and_then(|val: Raw<Vec<AnyStrippedStateEvent>>| {
|
||||
val.deserialize_as_unchecked().map_err(Into::into)
|
||||
})
|
||||
}
|
||||
match states {
|
||||
| (true, ..) => Some(MembershipState::Join),
|
||||
| (_, true, ..) => Some(MembershipState::Leave),
|
||||
| (_, _, true, ..) => Some(MembershipState::Knock),
|
||||
| (_, _, _, true, ..) => Some(MembershipState::Invite),
|
||||
| (false, false, false, false, true) => Some(MembershipState::Ban),
|
||||
| _ => None,
|
||||
}
|
||||
}
|
||||
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "trace")]
|
||||
pub async fn left_state(&self, user_id: &UserId, room_id: &RoomId) -> Result<Option<Pdu>> {
|
||||
let key = (user_id, room_id);
|
||||
self.db.userroomid_leftstate.qry(&key).await.deserialized()
|
||||
}
|
||||
/// Checks if a user has joined at least once before.
|
||||
pub async fn once_joined(&self, user_id: &UserId, room_id: &RoomId) -> bool {
|
||||
let key = (user_id, room_id);
|
||||
self.db.roomuseroncejoinedids.qry(&key).await.is_ok()
|
||||
}
|
||||
|
||||
/// Returns an iterator over all rooms a user left.
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "debug")]
|
||||
pub fn rooms_left<'a>(
|
||||
&'a self,
|
||||
user_id: &'a UserId,
|
||||
) -> impl Stream<Item = (OwnedRoomId, Option<Pdu>)> + Send + 'a {
|
||||
type KeyVal = (Key, Raw<Option<Pdu>>);
|
||||
type Key = (OwnedUserId, OwnedRoomId);
|
||||
/// Checks if a user is currently joined to a room.
|
||||
pub async fn is_joined<'a>(&'a self, user_id: &'a UserId, room_id: &'a RoomId) -> bool {
|
||||
let key = (user_id, room_id);
|
||||
self.db.userroomid_joined.qry(&key).await.is_ok()
|
||||
}
|
||||
|
||||
let prefix = (user_id, Interfix);
|
||||
self.db
|
||||
.userroomid_leftstate
|
||||
.stream_prefix(&prefix)
|
||||
.ignore_err()
|
||||
.map(|((_, room_id), state): KeyVal| (room_id, state))
|
||||
.map(|(room_id, state)| Ok((room_id, state.deserialize()?)))
|
||||
.ignore_err()
|
||||
}
|
||||
/// Checks if a user is currently knocking on a room.
|
||||
pub async fn is_knocked<'a>(&'a self, user_id: &'a UserId, room_id: &'a RoomId) -> bool {
|
||||
let key = (user_id, room_id);
|
||||
self.db.userroomid_knockedstate.qry(&key).await.is_ok()
|
||||
}
|
||||
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "trace")]
|
||||
pub async fn user_membership(
|
||||
&self,
|
||||
user_id: &UserId,
|
||||
room_id: &RoomId,
|
||||
) -> Option<MembershipState> {
|
||||
let states = join5(
|
||||
self.is_joined(user_id, room_id),
|
||||
self.is_left(user_id, room_id),
|
||||
self.is_knocked(user_id, room_id),
|
||||
self.is_invited(user_id, room_id),
|
||||
self.once_joined(user_id, room_id),
|
||||
)
|
||||
.await;
|
||||
/// Checks if a user is currently invited to a room.
|
||||
pub async fn is_invited(&self, user_id: &UserId, room_id: &RoomId) -> bool {
|
||||
let key = (user_id, room_id);
|
||||
self.db.userroomid_invitestate.qry(&key).await.is_ok()
|
||||
}
|
||||
|
||||
match states {
|
||||
| (true, ..) => Some(MembershipState::Join),
|
||||
| (_, true, ..) => Some(MembershipState::Leave),
|
||||
| (_, _, true, ..) => Some(MembershipState::Knock),
|
||||
| (_, _, _, true, ..) => Some(MembershipState::Invite),
|
||||
| (false, false, false, false, true) => Some(MembershipState::Ban),
|
||||
| _ => None,
|
||||
/// Checks if a user has left the room.
|
||||
pub async fn is_left(&self, user_id: &UserId, room_id: &RoomId) -> bool {
|
||||
let key = (user_id, room_id);
|
||||
self.db.userroomid_leftstate.qry(&key).await.is_ok()
|
||||
}
|
||||
|
||||
/// Fetches the sender who invited a target user.
|
||||
pub async fn invite_sender(&self, user_id: &UserId, room_id: &RoomId) -> Result<OwnedUserId> {
|
||||
let key = (user_id, room_id);
|
||||
self.db
|
||||
.userroomid_invitesender
|
||||
.qry(&key)
|
||||
.await
|
||||
.deserialized()
|
||||
}
|
||||
}
|
||||
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "debug")]
|
||||
pub async fn once_joined(&self, user_id: &UserId, room_id: &RoomId) -> bool {
|
||||
let key = (user_id, room_id);
|
||||
self.db.roomuseroncejoinedids.qry(&key).await.is_ok()
|
||||
}
|
||||
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "trace")]
|
||||
pub async fn is_joined<'a>(&'a self, user_id: &'a UserId, room_id: &'a RoomId) -> bool {
|
||||
let key = (user_id, room_id);
|
||||
self.db.userroomid_joined.qry(&key).await.is_ok()
|
||||
}
|
||||
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "trace")]
|
||||
pub async fn is_knocked<'a>(&'a self, user_id: &'a UserId, room_id: &'a RoomId) -> bool {
|
||||
let key = (user_id, room_id);
|
||||
self.db.userroomid_knockedstate.qry(&key).await.is_ok()
|
||||
}
|
||||
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "trace")]
|
||||
pub async fn is_invited(&self, user_id: &UserId, room_id: &RoomId) -> bool {
|
||||
let key = (user_id, room_id);
|
||||
self.db.userroomid_invitestate.qry(&key).await.is_ok()
|
||||
}
|
||||
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "trace")]
|
||||
pub async fn is_left(&self, user_id: &UserId, room_id: &RoomId) -> bool {
|
||||
let key = (user_id, room_id);
|
||||
self.db.userroomid_leftstate.qry(&key).await.is_ok()
|
||||
}
|
||||
|
||||
#[implement(Service)]
|
||||
#[tracing::instrument(skip(self), level = "trace")]
|
||||
pub async fn invite_sender(&self, user_id: &UserId, room_id: &RoomId) -> Result<OwnedUserId> {
|
||||
let key = (user_id, room_id);
|
||||
self.db
|
||||
.userroomid_invitesender
|
||||
.qry(&key)
|
||||
.await
|
||||
.deserialized()
|
||||
}
|
||||
@@ -1,6 +1,6 @@
|
||||
use std::collections::HashSet;
|
||||
|
||||
use conduwuit::{Err, Event, Pdu, Result, implement, is_not_empty, utils::ReadyExt, warn};
|
||||
use conduwuit::{Err, Event, Pdu, Result, is_not_empty, utils::ReadyExt};
|
||||
use database::{Json, serialize_key};
|
||||
use futures::StreamExt;
|
||||
use ruma::{
|
||||
@@ -19,369 +19,360 @@
|
||||
};
|
||||
use ruminuwuity::invite_permission_config::FilterLevel;
|
||||
|
||||
/// Update current membership data.
|
||||
#[implement(super::Service)]
|
||||
#[tracing::instrument(
|
||||
level = "debug",
|
||||
skip_all,
|
||||
fields(
|
||||
%room_id,
|
||||
%user_id,
|
||||
?pdu,
|
||||
),
|
||||
)]
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub async fn update_membership(
|
||||
&self,
|
||||
room_id: &RoomId,
|
||||
user_id: &UserId,
|
||||
pdu: &Pdu,
|
||||
update_joined_count: bool,
|
||||
) -> Result {
|
||||
let membership = pdu.get_content::<RoomMemberEventContent>()?;
|
||||
impl super::Service {
|
||||
/// Update current membership data.
|
||||
pub async fn update_membership(
|
||||
&self,
|
||||
room_id: &RoomId,
|
||||
user_id: &UserId,
|
||||
pdu: &Pdu,
|
||||
update_joined_count: bool,
|
||||
) -> Result {
|
||||
let membership = pdu.get_content::<RoomMemberEventContent>()?;
|
||||
|
||||
// Keep track what remote users exist by adding them as "deactivated" users
|
||||
//
|
||||
// TODO: use futures to update remote profiles without blocking the membership
|
||||
// update
|
||||
#[allow(clippy::collapsible_if)]
|
||||
if !self.services.globals.user_is_local(user_id) {
|
||||
if !self.services.users.exists(user_id).await {
|
||||
// Keep track what remote users exist by adding them as "deactivated" users
|
||||
//
|
||||
// TODO: use futures to update remote profiles without blocking the membership
|
||||
// update
|
||||
#[allow(clippy::collapsible_if)]
|
||||
if !self.services.globals.user_is_local(user_id)
|
||||
&& !self.services.users.exists(user_id).await
|
||||
{
|
||||
self.services.users.create(user_id, None).await?;
|
||||
}
|
||||
}
|
||||
|
||||
match &membership.membership {
|
||||
| MembershipState::Join => {
|
||||
// Check if the user never joined this room
|
||||
if !self.once_joined(user_id, room_id).await {
|
||||
// Add the user ID to the join list then
|
||||
self.mark_as_once_joined(user_id, room_id);
|
||||
match &membership.membership {
|
||||
| MembershipState::Join => {
|
||||
// Check if the user never joined this room
|
||||
if !self.once_joined(user_id, room_id).await {
|
||||
// Add the user ID to the join list then
|
||||
self.mark_as_once_joined(user_id, room_id);
|
||||
|
||||
// Check if the room has a predecessor
|
||||
if let Ok(Some(predecessor)) = self
|
||||
.services
|
||||
.state_accessor
|
||||
.room_state_get_content(room_id, &StateEventType::RoomCreate, "")
|
||||
.await
|
||||
.map(|content: RoomCreateEventContent| content.predecessor)
|
||||
{
|
||||
// Copy old tags to new room
|
||||
if let Ok(tag_event) = self
|
||||
// Check if the room has a predecessor
|
||||
if let Ok(Some(predecessor)) = self
|
||||
.services
|
||||
.account_data
|
||||
.get_room(&predecessor.room_id, user_id, RoomAccountDataEventType::Tag)
|
||||
.state_accessor
|
||||
.room_state_get_content(room_id, &StateEventType::RoomCreate, "")
|
||||
.await
|
||||
.map(|content: RoomCreateEventContent| content.predecessor)
|
||||
{
|
||||
self.services
|
||||
// Copy old tags to new room
|
||||
if let Ok(tag_event) = self
|
||||
.services
|
||||
.account_data
|
||||
.update(
|
||||
Some(room_id),
|
||||
.get_room(
|
||||
&predecessor.room_id,
|
||||
user_id,
|
||||
RoomAccountDataEventType::Tag,
|
||||
&tag_event,
|
||||
)
|
||||
.await
|
||||
.ok();
|
||||
}
|
||||
|
||||
// Copy direct chat flag
|
||||
if let Ok(mut direct_event) = self
|
||||
.services
|
||||
.account_data
|
||||
.get_global::<DirectEvent>(user_id, GlobalAccountDataEventType::Direct)
|
||||
.await
|
||||
{
|
||||
let mut room_ids_updated = false;
|
||||
for room_ids in direct_event.content.0.values_mut() {
|
||||
if room_ids.iter().any(|r| r == &predecessor.room_id) {
|
||||
room_ids.push(room_id.to_owned());
|
||||
room_ids_updated = true;
|
||||
}
|
||||
}
|
||||
|
||||
if room_ids_updated {
|
||||
{
|
||||
self.services
|
||||
.account_data
|
||||
.update(
|
||||
None,
|
||||
Some(room_id),
|
||||
user_id,
|
||||
GlobalAccountDataEventType::Direct.to_string().into(),
|
||||
&serde_json::to_value(&direct_event)
|
||||
.expect("to json always works"),
|
||||
RoomAccountDataEventType::Tag,
|
||||
&tag_event,
|
||||
)
|
||||
.await?;
|
||||
.await
|
||||
.ok();
|
||||
}
|
||||
|
||||
// Copy direct chat flag
|
||||
if let Ok(mut direct_event) = self
|
||||
.services
|
||||
.account_data
|
||||
.get_global::<DirectEvent>(
|
||||
user_id,
|
||||
GlobalAccountDataEventType::Direct,
|
||||
)
|
||||
.await
|
||||
{
|
||||
let mut room_ids_updated = false;
|
||||
for room_ids in direct_event.content.0.values_mut() {
|
||||
if room_ids.iter().any(|r| r == &predecessor.room_id) {
|
||||
room_ids.push(room_id.to_owned());
|
||||
room_ids_updated = true;
|
||||
}
|
||||
}
|
||||
|
||||
if room_ids_updated {
|
||||
self.services
|
||||
.account_data
|
||||
.update(
|
||||
None,
|
||||
user_id,
|
||||
GlobalAccountDataEventType::Direct.to_string().into(),
|
||||
&serde_json::to_value(&direct_event)
|
||||
.expect("to json always works"),
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
self.mark_as_joined(user_id, room_id);
|
||||
},
|
||||
| MembershipState::Invite => {
|
||||
let last_state = self
|
||||
.services
|
||||
.state
|
||||
.summary_stripped(pdu, room_id, user_id)
|
||||
.await;
|
||||
self.mark_as_joined(user_id, room_id);
|
||||
},
|
||||
| MembershipState::Invite => {
|
||||
let last_state = self
|
||||
.services
|
||||
.state
|
||||
.summary_stripped(pdu, room_id, user_id)
|
||||
.await;
|
||||
|
||||
self.mark_as_invited(user_id, room_id, pdu.sender(), last_state, None)
|
||||
.await?;
|
||||
},
|
||||
| MembershipState::Leave | MembershipState::Ban => {
|
||||
self.mark_as_left(user_id, room_id, Some(pdu.clone())).await;
|
||||
},
|
||||
| _ => {},
|
||||
self.mark_as_invited(user_id, room_id, pdu.sender(), last_state, None)
|
||||
.await?;
|
||||
},
|
||||
| MembershipState::Leave | MembershipState::Ban => {
|
||||
self.mark_as_left(user_id, room_id, Some(pdu.clone())).await;
|
||||
},
|
||||
| _ => {},
|
||||
}
|
||||
|
||||
if update_joined_count {
|
||||
self.update_joined_count(room_id).await;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
if update_joined_count {
|
||||
self.update_joined_count(room_id).await;
|
||||
/// Updates membership counts for the target room.
|
||||
pub async fn update_joined_count(&self, room_id: &RoomId) {
|
||||
let mut joinedcount = 0_u64;
|
||||
let mut invitedcount = 0_u64;
|
||||
let mut knockedcount = 0_u64;
|
||||
let mut joined_servers = HashSet::new();
|
||||
|
||||
self.room_members(room_id)
|
||||
.ready_for_each(|joined| {
|
||||
joined_servers.insert(joined.server_name().to_owned());
|
||||
joinedcount = joinedcount.saturating_add(1);
|
||||
})
|
||||
.await;
|
||||
|
||||
invitedcount = invitedcount.saturating_add(
|
||||
self.room_members_invited(room_id)
|
||||
.count()
|
||||
.await
|
||||
.try_into()
|
||||
.unwrap_or(0),
|
||||
);
|
||||
|
||||
knockedcount = knockedcount.saturating_add(
|
||||
self.room_members_knocked(room_id)
|
||||
.count()
|
||||
.await
|
||||
.try_into()
|
||||
.unwrap_or(0),
|
||||
);
|
||||
|
||||
self.db.roomid_joinedcount.raw_put(room_id, joinedcount);
|
||||
self.db.roomid_invitedcount.raw_put(room_id, invitedcount);
|
||||
self.db
|
||||
.roomuserid_knockedcount
|
||||
.raw_put(room_id, knockedcount);
|
||||
|
||||
self.room_servers(room_id)
|
||||
.ready_for_each(|old_joined_server| {
|
||||
if joined_servers.remove(&old_joined_server) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Server not in room anymore
|
||||
let roomserver_id = (room_id, old_joined_server.clone());
|
||||
let serverroom_id = (old_joined_server, room_id);
|
||||
|
||||
self.db.roomserverids.del(roomserver_id);
|
||||
self.db.serverroomids.del(serverroom_id);
|
||||
})
|
||||
.await;
|
||||
|
||||
// Now only new servers are in joined_servers anymore
|
||||
for server in &joined_servers {
|
||||
let roomserver_id = (room_id, server);
|
||||
let serverroom_id = (server, room_id);
|
||||
|
||||
self.db.roomserverids.put_raw(roomserver_id, []);
|
||||
self.db.serverroomids.put_raw(serverroom_id, []);
|
||||
}
|
||||
|
||||
self.appservice_in_room_cache.write().remove(room_id);
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
/// Direct DB function to directly mark a user as joined. It is not
|
||||
/// recommended to use this directly. You most likely should use
|
||||
/// `update_membership` instead
|
||||
pub fn mark_as_joined(&self, user_id: &UserId, room_id: &RoomId) {
|
||||
let userroom_id = (user_id, room_id);
|
||||
let userroom_id = serialize_key(userroom_id).expect("failed to serialize userroom_id");
|
||||
|
||||
#[implement(super::Service)]
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
pub async fn update_joined_count(&self, room_id: &RoomId) {
|
||||
let mut joinedcount = 0_u64;
|
||||
let mut invitedcount = 0_u64;
|
||||
let mut knockedcount = 0_u64;
|
||||
let mut joined_servers = HashSet::new();
|
||||
let roomuser_id = (room_id, user_id);
|
||||
let roomuser_id = serialize_key(roomuser_id).expect("failed to serialize roomuser_id");
|
||||
|
||||
self.room_members(room_id)
|
||||
.ready_for_each(|joined| {
|
||||
joined_servers.insert(joined.server_name().to_owned());
|
||||
joinedcount = joinedcount.saturating_add(1);
|
||||
})
|
||||
.await;
|
||||
self.db.userroomid_joined.insert(&userroom_id, []);
|
||||
self.db.roomuserid_joined.insert(&roomuser_id, []);
|
||||
|
||||
invitedcount = invitedcount.saturating_add(
|
||||
self.room_members_invited(room_id)
|
||||
.count()
|
||||
.await
|
||||
.try_into()
|
||||
.unwrap_or(0),
|
||||
);
|
||||
self.db.userroomid_invitestate.remove(&userroom_id);
|
||||
self.db.roomuserid_invitecount.remove(&roomuser_id);
|
||||
self.db.userroomid_invitesender.remove(&userroom_id);
|
||||
|
||||
knockedcount = knockedcount.saturating_add(
|
||||
self.room_members_knocked(room_id)
|
||||
.count()
|
||||
.await
|
||||
.try_into()
|
||||
.unwrap_or(0),
|
||||
);
|
||||
self.db.userroomid_leftstate.remove(&userroom_id);
|
||||
self.db.roomuserid_leftcount.remove(&roomuser_id);
|
||||
|
||||
self.db.roomid_joinedcount.raw_put(room_id, joinedcount);
|
||||
self.db.roomid_invitedcount.raw_put(room_id, invitedcount);
|
||||
self.db
|
||||
.roomuserid_knockedcount
|
||||
.raw_put(room_id, knockedcount);
|
||||
self.db.userroomid_knockedstate.remove(&userroom_id);
|
||||
self.db.roomuserid_knockedcount.remove(&roomuser_id);
|
||||
|
||||
self.room_servers(room_id)
|
||||
.ready_for_each(|old_joined_server| {
|
||||
if joined_servers.remove(&old_joined_server) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Server not in room anymore
|
||||
let roomserver_id = (room_id, old_joined_server.clone());
|
||||
let serverroom_id = (old_joined_server, room_id);
|
||||
|
||||
self.db.roomserverids.del(roomserver_id);
|
||||
self.db.serverroomids.del(serverroom_id);
|
||||
})
|
||||
.await;
|
||||
|
||||
// Now only new servers are in joined_servers anymore
|
||||
for server in &joined_servers {
|
||||
let roomserver_id = (room_id, server);
|
||||
let serverroom_id = (server, room_id);
|
||||
|
||||
self.db.roomserverids.put_raw(roomserver_id, []);
|
||||
self.db.serverroomids.put_raw(serverroom_id, []);
|
||||
self.db.roomid_inviteviaservers.remove(room_id);
|
||||
}
|
||||
|
||||
self.appservice_in_room_cache.write().remove(room_id);
|
||||
}
|
||||
/// Mark a user as having left a room.
|
||||
///
|
||||
/// `leave_pdu` represents the m.room.member event which the user sent to
|
||||
/// leave the room. If this is None, no event was actually sent, but we
|
||||
/// must still behave as if the user is no longer in the room. This may
|
||||
/// occur, for example, if the room being left has been server-banned by an
|
||||
/// administrator.
|
||||
pub async fn mark_as_left(&self, user_id: &UserId, room_id: &RoomId, leave_pdu: Option<Pdu>) {
|
||||
let userroom_id = (user_id, room_id);
|
||||
let userroom_id = serialize_key(userroom_id).expect("failed to serialize userroom_id");
|
||||
|
||||
/// Direct DB function to directly mark a user as joined. It is not
|
||||
/// recommended to use this directly. You most likely should use
|
||||
/// `update_membership` instead
|
||||
#[implement(super::Service)]
|
||||
#[tracing::instrument(skip(self), level = "debug")]
|
||||
pub fn mark_as_joined(&self, user_id: &UserId, room_id: &RoomId) {
|
||||
let userroom_id = (user_id, room_id);
|
||||
let userroom_id = serialize_key(userroom_id).expect("failed to serialize userroom_id");
|
||||
let roomuser_id = (room_id, user_id);
|
||||
let roomuser_id = serialize_key(roomuser_id).expect("failed to serialize roomuser_id");
|
||||
|
||||
let roomuser_id = (room_id, user_id);
|
||||
let roomuser_id = serialize_key(roomuser_id).expect("failed to serialize roomuser_id");
|
||||
self.db
|
||||
.userroomid_leftstate
|
||||
.raw_put(&userroom_id, Json(leave_pdu));
|
||||
self.db
|
||||
.roomuserid_leftcount
|
||||
.raw_aput::<8, _, _>(&roomuser_id, self.services.globals.next_count().unwrap());
|
||||
|
||||
self.db.userroomid_joined.insert(&userroom_id, []);
|
||||
self.db.roomuserid_joined.insert(&roomuser_id, []);
|
||||
self.db.userroomid_joined.remove(&userroom_id);
|
||||
self.db.roomuserid_joined.remove(&roomuser_id);
|
||||
|
||||
self.db.userroomid_invitestate.remove(&userroom_id);
|
||||
self.db.roomuserid_invitecount.remove(&roomuser_id);
|
||||
self.db.userroomid_invitesender.remove(&userroom_id);
|
||||
self.db.userroomid_invitestate.remove(&userroom_id);
|
||||
self.db.roomuserid_invitecount.remove(&roomuser_id);
|
||||
self.db.userroomid_invitesender.remove(&userroom_id);
|
||||
|
||||
self.db.userroomid_leftstate.remove(&userroom_id);
|
||||
self.db.roomuserid_leftcount.remove(&roomuser_id);
|
||||
self.db.userroomid_knockedstate.remove(&userroom_id);
|
||||
self.db.roomuserid_knockedcount.remove(&roomuser_id);
|
||||
|
||||
self.db.userroomid_knockedstate.remove(&userroom_id);
|
||||
self.db.roomuserid_knockedcount.remove(&roomuser_id);
|
||||
self.db.roomid_inviteviaservers.remove(room_id);
|
||||
|
||||
self.db.roomid_inviteviaservers.remove(room_id);
|
||||
}
|
||||
|
||||
/// Mark a user as having left a room.
|
||||
///
|
||||
/// `leave_pdu` represents the m.room.member event which the user sent to leave
|
||||
/// the room. If this is None, no event was actually sent, but we must still
|
||||
/// behave as if the user is no longer in the room. This may occur, for example,
|
||||
/// if the room being left has been server-banned by an administrator.
|
||||
#[implement(super::Service)]
|
||||
#[tracing::instrument(skip(self), level = "debug")]
|
||||
pub async fn mark_as_left(&self, user_id: &UserId, room_id: &RoomId, leave_pdu: Option<Pdu>) {
|
||||
let userroom_id = (user_id, room_id);
|
||||
let userroom_id = serialize_key(userroom_id).expect("failed to serialize userroom_id");
|
||||
|
||||
let roomuser_id = (room_id, user_id);
|
||||
let roomuser_id = serialize_key(roomuser_id).expect("failed to serialize roomuser_id");
|
||||
|
||||
self.db
|
||||
.userroomid_leftstate
|
||||
.raw_put(&userroom_id, Json(leave_pdu));
|
||||
self.db
|
||||
.roomuserid_leftcount
|
||||
.raw_aput::<8, _, _>(&roomuser_id, self.services.globals.next_count().unwrap());
|
||||
|
||||
self.db.userroomid_joined.remove(&userroom_id);
|
||||
self.db.roomuserid_joined.remove(&roomuser_id);
|
||||
|
||||
self.db.userroomid_invitestate.remove(&userroom_id);
|
||||
self.db.roomuserid_invitecount.remove(&roomuser_id);
|
||||
self.db.userroomid_invitesender.remove(&userroom_id);
|
||||
|
||||
self.db.userroomid_knockedstate.remove(&userroom_id);
|
||||
self.db.roomuserid_knockedcount.remove(&roomuser_id);
|
||||
|
||||
self.db.roomid_inviteviaservers.remove(room_id);
|
||||
|
||||
if self.services.globals.user_is_local(user_id)
|
||||
&& (self.services.config.forget_forced_upon_leave
|
||||
|| self.services.metadata.is_banned(room_id).await
|
||||
|| self.services.metadata.is_disabled(room_id).await)
|
||||
{
|
||||
self.forget(room_id, user_id);
|
||||
if self.services.globals.user_is_local(user_id)
|
||||
&& (self.services.config.forget_forced_upon_leave
|
||||
|| self.services.metadata.is_banned(room_id).await
|
||||
|| self.services.metadata.is_disabled(room_id).await)
|
||||
{
|
||||
self.forget(room_id, user_id);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Direct DB function to directly mark a user as knocked. It is not
|
||||
/// recommended to use this directly. You most likely should use
|
||||
/// `update_membership` instead
|
||||
#[implement(super::Service)]
|
||||
#[tracing::instrument(skip(self), level = "debug")]
|
||||
pub fn mark_as_knocked(
|
||||
&self,
|
||||
user_id: &UserId,
|
||||
room_id: &RoomId,
|
||||
knocked_state: Option<Vec<Raw<AnyStrippedStateEvent>>>,
|
||||
) {
|
||||
let userroom_id = (user_id, room_id);
|
||||
let userroom_id = serialize_key(userroom_id).expect("failed to serialize userroom_id");
|
||||
|
||||
let roomuser_id = (room_id, user_id);
|
||||
let roomuser_id = serialize_key(roomuser_id).expect("failed to serialize roomuser_id");
|
||||
|
||||
self.db
|
||||
.userroomid_knockedstate
|
||||
.raw_put(&userroom_id, Json(knocked_state.unwrap_or_default()));
|
||||
self.db
|
||||
.roomuserid_knockedcount
|
||||
.raw_aput::<8, _, _>(&roomuser_id, self.services.globals.next_count().unwrap());
|
||||
|
||||
self.db.userroomid_joined.remove(&userroom_id);
|
||||
self.db.roomuserid_joined.remove(&roomuser_id);
|
||||
|
||||
self.db.userroomid_invitestate.remove(&userroom_id);
|
||||
self.db.roomuserid_invitecount.remove(&roomuser_id);
|
||||
self.db.userroomid_invitesender.remove(&userroom_id);
|
||||
|
||||
self.db.userroomid_leftstate.remove(&userroom_id);
|
||||
self.db.roomuserid_leftcount.remove(&roomuser_id);
|
||||
|
||||
self.db.roomid_inviteviaservers.remove(room_id);
|
||||
}
|
||||
|
||||
/// Makes a user forget a room.
|
||||
#[implement(super::Service)]
|
||||
#[tracing::instrument(skip(self), level = "debug")]
|
||||
pub fn forget(&self, room_id: &RoomId, user_id: &UserId) {
|
||||
let userroom_id = (user_id, room_id);
|
||||
let roomuser_id = (room_id, user_id);
|
||||
|
||||
self.db.userroomid_leftstate.del(userroom_id);
|
||||
self.db.roomuserid_leftcount.del(roomuser_id);
|
||||
}
|
||||
|
||||
#[implement(super::Service)]
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
fn mark_as_once_joined(&self, user_id: &UserId, room_id: &RoomId) {
|
||||
let key = (user_id, room_id);
|
||||
self.db.roomuseroncejoinedids.put_raw(key, []);
|
||||
}
|
||||
|
||||
#[implement(super::Service)]
|
||||
#[tracing::instrument(level = "debug", skip(self, last_state, invite_via))]
|
||||
pub async fn mark_as_invited(
|
||||
&self,
|
||||
user_id: &UserId,
|
||||
room_id: &RoomId,
|
||||
sender_user: &UserId,
|
||||
last_state: Vec<RawStrippedState>,
|
||||
invite_via: Option<Vec<OwnedServerName>>,
|
||||
) -> Result<()> {
|
||||
// return an error for blocked invites. ignored invites aren't handled here
|
||||
// since the recipient's membership should still be changed to `invite`.
|
||||
// they're filtered out in the individual /sync handlers
|
||||
if matches!(
|
||||
self.services
|
||||
.users
|
||||
.invite_filter_level(sender_user, user_id)
|
||||
.await,
|
||||
FilterLevel::Block
|
||||
/// Direct DB function to directly mark a user as knocked. It is not
|
||||
/// recommended to use this directly. You most likely should use
|
||||
/// `update_membership` instead
|
||||
pub fn mark_as_knocked(
|
||||
&self,
|
||||
user_id: &UserId,
|
||||
room_id: &RoomId,
|
||||
knocked_state: Option<Vec<Raw<AnyStrippedStateEvent>>>,
|
||||
) {
|
||||
return Err!(Request(InviteBlocked("{user_id} has blocked invites from {sender_user}.")));
|
||||
let userroom_id = (user_id, room_id);
|
||||
let userroom_id = serialize_key(userroom_id).expect("failed to serialize userroom_id");
|
||||
|
||||
let roomuser_id = (room_id, user_id);
|
||||
let roomuser_id = serialize_key(roomuser_id).expect("failed to serialize roomuser_id");
|
||||
|
||||
self.db
|
||||
.userroomid_knockedstate
|
||||
.raw_put(&userroom_id, Json(knocked_state.unwrap_or_default()));
|
||||
self.db
|
||||
.roomuserid_knockedcount
|
||||
.raw_aput::<8, _, _>(&roomuser_id, self.services.globals.next_count().unwrap());
|
||||
|
||||
self.db.userroomid_joined.remove(&userroom_id);
|
||||
self.db.roomuserid_joined.remove(&roomuser_id);
|
||||
|
||||
self.db.userroomid_invitestate.remove(&userroom_id);
|
||||
self.db.roomuserid_invitecount.remove(&roomuser_id);
|
||||
self.db.userroomid_invitesender.remove(&userroom_id);
|
||||
|
||||
self.db.userroomid_leftstate.remove(&userroom_id);
|
||||
self.db.roomuserid_leftcount.remove(&roomuser_id);
|
||||
|
||||
self.db.roomid_inviteviaservers.remove(room_id);
|
||||
}
|
||||
|
||||
let roomuser_id = (room_id, user_id);
|
||||
let roomuser_id = serialize_key(roomuser_id).expect("failed to serialize roomuser_id");
|
||||
/// Makes a user forget a room.
|
||||
pub fn forget(&self, room_id: &RoomId, user_id: &UserId) {
|
||||
let userroom_id = (user_id, room_id);
|
||||
let roomuser_id = (room_id, user_id);
|
||||
|
||||
let userroom_id = (user_id, room_id);
|
||||
let userroom_id = serialize_key(userroom_id).expect("failed to serialize userroom_id");
|
||||
|
||||
self.db
|
||||
.userroomid_invitestate
|
||||
.raw_put(&userroom_id, Json(last_state));
|
||||
self.db
|
||||
.roomuserid_invitecount
|
||||
.raw_aput::<8, _, _>(&roomuser_id, self.services.globals.next_count().unwrap());
|
||||
self.db
|
||||
.userroomid_invitesender
|
||||
.insert(&userroom_id, sender_user);
|
||||
|
||||
self.db.userroomid_joined.remove(&userroom_id);
|
||||
self.db.roomuserid_joined.remove(&roomuser_id);
|
||||
|
||||
self.db.userroomid_leftstate.remove(&userroom_id);
|
||||
self.db.roomuserid_leftcount.remove(&roomuser_id);
|
||||
|
||||
self.db.userroomid_knockedstate.remove(&userroom_id);
|
||||
self.db.roomuserid_knockedcount.remove(&roomuser_id);
|
||||
|
||||
if let Some(servers) = invite_via.filter(is_not_empty!()) {
|
||||
self.add_servers_invite_via(room_id, servers).await;
|
||||
self.db.userroomid_leftstate.del(userroom_id);
|
||||
self.db.roomuserid_leftcount.del(roomuser_id);
|
||||
}
|
||||
|
||||
Ok(())
|
||||
/// Immutably a user as having joined a room.
|
||||
fn mark_as_once_joined(&self, user_id: &UserId, room_id: &RoomId) {
|
||||
let key = (user_id, room_id);
|
||||
self.db.roomuseroncejoinedids.put_raw(key, []);
|
||||
}
|
||||
|
||||
/// Marks a user as having been invited to a room, saving the relevant
|
||||
/// invite state.
|
||||
pub async fn mark_as_invited(
|
||||
&self,
|
||||
user_id: &UserId,
|
||||
room_id: &RoomId,
|
||||
sender_user: &UserId,
|
||||
last_state: Vec<RawStrippedState>,
|
||||
invite_via: Option<Vec<OwnedServerName>>,
|
||||
) -> Result<()> {
|
||||
// return an error for blocked invites. ignored invites aren't handled here
|
||||
// since the recipient's membership should still be changed to `invite`.
|
||||
// they're filtered out in the individual /sync handlers
|
||||
if matches!(
|
||||
self.services
|
||||
.users
|
||||
.invite_filter_level(sender_user, user_id)
|
||||
.await,
|
||||
FilterLevel::Block
|
||||
) {
|
||||
return Err!(Request(InviteBlocked(
|
||||
"{user_id} has blocked invites from {sender_user}."
|
||||
)));
|
||||
}
|
||||
|
||||
let roomuser_id = (room_id, user_id);
|
||||
let roomuser_id = serialize_key(roomuser_id).expect("failed to serialize roomuser_id");
|
||||
|
||||
let userroom_id = (user_id, room_id);
|
||||
let userroom_id = serialize_key(userroom_id).expect("failed to serialize userroom_id");
|
||||
|
||||
self.db
|
||||
.userroomid_invitestate
|
||||
.raw_put(&userroom_id, Json(last_state));
|
||||
self.db
|
||||
.roomuserid_invitecount
|
||||
.raw_aput::<8, _, _>(&roomuser_id, self.services.globals.next_count().unwrap());
|
||||
self.db
|
||||
.userroomid_invitesender
|
||||
.insert(&userroom_id, sender_user);
|
||||
|
||||
self.db.userroomid_joined.remove(&userroom_id);
|
||||
self.db.roomuserid_joined.remove(&roomuser_id);
|
||||
|
||||
self.db.userroomid_leftstate.remove(&userroom_id);
|
||||
self.db.roomuserid_leftcount.remove(&roomuser_id);
|
||||
|
||||
self.db.userroomid_knockedstate.remove(&userroom_id);
|
||||
self.db.roomuserid_knockedcount.remove(&roomuser_id);
|
||||
|
||||
if let Some(servers) = invite_via.filter(is_not_empty!()) {
|
||||
self.add_servers_invite_via(room_id, servers).await;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@@ -1,7 +1,6 @@
|
||||
use conduwuit::{
|
||||
Result, implement,
|
||||
Result,
|
||||
utils::{StreamTools, stream::TryIgnore},
|
||||
warn,
|
||||
};
|
||||
use database::Ignore;
|
||||
use futures::{Stream, StreamExt, stream::iter};
|
||||
@@ -12,80 +11,78 @@
|
||||
int,
|
||||
};
|
||||
|
||||
#[implement(super::Service)]
|
||||
#[tracing::instrument(level = "debug", skip(self, servers))]
|
||||
pub async fn add_servers_invite_via(&self, room_id: &RoomId, servers: Vec<OwnedServerName>) {
|
||||
let mut servers: Vec<_> = self
|
||||
.servers_invite_via(room_id)
|
||||
.chain(iter(servers.into_iter()))
|
||||
.collect()
|
||||
.await;
|
||||
impl super::Service {
|
||||
/// Adds N servers that are likely to be in the room in the future.
|
||||
pub async fn add_servers_invite_via(&self, room_id: &RoomId, servers: Vec<OwnedServerName>) {
|
||||
let mut servers: Vec<_> = self
|
||||
.servers_invite_via(room_id)
|
||||
.chain(iter(servers))
|
||||
.collect()
|
||||
.await;
|
||||
|
||||
servers.sort_unstable();
|
||||
servers.dedup();
|
||||
servers.sort_unstable();
|
||||
servers.dedup();
|
||||
|
||||
let servers = servers
|
||||
.iter()
|
||||
.map(|server| server.as_bytes())
|
||||
.collect_vec()
|
||||
.join(&[0xFF][..]);
|
||||
let servers = servers
|
||||
.iter()
|
||||
.map(|server| server.as_bytes())
|
||||
.collect_vec()
|
||||
.join(&[0xFF][..]);
|
||||
|
||||
self.db
|
||||
.roomid_inviteviaservers
|
||||
.insert(room_id.as_bytes(), &servers);
|
||||
}
|
||||
|
||||
/// Gets up to five servers that are likely to be in the room in the
|
||||
/// distant future.
|
||||
///
|
||||
/// See <https://spec.matrix.org/latest/appendices/#routing>
|
||||
#[implement(super::Service)]
|
||||
#[tracing::instrument(skip(self), level = "trace")]
|
||||
pub async fn servers_route_via(&self, room_id: &RoomId) -> Result<Vec<OwnedServerName>> {
|
||||
let most_powerful_user_server = self
|
||||
.services
|
||||
.state_accessor
|
||||
.room_state_get_content(room_id, &StateEventType::RoomPowerLevels, "")
|
||||
.await
|
||||
.map(|content: RoomPowerLevelsEventContent| {
|
||||
content
|
||||
.users
|
||||
.iter()
|
||||
.max_by_key(|(_, power)| *power)
|
||||
.and_then(|x| (x.1 >= &int!(50)).then_some(x))
|
||||
.map(|(user, _power)| user.server_name().to_owned())
|
||||
});
|
||||
|
||||
let mut servers: Vec<OwnedServerName> = self
|
||||
.room_members(room_id)
|
||||
.counts_by(|user| user.server_name().to_owned())
|
||||
.await
|
||||
.into_iter()
|
||||
.sorted_by_key(|(_, users)| *users)
|
||||
.map(|(server, _)| server)
|
||||
.rev()
|
||||
.take(5)
|
||||
.collect();
|
||||
|
||||
if let Ok(Some(server)) = most_powerful_user_server {
|
||||
servers.insert(0, server);
|
||||
servers.truncate(5);
|
||||
self.db
|
||||
.roomid_inviteviaservers
|
||||
.insert(room_id.as_bytes(), &servers);
|
||||
}
|
||||
|
||||
Ok(servers)
|
||||
}
|
||||
/// Gets up to five servers that are likely to be in the room in the
|
||||
/// distant future.
|
||||
///
|
||||
/// See <https://spec.matrix.org/latest/appendices/#routing>
|
||||
pub async fn servers_route_via(&self, room_id: &RoomId) -> Result<Vec<OwnedServerName>> {
|
||||
let most_powerful_user_server = self
|
||||
.services
|
||||
.state_accessor
|
||||
.room_state_get_content(room_id, &StateEventType::RoomPowerLevels, "")
|
||||
.await
|
||||
.map(|content: RoomPowerLevelsEventContent| {
|
||||
content
|
||||
.users
|
||||
.iter()
|
||||
.max_by_key(|(_, power)| *power)
|
||||
.and_then(|x| (x.1 >= &int!(50)).then_some(x))
|
||||
.map(|(user, _power)| user.server_name().to_owned())
|
||||
});
|
||||
|
||||
#[implement(super::Service)]
|
||||
#[tracing::instrument(skip(self), level = "debug")]
|
||||
pub fn servers_invite_via<'a>(
|
||||
&'a self,
|
||||
room_id: &'a RoomId,
|
||||
) -> impl Stream<Item = OwnedServerName> + Send + 'a {
|
||||
type KeyVal = (Ignore, Vec<OwnedServerName>);
|
||||
let mut servers: Vec<OwnedServerName> = self
|
||||
.room_members(room_id)
|
||||
.counts_by(|user| user.server_name().to_owned())
|
||||
.await
|
||||
.into_iter()
|
||||
.sorted_by_key(|(_, users)| *users)
|
||||
.map(|(server, _)| server)
|
||||
.rev()
|
||||
.take(5)
|
||||
.collect();
|
||||
|
||||
self.db
|
||||
.roomid_inviteviaservers
|
||||
.stream_raw_prefix(room_id)
|
||||
.ignore_err()
|
||||
.map(|(_, mut servers): KeyVal| servers.pop().expect("at least one server"))
|
||||
if let Ok(Some(server)) = most_powerful_user_server {
|
||||
servers.insert(0, server);
|
||||
servers.truncate(5);
|
||||
}
|
||||
|
||||
Ok(servers)
|
||||
}
|
||||
|
||||
/// Fetches a list of servers that may be in the room as of an invite.
|
||||
pub fn servers_invite_via<'a>(
|
||||
&'a self,
|
||||
room_id: &'a RoomId,
|
||||
) -> impl Stream<Item = OwnedServerName> + Send + 'a {
|
||||
type KeyVal = (Ignore, Vec<OwnedServerName>);
|
||||
|
||||
self.db
|
||||
.roomid_inviteviaservers
|
||||
.stream_raw_prefix(room_id)
|
||||
.ignore_err()
|
||||
.map(|(_, mut servers): KeyVal| servers.pop().expect("at least one server"))
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user