serv/conduit
1
0
Fork 0
mirror of https://gitlab.com/famedly/conduit.git synced 2025-04-22 14:10:16 +03:00
Code Issues Projects Releases Packages Wiki Activity Actions

refactor: move duplicate code and some other small optimizations

Browse source
This commit is contained in:
Matthias Ahouansou 2025-03-01 19:17:32 +00:00
parent 8acacdebc8
commit f4d90e9989
No known key found for this signature in database
9 changed files with 283 additions and 354 deletions
src
api/client_server
membership.rs
database
key_value/rooms
mod.rsstate_cache.rstimeline.rsuser.rs
mod.rs
service/rooms
state
mod.rs
state_cache
mod.rs
timeline
mod.rs

339
src/api/client_server/membership.rs
View file

@ -18,8 +18,10 @@ use ruma::{
},
StateEventType, TimelineEventType,
},
serde::Raw,
state_res, CanonicalJsonObject, CanonicalJsonValue, EventId, MilliSecondsSinceUnixEpoch,
OwnedEventId, OwnedRoomId, OwnedServerName, OwnedUserId, RoomId, RoomVersionId, UserId,
OwnedEventId, OwnedRoomId, OwnedRoomOrAliasId, OwnedServerName, OwnedUserId, RoomId,
RoomVersionId, UserId,
};
use serde_json::value::{to_raw_value, RawValue as RawJsonValue};
use std::{
@ -74,7 +76,7 @@ pub async fn join_room_by_id_route(
);
join_room_by_id_helper(
body.sender_user.as_deref(),
body.sender_user.as_deref().expect("user is authenticated"),
&body.room_id,
body.reason.clone(),
&servers,
@ -95,9 +97,122 @@ pub async fn join_room_by_id_or_alias_route(
let sender_user = body.sender_user.as_deref().expect("user is authenticated");
let body = body.body;
let (servers, room_id) = match OwnedRoomId::try_from(body.room_id_or_alias) {
let (servers, room_id) =
get_room_id_and_via_servers(sender_user, body.room_id_or_alias, body.via).await?;
let join_room_response = join_room_by_id_helper(
sender_user,
&room_id,
body.reason.clone(),
&servers,
body.third_party_signed.as_ref(),
)
.await?;
Ok(join_room_by_id_or_alias::v3::Response {
room_id: join_room_response.room_id,
})
}
/// Takes a membership template, as returned from the `/federation/*/make_*` endpoints, and
/// populates them to the point as to where they are a full pdu, ready to be appended to the timeline
///
/// Returns the event id, the pdu, and whether this event is a restricted join
fn populate_membership_template(
member_template: &RawJsonValue,
sender_user: &UserId,
reason: Option<String>,
room_version_id: &RoomVersionId,
membership: MembershipState,
) -> Result<(OwnedEventId, BTreeMap<String, CanonicalJsonValue>, bool), Error> {
let mut member_event_stub: CanonicalJsonObject = serde_json::from_str(member_template.get())
.map_err(|_| {
Error::BadServerResponse("Invalid make_knock event json received from server.")
})?;
let join_authorized_via_users_server = member_event_stub
.get("content")
.map(|s| {
s.as_object()?
.get("join_authorised_via_users_server")?
.as_str()
})
.and_then(|s| OwnedUserId::try_from(s.unwrap_or_default()).ok());
let restricted_join = join_authorized_via_users_server.is_some();
member_event_stub.insert(
"origin".to_owned(),
CanonicalJsonValue::String(services().globals.server_name().as_str().to_owned()),
);
member_event_stub.insert(
"origin_server_ts".to_owned(),
CanonicalJsonValue::Integer(
utils::millis_since_unix_epoch()
.try_into()
.expect("Timestamp is valid js_int value"),
),
);
member_event_stub.insert(
"content".to_owned(),
to_canonical_value(RoomMemberEventContent {
membership,
displayname: services().users.displayname(sender_user)?,
avatar_url: services().users.avatar_url(sender_user)?,
is_direct: None,
third_party_invite: None,
blurhash: services().users.blurhash(sender_user)?,
reason: reason.clone(),
join_authorized_via_users_server,
})
.expect("event is valid, we just created it"),
);
member_event_stub.remove("event_id");
ruma::signatures::hash_and_sign_event(
services().globals.server_name().as_str(),
services().globals.keypair(),
&mut member_event_stub,
room_version_id,
)
.expect("event is valid, we just created it");
let event_id = format!(
"${}",
ruma::signatures::reference_hash(&member_event_stub, room_version_id)
.expect("Event format validated when event was hashed")
);
let event_id =
<OwnedEventId>::try_from(event_id).expect("ruma's reference hashes are valid event ids");
member_event_stub.insert(
"event_id".to_owned(),
CanonicalJsonValue::String(event_id.as_str().to_owned()),
);
Ok((event_id, member_event_stub, restricted_join))
}
/// Function to assist performing a membership event that may require help from a remote server
///
/// If a room id is provided, the servers returned will consist of:
/// - the `via` argument, provided by the client
/// - servers of the senders of the stripped state events we are given
/// - the server in the room id
///
/// Otherwise, the servers returned will come from the response when resolving the alias.
async fn get_room_id_and_via_servers(
sender_user: &UserId,
room_id_or_alias: OwnedRoomOrAliasId,
via: Vec<OwnedServerName>,
) -> Result<(Vec<OwnedServerName>, OwnedRoomId), Error> {
let (servers, room_id) = match OwnedRoomId::try_from(room_id_or_alias) {
Ok(room_id) => {
let mut servers = body.via.clone();
let mut servers = via.clone();
servers.extend(
services()
.rooms
@ -127,19 +242,7 @@ pub async fn join_room_by_id_or_alias_route(
(response.servers, response.room_id)
}
};
let join_room_response = join_room_by_id_helper(
Some(sender_user),
&room_id,
body.reason.clone(),
&servers,
body.third_party_signed.as_ref(),
)
.await?;
Ok(join_room_by_id_or_alias::v3::Response {
room_id: join_room_response.room_id,
})
Ok((servers, room_id))
}
/// # `POST /_matrix/client/r0/rooms/{roomId}/leave`
@ -519,14 +622,12 @@ pub async fn joined_members_route(
}
async fn join_room_by_id_helper(
sender_user: Option<&UserId>,
sender_user: &UserId,
room_id: &RoomId,
reason: Option<String>,
servers: &[OwnedServerName],
_third_party_signed: Option<&ThirdPartySigned>,
) -> Result<join_room_by_id::v3::Response> {
let sender_user = sender_user.expect("user is authenticated");
if let Ok(true) = services().rooms.state_cache.is_joined(sender_user, room_id) {
return Ok(join_room_by_id::v3::Response {
room_id: room_id.into(),
@ -569,77 +670,13 @@ async fn join_room_by_id_helper(
_ => return Err(Error::BadServerResponse("Room version is not supported")),
};
let mut join_event_stub: CanonicalJsonObject =
serde_json::from_str(make_join_response.event.get()).map_err(|_| {
Error::BadServerResponse("Invalid make_join event json received from server.")
})?;
let join_authorized_via_users_server = join_event_stub
.get("content")
.map(|s| {
s.as_object()?
.get("join_authorised_via_users_server")?
.as_str()
})
.and_then(|s| OwnedUserId::try_from(s.unwrap_or_default()).ok());
// TODO: Is origin needed?
join_event_stub.insert(
"origin".to_owned(),
CanonicalJsonValue::String(services().globals.server_name().as_str().to_owned()),
);
join_event_stub.insert(
"origin_server_ts".to_owned(),
CanonicalJsonValue::Integer(
utils::millis_since_unix_epoch()
.try_into()
.expect("Timestamp is valid js_int value"),
),
);
join_event_stub.insert(
"content".to_owned(),
to_canonical_value(RoomMemberEventContent {
membership: MembershipState::Join,
displayname: services().users.displayname(sender_user)?,
avatar_url: services().users.avatar_url(sender_user)?,
is_direct: None,
third_party_invite: None,
blurhash: services().users.blurhash(sender_user)?,
reason,
join_authorized_via_users_server,
})
.expect("event is valid, we just created it"),
);
// We don't leave the event id in the pdu because that's only allowed in v1 or v2 rooms
join_event_stub.remove("event_id");
// In order to create a compatible ref hash (EventID) the `hashes` field needs to be present
ruma::signatures::hash_and_sign_event(
services().globals.server_name().as_str(),
services().globals.keypair(),
&mut join_event_stub,
let (event_id, mut join_event, _) = populate_membership_template(
&make_join_response.event,
sender_user,
reason,
&room_version_id,
)
.expect("event is valid, we just created it");
// Generate event id
let event_id = format!(
"${}",
ruma::signatures::reference_hash(&join_event_stub, &room_version_id)
.expect("Event format validated when event was hashed")
);
let event_id = <&EventId>::try_from(event_id.as_str())
.expect("ruma's reference hashes are valid event ids");
// Add event_id back
join_event_stub.insert(
"event_id".to_owned(),
CanonicalJsonValue::String(event_id.as_str().to_owned()),
);
// It has enough fields to be called a proper event now
let mut join_event = join_event_stub;
MembershipState::Join,
)?;
info!("Asking {remote_server} for send_join");
let send_join_response = services()
@ -709,7 +746,7 @@ async fn join_room_by_id_helper(
services().rooms.short.get_or_create_shortroomid(room_id)?;
info!("Parsing join event");
let parsed_join_pdu = PduEvent::from_id_val(event_id, join_event.clone())
let parsed_join_pdu = PduEvent::from_id_val(&event_id, join_event.clone())
.map_err(|_| Error::BadServerResponse("Invalid join event PDU."))?;
let mut state = HashMap::new();
@ -854,21 +891,7 @@ async fn join_room_by_id_helper(
} else {
info!("We can join locally");
let join_rules_event = services().rooms.state_accessor.room_state_get(
room_id,
&StateEventType::RoomJoinRules,
"",
)?;
let join_rules_event_content: Option<RoomJoinRulesEventContent> = join_rules_event
.as_ref()
.map(|join_rules_event| {
serde_json::from_str(join_rules_event.content.get()).map_err(|e| {
warn!("Invalid join rules event: {}", e);
Error::bad_database("Invalid join rules event in db.")
})
})
.transpose()?;
let join_rules_event_content = get_join_rules(room_id)?;
let restriction_rooms = match join_rules_event_content {
Some(RoomJoinRulesEventContent {
@ -930,7 +953,7 @@ async fn join_room_by_id_helper(
};
// Try normal join first
let error = match services()
let Err(error) = services()
.rooms
.timeline
.build_and_append_pdu(
@ -947,9 +970,8 @@ async fn join_room_by_id_helper(
&state_lock,
)
.await
{
Ok(_event_id) => return Ok(join_room_by_id::v3::Response::new(room_id.to_owned())),
Err(e) => e,
else {
return Ok(join_room_by_id::v3::Response::new(room_id.to_owned()));
};
if !restriction_rooms.is_empty()
@ -974,77 +996,14 @@ async fn join_room_by_id_helper(
}
_ => return Err(Error::BadServerResponse("Room version is not supported")),
};
let mut join_event_stub: CanonicalJsonObject =
serde_json::from_str(make_join_response.event.get()).map_err(|_| {
Error::BadServerResponse("Invalid make_join event json received from server.")
})?;
let join_authorized_via_users_server = join_event_stub
.get("content")
.map(|s| {
s.as_object()?
.get("join_authorised_via_users_server")?
.as_str()
})
.and_then(|s| OwnedUserId::try_from(s.unwrap_or_default()).ok());
let restricted_join = join_authorized_via_users_server.is_some();
// TODO: Is origin needed?
join_event_stub.insert(
"origin".to_owned(),
CanonicalJsonValue::String(services().globals.server_name().as_str().to_owned()),
);
join_event_stub.insert(
"origin_server_ts".to_owned(),
CanonicalJsonValue::Integer(
utils::millis_since_unix_epoch()
.try_into()
.expect("Timestamp is valid js_int value"),
),
);
join_event_stub.insert(
"content".to_owned(),
to_canonical_value(RoomMemberEventContent {
membership: MembershipState::Join,
displayname: services().users.displayname(sender_user)?,
avatar_url: services().users.avatar_url(sender_user)?,
is_direct: None,
third_party_invite: None,
blurhash: services().users.blurhash(sender_user)?,
reason,
join_authorized_via_users_server,
})
.expect("event is valid, we just created it"),
);
// We don't leave the event id in the pdu because that's only allowed in v1 or v2 rooms
join_event_stub.remove("event_id");
// In order to create a compatible ref hash (EventID) the `hashes` field needs to be present
ruma::signatures::hash_and_sign_event(
services().globals.server_name().as_str(),
services().globals.keypair(),
&mut join_event_stub,
let (event_id, join_event, restricted_join) = populate_membership_template(
&make_join_response.event,
sender_user,
reason,
&room_version_id,
)
.expect("event is valid, we just created it");
// Generate event id
let event_id = format!(
"${}",
ruma::signatures::reference_hash(&join_event_stub, &room_version_id)
.expect("ruma can calculate reference hashes")
);
let event_id = OwnedEventId::try_from(event_id)
.expect("ruma's reference hashes are valid event ids");
// Add event_id back
join_event_stub.insert(
"event_id".to_owned(),
CanonicalJsonValue::String(event_id.as_str().to_owned()),
);
// It has enough fields to be called a proper event now
let join_event = join_event_stub;
MembershipState::Join,
)?;
let send_join_response = services()
.sending
@ -1093,6 +1052,26 @@ async fn join_room_by_id_helper(
Ok(join_room_by_id::v3::Response::new(room_id.to_owned()))
}
/// Returns the join rules event content of a room, if there are any and we are aware of it locally
fn get_join_rules(room_id: &RoomId) -> Result<Option<RoomJoinRulesEventContent>, Error> {
let join_rules_event = services().rooms.state_accessor.room_state_get(
room_id,
&StateEventType::RoomJoinRules,
"",
)?;
join_rules_event
.as_ref()
.map(|join_rules_event| {
serde_json::from_str::<RoomJoinRulesEventContent>(join_rules_event.content.get())
.map_err(|e| {
warn!("Invalid join rules event: {}", e);
Error::bad_database("Invalid join rules event in db.")
})
})
.transpose()
}
async fn make_join_request(
sender_user: &UserId,
room_id: &RoomId,
@ -1270,7 +1249,7 @@ pub(crate) async fn invite_helper<'a>(
&state_lock,
)?;
let invite_room_state = services().rooms.state.calculate_invite_state(&pdu)?;
let invite_room_state = services().rooms.state.stripped_state(&pdu.room_id)?;
drop(state_lock);

24
src/database/key_value/rooms/mod.rs
View file

@ -16,6 +16,30 @@ mod threads;
mod timeline;
mod user;
use ruma::{RoomId, UserId};
use crate::{database::KeyValueDatabase, service};
impl service::rooms::Data for KeyValueDatabase {}
/// Constructs roomuser_id and userroom_id respectively in byte form
fn get_room_and_user_byte_ids(room_id: &RoomId, user_id: &UserId) -> (Vec<u8>, Vec<u8>) {
(
get_roomuser_id_bytes(room_id, user_id),
get_userroom_id_bytes(user_id, room_id),
)
}
fn get_roomuser_id_bytes(room_id: &RoomId, user_id: &UserId) -> Vec<u8> {
let mut roomuser_id = room_id.as_bytes().to_vec();
roomuser_id.push(0xff);
roomuser_id.extend_from_slice(user_id.as_bytes());
roomuser_id
}
fn get_userroom_id_bytes(user_id: &UserId, room_id: &RoomId) -> Vec<u8> {
let mut userroom_id = user_id.as_bytes().to_vec();
userroom_id.push(0xff);
userroom_id.extend_from_slice(room_id.as_bytes());
userroom_id
}

161
src/database/key_value/rooms/state_cache.rs
View file

@ -7,27 +7,21 @@ use ruma::{
};
use crate::{
database::KeyValueDatabase,
database::{abstraction::KvTree, KeyValueDatabase},
service::{self, appservice::RegistrationInfo},
services, utils, Error, Result,
};
use super::{get_room_and_user_byte_ids, get_userroom_id_bytes};
impl service::rooms::state_cache::Data for KeyValueDatabase {
fn mark_as_once_joined(&self, user_id: &UserId, room_id: &RoomId) -> Result<()> {
let mut userroom_id = user_id.as_bytes().to_vec();
userroom_id.push(0xff);
userroom_id.extend_from_slice(room_id.as_bytes());
let userroom_id = get_userroom_id_bytes(user_id, room_id);
self.roomuseroncejoinedids.insert(&userroom_id, &[])
}
fn mark_as_joined(&self, user_id: &UserId, room_id: &RoomId) -> Result<()> {
let mut roomuser_id = room_id.as_bytes().to_vec();
roomuser_id.push(0xff);
roomuser_id.extend_from_slice(user_id.as_bytes());
let mut userroom_id = user_id.as_bytes().to_vec();
userroom_id.push(0xff);
userroom_id.extend_from_slice(room_id.as_bytes());
let (roomuser_id, userroom_id) = get_room_and_user_byte_ids(room_id, user_id);
self.userroomid_joined.insert(&userroom_id, &[])?;
self.roomuserid_joined.insert(&roomuser_id, &[])?;
@ -45,13 +39,7 @@ impl service::rooms::state_cache::Data for KeyValueDatabase {
room_id: &RoomId,
last_state: Option<Vec<Raw<AnyStrippedStateEvent>>>,
) -> Result<()> {
let mut roomuser_id = room_id.as_bytes().to_vec();
roomuser_id.push(0xff);
roomuser_id.extend_from_slice(user_id.as_bytes());
let mut userroom_id = user_id.as_bytes().to_vec();
userroom_id.push(0xff);
userroom_id.extend_from_slice(room_id.as_bytes());
let (roomuser_id, userroom_id) = get_room_and_user_byte_ids(room_id, user_id);
self.userroomid_invitestate.insert(
&userroom_id,
@ -70,14 +58,9 @@ impl service::rooms::state_cache::Data for KeyValueDatabase {
Ok(())
}
fn mark_as_left(&self, user_id: &UserId, room_id: &RoomId) -> Result<()> {
let mut roomuser_id = room_id.as_bytes().to_vec();
roomuser_id.push(0xff);
roomuser_id.extend_from_slice(user_id.as_bytes());
let mut userroom_id = user_id.as_bytes().to_vec();
userroom_id.push(0xff);
userroom_id.extend_from_slice(room_id.as_bytes());
fn mark_as_left(&self, user_id: &UserId, room_id: &RoomId) -> Result<()> {
let (roomuser_id, userroom_id) = get_room_and_user_byte_ids(room_id, user_id);
self.userroomid_leftstate.insert(
&userroom_id,
@ -225,13 +208,7 @@ impl service::rooms::state_cache::Data for KeyValueDatabase {
/// Makes a user forget a room.
#[tracing::instrument(skip(self))]
fn forget(&self, room_id: &RoomId, user_id: &UserId) -> Result<()> {
let mut userroom_id = user_id.as_bytes().to_vec();
userroom_id.push(0xff);
userroom_id.extend_from_slice(room_id.as_bytes());
let mut roomuser_id = room_id.as_bytes().to_vec();
roomuser_id.push(0xff);
roomuser_id.extend_from_slice(user_id.as_bytes());
let (roomuser_id, userroom_id) = get_room_and_user_byte_ids(room_id, user_id);
self.userroomid_leftstate.remove(&userroom_id)?;
self.roomuserid_leftcount.remove(&roomuser_id)?;
@ -460,34 +437,7 @@ impl service::rooms::state_cache::Data for KeyValueDatabase {
&'a self,
user_id: &UserId,
) -> Box<dyn Iterator<Item = Result<(OwnedRoomId, Vec<Raw<AnyStrippedStateEvent>>)>> + 'a> {
let mut prefix = user_id.as_bytes().to_vec();
prefix.push(0xff);
Box::new(
self.userroomid_invitestate
.scan_prefix(prefix)
.map(|(key, state)| {
let room_id = RoomId::parse(
utils::string_from_bytes(
key.rsplit(|&b| b == 0xff)
.next()
.expect("rsplit always returns an element"),
)
.map_err(|_| {
Error::bad_database("Room ID in userroomid_invited is invalid unicode.")
})?,
)
.map_err(|_| {
Error::bad_database("Room ID in userroomid_invited is invalid.")
})?;
let state = serde_json::from_slice(&state).map_err(|_| {
Error::bad_database("Invalid state in userroomid_invitestate.")
})?;
Ok((room_id, state))
}),
)
scan_userroom_id_memberstate_tree(user_id, &self.userroomid_invitestate)
}
#[tracing::instrument(skip(self))]
@ -539,69 +489,80 @@ impl service::rooms::state_cache::Data for KeyValueDatabase {
&'a self,
user_id: &UserId,
) -> Box<dyn Iterator<Item = Result<(OwnedRoomId, Vec<Raw<AnySyncStateEvent>>)>> + 'a> {
let mut prefix = user_id.as_bytes().to_vec();
prefix.push(0xff);
Box::new(
self.userroomid_leftstate
.scan_prefix(prefix)
.map(|(key, state)| {
let room_id = RoomId::parse(
utils::string_from_bytes(
key.rsplit(|&b| b == 0xff)
.next()
.expect("rsplit always returns an element"),
)
.map_err(|_| {
Error::bad_database("Room ID in userroomid_invited is invalid unicode.")
})?,
)
.map_err(|_| {
Error::bad_database("Room ID in userroomid_invited is invalid.")
})?;
let state = serde_json::from_slice(&state).map_err(|_| {
Error::bad_database("Invalid state in userroomid_leftstate.")
})?;
Ok((room_id, state))
}),
)
scan_userroom_id_memberstate_tree(user_id, &self.userroomid_leftstate)
}
#[tracing::instrument(skip(self))]
fn once_joined(&self, user_id: &UserId, room_id: &RoomId) -> Result<bool> {
let mut userroom_id = user_id.as_bytes().to_vec();
userroom_id.push(0xff);
userroom_id.extend_from_slice(room_id.as_bytes());
let userroom_id = get_userroom_id_bytes(user_id, room_id);
Ok(self.roomuseroncejoinedids.get(&userroom_id)?.is_some())
}
#[tracing::instrument(skip(self))]
fn is_joined(&self, user_id: &UserId, room_id: &RoomId) -> Result<bool> {
let mut userroom_id = user_id.as_bytes().to_vec();
userroom_id.push(0xff);
userroom_id.extend_from_slice(room_id.as_bytes());
let userroom_id = get_userroom_id_bytes(user_id, room_id);
Ok(self.userroomid_joined.get(&userroom_id)?.is_some())
}
#[tracing::instrument(skip(self))]
fn is_invited(&self, user_id: &UserId, room_id: &RoomId) -> Result<bool> {
let mut userroom_id = user_id.as_bytes().to_vec();
userroom_id.push(0xff);
userroom_id.extend_from_slice(room_id.as_bytes());
let userroom_id = get_userroom_id_bytes(user_id, room_id);
Ok(self.userroomid_invitestate.get(&userroom_id)?.is_some())
}
#[tracing::instrument(skip(self))]
fn is_knocked(&self, user_id: &UserId, room_id: &RoomId) -> Result<bool> {
let userroom_id = get_userroom_id_bytes(user_id, room_id);
Ok(self.userroomid_knockstate.get(&userroom_id)?.is_some())
}
#[tracing::instrument(skip(self))]
fn is_left(&self, user_id: &UserId, room_id: &RoomId) -> Result<bool> {
let mut userroom_id = user_id.as_bytes().to_vec();
userroom_id.push(0xff);
userroom_id.extend_from_slice(room_id.as_bytes());
let userroom_id = get_userroom_id_bytes(user_id, room_id);
Ok(self.userroomid_leftstate.get(&userroom_id)?.is_some())
}
}
/// Scans the given userroom_id_`member`state tree for rooms, returning an iterator of room_ids
/// and a vector of raw state events
#[allow(clippy::type_complexity)]
fn scan_userroom_id_memberstate_tree<'a, T>(
user_id: &UserId,
userroom_id_memberstate_tree: &'a Arc<dyn KvTree>,
) -> Box<dyn Iterator<Item = Result<(OwnedRoomId, Vec<Raw<T>>)>> + 'a> {
let mut prefix = user_id.as_bytes().to_vec();
prefix.push(0xff);
Box::new(
userroom_id_memberstate_tree
.scan_prefix(prefix)
.map(|(key, state)| {
let room_id = RoomId::parse(
utils::string_from_bytes(
key.rsplit(|&b| b == 0xff)
.next()
.expect("rsplit always returns an element"),
)
.map_err(|_| {
Error::bad_database(
"Room ID in userroomid_<member>state is invalid unicode.",
)
})?,
)
.map_err(|_| {
Error::bad_database("Room ID in userroomid_<member>state is invalid.")
})?;
let state = serde_json::from_slice(&state).map_err(|_| {
Error::bad_database("Invalid state in userroomid_<member>state.")
})?;
Ok((room_id, state))
}),
)
}

10
src/database/key_value/rooms/timeline.rs
View file

@ -9,6 +9,8 @@ use crate::{database::KeyValueDatabase, service, services, utils, Error, PduEven
use service::rooms::timeline::PduCount;
use super::get_userroom_id_bytes;
impl service::rooms::timeline::Data for KeyValueDatabase {
fn last_timeline_count(&self, sender_user: &UserId, room_id: &RoomId) -> Result<PduCount> {
match self
@ -286,15 +288,11 @@ impl service::rooms::timeline::Data for KeyValueDatabase {
let mut notifies_batch = Vec::new();
let mut highlights_batch = Vec::new();
for user in notifies {
let mut userroom_id = user.as_bytes().to_vec();
userroom_id.push(0xff);
userroom_id.extend_from_slice(room_id.as_bytes());
let userroom_id = get_userroom_id_bytes(&user, room_id);
notifies_batch.push(userroom_id);
}
for user in highlights {
let mut userroom_id = user.as_bytes().to_vec();
userroom_id.push(0xff);
userroom_id.extend_from_slice(room_id.as_bytes());
let userroom_id = get_userroom_id_bytes(&user, room_id);
highlights_batch.push(userroom_id);
}

17
src/database/key_value/rooms/user.rs
View file

@ -2,14 +2,11 @@ use ruma::{OwnedRoomId, OwnedUserId, RoomId, UserId};
use crate::{database::KeyValueDatabase, service, services, utils, Error, Result};
use super::{get_room_and_user_byte_ids, get_userroom_id_bytes};
impl service::rooms::user::Data for KeyValueDatabase {
fn reset_notification_counts(&self, user_id: &UserId, room_id: &RoomId) -> Result<()> {
let mut userroom_id = user_id.as_bytes().to_vec();
userroom_id.push(0xff);
userroom_id.extend_from_slice(room_id.as_bytes());
let mut roomuser_id = room_id.as_bytes().to_vec();
roomuser_id.push(0xff);
roomuser_id.extend_from_slice(user_id.as_bytes());
let (roomuser_id, userroom_id) = get_room_and_user_byte_ids(room_id, user_id);
self.userroomid_notificationcount
.insert(&userroom_id, &0_u64.to_be_bytes())?;
@ -25,9 +22,7 @@ impl service::rooms::user::Data for KeyValueDatabase {
}
fn notification_count(&self, user_id: &UserId, room_id: &RoomId) -> Result<u64> {
let mut userroom_id = user_id.as_bytes().to_vec();
userroom_id.push(0xff);
userroom_id.extend_from_slice(room_id.as_bytes());
let userroom_id = get_userroom_id_bytes(user_id, room_id);
self.userroomid_notificationcount
.get(&userroom_id)?
@ -39,9 +34,7 @@ impl service::rooms::user::Data for KeyValueDatabase {
}
fn highlight_count(&self, user_id: &UserId, room_id: &RoomId) -> Result<u64> {
let mut userroom_id = user_id.as_bytes().to_vec();
userroom_id.push(0xff);
userroom_id.extend_from_slice(room_id.as_bytes());
let userroom_id = get_userroom_id_bytes(user_id, room_id);
self.userroomid_highlightcount
.get(&userroom_id)?

2
src/database/mod.rs
View file

@ -97,7 +97,7 @@ pub struct KeyValueDatabase {
pub(super) roomid_joinedcount: Arc<dyn KvTree>,
pub(super) roomid_invitedcount: Arc<dyn KvTree>,
pub(super) roomuseroncejoinedids: Arc<dyn KvTree>,
pub(super) userroomid_invitestate: Arc<dyn KvTree>, // InviteState = Vec<Raw<Pdu>>
pub(super) userroomid_invitestate: Arc<dyn KvTree>, // InviteState = Vec<Raw<AnyStrippedStateEvent>>
pub(super) roomuserid_invitecount: Arc<dyn KvTree>, // InviteCount = Count
pub(super) userroomid_leftstate: Arc<dyn KvTree>,
pub(super) roomuserid_leftcount: Arc<dyn KvTree>,

69
src/service/rooms/state/mod.rs
View file

@ -10,6 +10,7 @@ use ruma::{
events::{
room::{create::RoomCreateEventContent, member::MembershipState},
AnyStrippedStateEvent, StateEventType, TimelineEventType,
RECOMMENDED_STRIPPED_STATE_EVENT_TYPES,
},
serde::Raw,
state_res::{self, StateMap},
@ -258,58 +259,22 @@ impl Service {
}
}
#[tracing::instrument(skip(self, invite_event))]
pub fn calculate_invite_state(
&self,
invite_event: &PduEvent,
) -> Result<Vec<Raw<AnyStrippedStateEvent>>> {
let mut state = Vec::new();
// Add recommended events
if let Some(e) = services().rooms.state_accessor.room_state_get(
&invite_event.room_id,
&StateEventType::RoomCreate,
"",
)? {
state.push(e.to_stripped_state_event());
}
if let Some(e) = services().rooms.state_accessor.room_state_get(
&invite_event.room_id,
&StateEventType::RoomJoinRules,
"",
)? {
state.push(e.to_stripped_state_event());
}
if let Some(e) = services().rooms.state_accessor.room_state_get(
&invite_event.room_id,
&StateEventType::RoomCanonicalAlias,
"",
)? {
state.push(e.to_stripped_state_event());
}
if let Some(e) = services().rooms.state_accessor.room_state_get(
&invite_event.room_id,
&StateEventType::RoomAvatar,
"",
)? {
state.push(e.to_stripped_state_event());
}
if let Some(e) = services().rooms.state_accessor.room_state_get(
&invite_event.room_id,
&StateEventType::RoomName,
"",
)? {
state.push(e.to_stripped_state_event());
}
if let Some(e) = services().rooms.state_accessor.room_state_get(
&invite_event.room_id,
&StateEventType::RoomMember,
invite_event.sender.as_str(),
)? {
state.push(e.to_stripped_state_event());
}
state.push(invite_event.to_stripped_state_event());
Ok(state)
#[tracing::instrument(skip(self, room_id))]
/// Gets all the [recommended stripped state events] from the given room
///
/// [recommended stripped state events]: https://spec.matrix.org/v1.13/client-server-api/#stripped-state
pub fn stripped_state(&self, room_id: &RoomId) -> Result<Vec<Raw<AnyStrippedStateEvent>>> {
RECOMMENDED_STRIPPED_STATE_EVENT_TYPES
.iter()
.filter_map(|state_event_type| {
services()
.rooms
.state_accessor
.room_state_get(room_id, state_event_type, "")
.transpose()
})
.map(|e| e.map(|e| e.to_stripped_state_event()))
.collect::<Result<Vec<_>>>()
}
/// Set the state hash to a new version, but does not update state_cache.

7
src/service/rooms/state_cache/mod.rs
View file

@ -40,6 +40,13 @@ impl Service {
// TODO: displayname, avatar url
}
// We don't need to store stripped state on behalf of remote users, since these events are only used on `/sync`
let last_state = if user_id.server_name() == services().globals.server_name() {
last_state
} else {
None
};
match &membership {
MembershipState::Join => {
// Check if the user never joined this room

8
src/service/rooms/timeline/mod.rs
View file

@ -450,9 +450,11 @@ impl Service {
let content = serde_json::from_str::<ExtractMembership>(pdu.content.get())
.map_err(|_| Error::bad_database("Invalid content in pdu."))?;
let invite_state = match content.membership {
let stripped_state = match content.membership {
MembershipState::Invite => {
let state = services().rooms.state.calculate_invite_state(pdu)?;
let mut state = services().rooms.state.stripped_state(&pdu.room_id)?;
// So that clients can get info about who invitied them, the reason, when, etc.
state.push(pdu.to_stripped_state_event());
Some(state)
}
_ => None,
@ -465,7 +467,7 @@ impl Service {
&target_user_id,
content.membership,
&pdu.sender,
invite_state,
stripped_state,
true,
)?;
}