serv/conduit
1
0
Fork 0
mirror of https://gitlab.com/famedly/conduit.git synced 2024-12-27 04:53:50 +03:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat: Keep track of avatar urls, displaynames, and blurhashes of remote users for the room directory

Browse source
This commit is contained in:
Nyaaori 2022-12-21 17:28:21 +01:00
parent c86313d4fa
commit 3ac3bdbac0
No known key found for this signature in database
GPG key ID: E7819C3ED4D1F82E
14 changed files with 908 additions and 680 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -198,18 +198,22 @@ pub async fn kick_user_route(
); );
let state_lock = mutex_state.lock().await; let state_lock = mutex_state.lock().await;
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomMember, .timeline
content: to_raw_value(&event).expect("event is valid, we just created it"), .build_and_append_pdu(
unsigned: None, PduBuilder {
state_key: Some(body.user_id.to_string()), event_type: RoomEventType::RoomMember,
redacts: None, content: to_raw_value(&event).expect("event is valid, we just created it"),
}, unsigned: None,
sender_user, state_key: Some(body.user_id.to_string()),
&body.room_id, redacts: None,
&state_lock, },
)?; sender_user,
&body.room_id,
&state_lock,
)
.await?;
drop(state_lock); drop(state_lock);
@ -264,18 +268,22 @@ pub async fn ban_user_route(body: Ruma<ban_user::v3::Request>) -> Result<ban_use
); );
let state_lock = mutex_state.lock().await; let state_lock = mutex_state.lock().await;
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomMember, .timeline
content: to_raw_value(&event).expect("event is valid, we just created it"), .build_and_append_pdu(
unsigned: None, PduBuilder {
state_key: Some(body.user_id.to_string()), event_type: RoomEventType::RoomMember,
redacts: None, content: to_raw_value(&event).expect("event is valid, we just created it"),
}, unsigned: None,
sender_user, state_key: Some(body.user_id.to_string()),
&body.room_id, redacts: None,
&state_lock, },
)?; sender_user,
&body.room_id,
&state_lock,
)
.await?;
drop(state_lock); drop(state_lock);
@ -321,18 +329,22 @@ pub async fn unban_user_route(
); );
let state_lock = mutex_state.lock().await; let state_lock = mutex_state.lock().await;
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomMember, .timeline
content: to_raw_value(&event).expect("event is valid, we just created it"), .build_and_append_pdu(
unsigned: None, PduBuilder {
state_key: Some(body.user_id.to_string()), event_type: RoomEventType::RoomMember,
redacts: None, content: to_raw_value(&event).expect("event is valid, we just created it"),
}, unsigned: None,
sender_user, state_key: Some(body.user_id.to_string()),
&body.room_id, redacts: None,
&state_lock, },
)?; sender_user,
&body.room_id,
&state_lock,
)
.await?;
drop(state_lock); drop(state_lock);
@ -738,12 +750,16 @@ async fn join_room_by_id_helper(
// pdu without it's state. This is okay because append_pdu can't fail. // pdu without it's state. This is okay because append_pdu can't fail.
let statehash_after_join = services().rooms.state.append_to_state(&parsed_join_pdu)?; let statehash_after_join = services().rooms.state.append_to_state(&parsed_join_pdu)?;
services().rooms.timeline.append_pdu( services()
&parsed_join_pdu, .rooms
join_event, .timeline
vec![(*parsed_join_pdu.event_id).to_owned()], .append_pdu(
&state_lock, &parsed_join_pdu,
)?; join_event,
vec![(*parsed_join_pdu.event_id).to_owned()],
&state_lock,
)
.await?;
// We set the room state after inserting the pdu, so that we never have a moment in time // We set the room state after inserting the pdu, so that we never have a moment in time
// where events in the current room state do not exist // where events in the current room state do not exist
@ -853,18 +869,23 @@ async fn join_room_by_id_helper(
}; };
// Try normal join first // Try normal join first
let error = match services().rooms.timeline.build_and_append_pdu( let error = match services()
PduBuilder { .rooms
event_type: RoomEventType::RoomMember, .timeline
content: to_raw_value(&event).expect("event is valid, we just created it"), .build_and_append_pdu(
unsigned: None, PduBuilder {
state_key: Some(sender_user.to_string()), event_type: RoomEventType::RoomMember,
redacts: None, content: to_raw_value(&event).expect("event is valid, we just created it"),
}, unsigned: None,
sender_user, state_key: Some(sender_user.to_string()),
room_id, redacts: None,
&state_lock, },
) { sender_user,
room_id,
&state_lock,
)
.await
{
Ok(_event_id) => return Ok(join_room_by_id::v3::Response::new(room_id.to_owned())), Ok(_event_id) => return Ok(join_room_by_id::v3::Response::new(room_id.to_owned())),
Err(e) => e, Err(e) => e,
}; };
@ -1259,28 +1280,32 @@ pub(crate) async fn invite_helper<'a>(
); );
let state_lock = mutex_state.lock().await; let state_lock = mutex_state.lock().await;
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomMember, .timeline
content: to_raw_value(&RoomMemberEventContent { .build_and_append_pdu(
membership: MembershipState::Invite, PduBuilder {
displayname: services().users.displayname(user_id)?, event_type: RoomEventType::RoomMember,
avatar_url: services().users.avatar_url(user_id)?, content: to_raw_value(&RoomMemberEventContent {
is_direct: Some(is_direct), membership: MembershipState::Invite,
third_party_invite: None, displayname: services().users.displayname(user_id)?,
blurhash: services().users.blurhash(user_id)?, avatar_url: services().users.avatar_url(user_id)?,
reason: None, is_direct: Some(is_direct),
join_authorized_via_users_server: None, third_party_invite: None,
}) blurhash: services().users.blurhash(user_id)?,
.expect("event is valid, we just created it"), reason: None,
unsigned: None, join_authorized_via_users_server: None,
state_key: Some(user_id.to_string()), })
redacts: None, .expect("event is valid, we just created it"),
}, unsigned: None,
sender_user, state_key: Some(user_id.to_string()),
room_id, redacts: None,
&state_lock, },
)?; sender_user,
room_id,
&state_lock,
)
.await?;
drop(state_lock); drop(state_lock);
@ -1334,14 +1359,18 @@ pub async fn leave_room(user_id: &UserId, room_id: &RoomId) -> Result<()> {
)?; )?;
// We always drop the invite, we can't rely on other servers // We always drop the invite, we can't rely on other servers
services().rooms.state_cache.update_membership( services()
room_id, .rooms
user_id, .state_cache
MembershipState::Leave, .update_membership(
user_id, room_id,
last_state, user_id,
true, RoomMemberEventContent::new(MembershipState::Leave),
)?; user_id,
last_state,
true,
)
.await?;
} else { } else {
let mutex_state = Arc::clone( let mutex_state = Arc::clone(
services() services()
@ -1365,14 +1394,18 @@ pub async fn leave_room(user_id: &UserId, room_id: &RoomId) -> Result<()> {
None => { None => {
error!("Trying to leave a room you are not a member of."); error!("Trying to leave a room you are not a member of.");
services().rooms.state_cache.update_membership( services()
room_id, .rooms
user_id, .state_cache
MembershipState::Leave, .update_membership(
user_id, room_id,
None, user_id,
true, RoomMemberEventContent::new(MembershipState::Leave),
)?; user_id,
None,
true,
)
.await?;
return Ok(()); return Ok(());
} }
Some(e) => e, Some(e) => e,
@ -1383,18 +1416,22 @@ pub async fn leave_room(user_id: &UserId, room_id: &RoomId) -> Result<()> {
event.membership = MembershipState::Leave; event.membership = MembershipState::Leave;
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomMember, .timeline
content: to_raw_value(&event).expect("event is valid, we just created it"), .build_and_append_pdu(
unsigned: None, PduBuilder {
state_key: Some(user_id.to_string()), event_type: RoomEventType::RoomMember,
redacts: None, content: to_raw_value(&event).expect("event is valid, we just created it"),
}, unsigned: None,
user_id, state_key: Some(user_id.to_string()),
room_id, redacts: None,
&state_lock, },
)?; user_id,
room_id,
&state_lock,
)
.await?;
} }
Ok(()) Ok(())

30
src/api/client_server/message.rs
View file

@ -70,19 +70,23 @@ pub async fn send_message_event_route(
let mut unsigned = BTreeMap::new(); let mut unsigned = BTreeMap::new();
unsigned.insert("transaction_id".to_owned(), body.txn_id.to_string().into()); unsigned.insert("transaction_id".to_owned(), body.txn_id.to_string().into());
let event_id = services().rooms.timeline.build_and_append_pdu( let event_id = services()
PduBuilder { .rooms
event_type: body.event_type.to_string().into(), .timeline
content: serde_json::from_str(body.body.body.json().get()) .build_and_append_pdu(
.map_err(|_| Error::BadRequest(ErrorKind::BadJson, "Invalid JSON body."))?, PduBuilder {
unsigned: Some(unsigned), event_type: body.event_type.to_string().into(),
state_key: None, content: serde_json::from_str(body.body.body.json().get())
redacts: None, .map_err(|_| Error::BadRequest(ErrorKind::BadJson, "Invalid JSON body."))?,
}, unsigned: Some(unsigned),
sender_user, state_key: None,
&body.room_id, redacts: None,
&state_lock, },
)?; sender_user,
&body.room_id,
&state_lock,
)
.await?;
services().transaction_ids.add_txnid( services().transaction_ids.add_txnid(
sender_user, sender_user,

80
src/api/client_server/profile.rs
View file

@ -83,12 +83,11 @@ pub async fn set_displayname_route(
); );
let state_lock = mutex_state.lock().await; let state_lock = mutex_state.lock().await;
let _ = services().rooms.timeline.build_and_append_pdu( let _ = services()
pdu_builder, .rooms
sender_user, .timeline
&room_id, .build_and_append_pdu(pdu_builder, sender_user, &room_id, &state_lock)
&state_lock, .await;
);
// Presence update // Presence update
services().rooms.edus.presence.update_presence( services().rooms.edus.presence.update_presence(
@ -115,15 +114,17 @@ pub async fn set_displayname_route(
Ok(set_display_name::v3::Response {}) Ok(set_display_name::v3::Response {})
} }
/// # `GET /_matrix/client/r0/profile/{userId}/displayname` /// # `GET /_matrix/client/v3/profile/{userId}/displayname`
/// ///
/// Returns the displayname of the user. /// Returns the displayname of the user.
/// ///
/// - If user is on another server: Fetches displayname over federation /// - If user is on another server and we do not have a copy, fetch over federation
pub async fn get_displayname_route( pub async fn get_displayname_route(
body: Ruma<get_display_name::v3::Request>, body: Ruma<get_display_name::v3::Request>,
) -> Result<get_display_name::v3::Response> { ) -> Result<get_display_name::v3::Response> {
if body.user_id.server_name() != services().globals.server_name() { if (services().users.exists(&body.user_id)?)
&& (body.user_id.server_name() != services().globals.server_name())
{
let response = services() let response = services()
.sending .sending
.send_federation_request( .send_federation_request(
@ -135,6 +136,18 @@ pub async fn get_displayname_route(
) )
.await?; .await?;
// Create and update our local copy of the user
let _ = services().users.create(&body.user_id, None);
let _ = services()
.users
.set_displayname(&body.user_id, response.displayname.clone());
let _ = services()
.users
.set_avatar_url(&body.user_id, response.avatar_url);
let _ = services()
.users
.set_blurhash(&body.user_id, response.blurhash);
return Ok(get_display_name::v3::Response { return Ok(get_display_name::v3::Response {
displayname: response.displayname, displayname: response.displayname,
}); });
@ -218,12 +231,11 @@ pub async fn set_avatar_url_route(
); );
let state_lock = mutex_state.lock().await; let state_lock = mutex_state.lock().await;
let _ = services().rooms.timeline.build_and_append_pdu( let _ = services()
pdu_builder, .rooms
sender_user, .timeline
&room_id, .build_and_append_pdu(pdu_builder, sender_user, &room_id, &state_lock)
&state_lock, .await;
);
// Presence update // Presence update
services().rooms.edus.presence.update_presence( services().rooms.edus.presence.update_presence(
@ -250,15 +262,17 @@ pub async fn set_avatar_url_route(
Ok(set_avatar_url::v3::Response {}) Ok(set_avatar_url::v3::Response {})
} }
/// # `GET /_matrix/client/r0/profile/{userId}/avatar_url` /// # `GET /_matrix/client/v3/profile/{userId}/avatar_url`
/// ///
/// Returns the avatar_url and blurhash of the user. /// Returns the avatar_url and blurhash of the user.
/// ///
/// - If user is on another server: Fetches avatar_url and blurhash over federation /// - If user is on another server and we do not have a copy, fetch over federation
pub async fn get_avatar_url_route( pub async fn get_avatar_url_route(
body: Ruma<get_avatar_url::v3::Request>, body: Ruma<get_avatar_url::v3::Request>,
) -> Result<get_avatar_url::v3::Response> { ) -> Result<get_avatar_url::v3::Response> {
if body.user_id.server_name() != services().globals.server_name() { if (services().users.exists(&body.user_id)?)
&& (body.user_id.server_name() != services().globals.server_name())
{
let response = services() let response = services()
.sending .sending
.send_federation_request( .send_federation_request(
@ -270,6 +284,18 @@ pub async fn get_avatar_url_route(
) )
.await?; .await?;
// Create and update our local copy of the user
let _ = services().users.create(&body.user_id, None);
let _ = services()
.users
.set_displayname(&body.user_id, response.displayname);
let _ = services()
.users
.set_avatar_url(&body.user_id, response.avatar_url.clone());
let _ = services()
.users
.set_blurhash(&body.user_id, response.blurhash.clone());
return Ok(get_avatar_url::v3::Response { return Ok(get_avatar_url::v3::Response {
avatar_url: response.avatar_url, avatar_url: response.avatar_url,
blurhash: response.blurhash, blurhash: response.blurhash,
@ -286,11 +312,13 @@ pub async fn get_avatar_url_route(
/// ///
/// Returns the displayname, avatar_url and blurhash of the user. /// Returns the displayname, avatar_url and blurhash of the user.
/// ///
/// - If user is on another server: Fetches profile over federation /// - If user is on another server and we do not have a copy, fetch over federation
pub async fn get_profile_route( pub async fn get_profile_route(
body: Ruma<get_profile::v3::Request>, body: Ruma<get_profile::v3::Request>,
) -> Result<get_profile::v3::Response> { ) -> Result<get_profile::v3::Response> {
if body.user_id.server_name() != services().globals.server_name() { if (services().users.exists(&body.user_id)?)
&& (body.user_id.server_name() != services().globals.server_name())
{
let response = services() let response = services()
.sending .sending
.send_federation_request( .send_federation_request(
@ -302,6 +330,18 @@ pub async fn get_profile_route(
) )
.await?; .await?;
// Create and update our local copy of the user
let _ = services().users.create(&body.user_id, None);
let _ = services()
.users
.set_displayname(&body.user_id, response.displayname.clone());
let _ = services()
.users
.set_avatar_url(&body.user_id, response.avatar_url.clone());
let _ = services()
.users
.set_blurhash(&body.user_id, response.blurhash.clone());
return Ok(get_profile::v3::Response { return Ok(get_profile::v3::Response {
displayname: response.displayname, displayname: response.displayname,
avatar_url: response.avatar_url, avatar_url: response.avatar_url,

34
src/api/client_server/redact.rs
View file

@ -30,21 +30,25 @@ pub async fn redact_event_route(
); );
let state_lock = mutex_state.lock().await; let state_lock = mutex_state.lock().await;
let event_id = services().rooms.timeline.build_and_append_pdu( let event_id = services()
PduBuilder { .rooms
event_type: RoomEventType::RoomRedaction, .timeline
content: to_raw_value(&RoomRedactionEventContent { .build_and_append_pdu(
reason: body.reason.clone(), PduBuilder {
}) event_type: RoomEventType::RoomRedaction,
.expect("event is valid, we just created it"), content: to_raw_value(&RoomRedactionEventContent {
unsigned: None, reason: body.reason.clone(),
state_key: None, })
redacts: Some(body.event_id.into()), .expect("event is valid, we just created it"),
}, unsigned: None,
sender_user, state_key: None,
&body.room_id, redacts: Some(body.event_id.into()),
&state_lock, },
)?; sender_user,
&body.room_id,
&state_lock,
)
.await?;
drop(state_lock); drop(state_lock);

487
src/api/client_server/room.rs
View file

@ -173,42 +173,50 @@ pub async fn create_room_route(
} }
// 1. The room create event // 1. The room create event
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomCreate, .timeline
content: to_raw_value(&content).expect("event is valid, we just created it"), .build_and_append_pdu(
unsigned: None, PduBuilder {
state_key: Some("".to_owned()), event_type: RoomEventType::RoomCreate,
redacts: None, content: to_raw_value(&content).expect("event is valid, we just created it"),
}, unsigned: None,
sender_user, state_key: Some("".to_owned()),
&room_id, redacts: None,
&state_lock, },
)?; sender_user,
&room_id,
&state_lock,
)
.await?;
// 2. Let the room creator join // 2. Let the room creator join
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomMember, .timeline
content: to_raw_value(&RoomMemberEventContent { .build_and_append_pdu(
membership: MembershipState::Join, PduBuilder {
displayname: services().users.displayname(sender_user)?, event_type: RoomEventType::RoomMember,
avatar_url: services().users.avatar_url(sender_user)?, content: to_raw_value(&RoomMemberEventContent {
is_direct: Some(body.is_direct), membership: MembershipState::Join,
third_party_invite: None, displayname: services().users.displayname(sender_user)?,
blurhash: services().users.blurhash(sender_user)?, avatar_url: services().users.avatar_url(sender_user)?,
reason: None, is_direct: Some(body.is_direct),
join_authorized_via_users_server: None, third_party_invite: None,
}) blurhash: services().users.blurhash(sender_user)?,
.expect("event is valid, we just created it"), reason: None,
unsigned: None, join_authorized_via_users_server: None,
state_key: Some(sender_user.to_string()), })
redacts: None, .expect("event is valid, we just created it"),
}, unsigned: None,
sender_user, state_key: Some(sender_user.to_string()),
&room_id, redacts: None,
&state_lock, },
)?; sender_user,
&room_id,
&state_lock,
)
.await?;
// 3. Power levels // 3. Power levels
@ -245,30 +253,14 @@ pub async fn create_room_route(
} }
} }
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomPowerLevels, .timeline
content: to_raw_value(&power_levels_content) .build_and_append_pdu(
.expect("to_raw_value always works on serde_json::Value"),
unsigned: None,
state_key: Some("".to_owned()),
redacts: None,
},
sender_user,
&room_id,
&state_lock,
)?;
// 4. Canonical room alias
if let Some(room_alias_id) = &alias {
services().rooms.timeline.build_and_append_pdu(
PduBuilder { PduBuilder {
event_type: RoomEventType::RoomCanonicalAlias, event_type: RoomEventType::RoomPowerLevels,
content: to_raw_value(&RoomCanonicalAliasEventContent { content: to_raw_value(&power_levels_content)
alias: Some(room_alias_id.to_owned()), .expect("to_raw_value always works on serde_json::Value"),
alt_aliases: vec![],
})
.expect("We checked that alias earlier, it must be fine"),
unsigned: None, unsigned: None,
state_key: Some("".to_owned()), state_key: Some("".to_owned()),
redacts: None, redacts: None,
@ -276,64 +268,100 @@ pub async fn create_room_route(
sender_user, sender_user,
&room_id, &room_id,
&state_lock, &state_lock,
)?; )
.await?;
// 4. Canonical room alias
if let Some(room_alias_id) = &alias {
services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder {
event_type: RoomEventType::RoomCanonicalAlias,
content: to_raw_value(&RoomCanonicalAliasEventContent {
alias: Some(room_alias_id.to_owned()),
alt_aliases: vec![],
})
.expect("We checked that alias earlier, it must be fine"),
unsigned: None,
state_key: Some("".to_owned()),
redacts: None,
},
sender_user,
&room_id,
&state_lock,
)
.await?;
} }
// 5. Events set by preset // 5. Events set by preset
// 5.1 Join Rules // 5.1 Join Rules
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomJoinRules, .timeline
content: to_raw_value(&RoomJoinRulesEventContent::new(match preset { .build_and_append_pdu(
RoomPreset::PublicChat => JoinRule::Public, PduBuilder {
// according to spec "invite" is the default event_type: RoomEventType::RoomJoinRules,
_ => JoinRule::Invite, content: to_raw_value(&RoomJoinRulesEventContent::new(match preset {
})) RoomPreset::PublicChat => JoinRule::Public,
.expect("event is valid, we just created it"), // according to spec "invite" is the default
unsigned: None, _ => JoinRule::Invite,
state_key: Some("".to_owned()), }))
redacts: None, .expect("event is valid, we just created it"),
}, unsigned: None,
sender_user, state_key: Some("".to_owned()),
&room_id, redacts: None,
&state_lock, },
)?; sender_user,
&room_id,
&state_lock,
)
.await?;
// 5.2 History Visibility // 5.2 History Visibility
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomHistoryVisibility, .timeline
content: to_raw_value(&RoomHistoryVisibilityEventContent::new( .build_and_append_pdu(
HistoryVisibility::Shared, PduBuilder {
)) event_type: RoomEventType::RoomHistoryVisibility,
.expect("event is valid, we just created it"), content: to_raw_value(&RoomHistoryVisibilityEventContent::new(
unsigned: None, HistoryVisibility::Shared,
state_key: Some("".to_owned()), ))
redacts: None, .expect("event is valid, we just created it"),
}, unsigned: None,
sender_user, state_key: Some("".to_owned()),
&room_id, redacts: None,
&state_lock, },
)?; sender_user,
&room_id,
&state_lock,
)
.await?;
// 5.3 Guest Access // 5.3 Guest Access
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomGuestAccess, .timeline
content: to_raw_value(&RoomGuestAccessEventContent::new(match preset { .build_and_append_pdu(
RoomPreset::PublicChat => GuestAccess::Forbidden, PduBuilder {
_ => GuestAccess::CanJoin, event_type: RoomEventType::RoomGuestAccess,
})) content: to_raw_value(&RoomGuestAccessEventContent::new(match preset {
.expect("event is valid, we just created it"), RoomPreset::PublicChat => GuestAccess::Forbidden,
unsigned: None, _ => GuestAccess::CanJoin,
state_key: Some("".to_owned()), }))
redacts: None, .expect("event is valid, we just created it"),
}, unsigned: None,
sender_user, state_key: Some("".to_owned()),
&room_id, redacts: None,
&state_lock, },
)?; sender_user,
&room_id,
&state_lock,
)
.await?;
// 6. Events listed in initial_state // 6. Events listed in initial_state
for event in &body.initial_state { for event in &body.initial_state {
@ -352,47 +380,54 @@ pub async fn create_room_route(
continue; continue;
} }
services().rooms.timeline.build_and_append_pdu( services()
pdu_builder, .rooms
sender_user, .timeline
&room_id, .build_and_append_pdu(pdu_builder, sender_user, &room_id, &state_lock)
&state_lock, .await?;
)?;
} }
// 7. Events implied by name and topic // 7. Events implied by name and topic
if let Some(name) = &body.name { if let Some(name) = &body.name {
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomName, .timeline
content: to_raw_value(&RoomNameEventContent::new(Some(name.clone()))) .build_and_append_pdu(
.expect("event is valid, we just created it"), PduBuilder {
unsigned: None, event_type: RoomEventType::RoomName,
state_key: Some("".to_owned()), content: to_raw_value(&RoomNameEventContent::new(Some(name.clone())))
redacts: None, .expect("event is valid, we just created it"),
}, unsigned: None,
sender_user, state_key: Some("".to_owned()),
&room_id, redacts: None,
&state_lock, },
)?; sender_user,
&room_id,
&state_lock,
)
.await?;
} }
if let Some(topic) = &body.topic { if let Some(topic) = &body.topic {
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomTopic, .timeline
content: to_raw_value(&RoomTopicEventContent { .build_and_append_pdu(
topic: topic.clone(), PduBuilder {
}) event_type: RoomEventType::RoomTopic,
.expect("event is valid, we just created it"), content: to_raw_value(&RoomTopicEventContent {
unsigned: None, topic: topic.clone(),
state_key: Some("".to_owned()), })
redacts: None, .expect("event is valid, we just created it"),
}, unsigned: None,
sender_user, state_key: Some("".to_owned()),
&room_id, redacts: None,
&state_lock, },
)?; sender_user,
&room_id,
&state_lock,
)
.await?;
} }
// 8. Events implied by invite (and TODO: invite_3pid) // 8. Events implied by invite (and TODO: invite_3pid)
@ -523,22 +558,26 @@ pub async fn upgrade_room_route(
// Send a m.room.tombstone event to the old room to indicate that it is not intended to be used any further // Send a m.room.tombstone event to the old room to indicate that it is not intended to be used any further
// Fail if the sender does not have the required permissions // Fail if the sender does not have the required permissions
let tombstone_event_id = services().rooms.timeline.build_and_append_pdu( let tombstone_event_id = services()
PduBuilder { .rooms
event_type: RoomEventType::RoomTombstone, .timeline
content: to_raw_value(&RoomTombstoneEventContent { .build_and_append_pdu(
body: "This room has been replaced".to_owned(), PduBuilder {
replacement_room: replacement_room.clone(), event_type: RoomEventType::RoomTombstone,
}) content: to_raw_value(&RoomTombstoneEventContent {
.expect("event is valid, we just created it"), body: "This room has been replaced".to_owned(),
unsigned: None, replacement_room: replacement_room.clone(),
state_key: Some("".to_owned()), })
redacts: None, .expect("event is valid, we just created it"),
}, unsigned: None,
sender_user, state_key: Some("".to_owned()),
&body.room_id, redacts: None,
&state_lock, },
)?; sender_user,
&body.room_id,
&state_lock,
)
.await?;
// Change lock to replacement room // Change lock to replacement room
drop(state_lock); drop(state_lock);
@ -605,43 +644,51 @@ pub async fn upgrade_room_route(
)); ));
} }
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomCreate, .timeline
content: to_raw_value(&create_event_content) .build_and_append_pdu(
.expect("event is valid, we just created it"), PduBuilder {
unsigned: None, event_type: RoomEventType::RoomCreate,
state_key: Some("".to_owned()), content: to_raw_value(&create_event_content)
redacts: None, .expect("event is valid, we just created it"),
}, unsigned: None,
sender_user, state_key: Some("".to_owned()),
&replacement_room, redacts: None,
&state_lock, },
)?; sender_user,
&replacement_room,
&state_lock,
)
.await?;
// Join the new room // Join the new room
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomMember, .timeline
content: to_raw_value(&RoomMemberEventContent { .build_and_append_pdu(
membership: MembershipState::Join, PduBuilder {
displayname: services().users.displayname(sender_user)?, event_type: RoomEventType::RoomMember,
avatar_url: services().users.avatar_url(sender_user)?, content: to_raw_value(&RoomMemberEventContent {
is_direct: None, membership: MembershipState::Join,
third_party_invite: None, displayname: services().users.displayname(sender_user)?,
blurhash: services().users.blurhash(sender_user)?, avatar_url: services().users.avatar_url(sender_user)?,
reason: None, is_direct: None,
join_authorized_via_users_server: None, third_party_invite: None,
}) blurhash: services().users.blurhash(sender_user)?,
.expect("event is valid, we just created it"), reason: None,
unsigned: None, join_authorized_via_users_server: None,
state_key: Some(sender_user.to_string()), })
redacts: None, .expect("event is valid, we just created it"),
}, unsigned: None,
sender_user, state_key: Some(sender_user.to_string()),
&replacement_room, redacts: None,
&state_lock, },
)?; sender_user,
&replacement_room,
&state_lock,
)
.await?;
// Recommended transferable state events list from the specs // Recommended transferable state events list from the specs
let transferable_state_events = vec![ let transferable_state_events = vec![
@ -668,18 +715,22 @@ pub async fn upgrade_room_route(
None => continue, // Skipping missing events. None => continue, // Skipping missing events.
}; };
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: event_type.to_string().into(), .timeline
content: event_content, .build_and_append_pdu(
unsigned: None, PduBuilder {
state_key: Some("".to_owned()), event_type: event_type.to_string().into(),
redacts: None, content: event_content,
}, unsigned: None,
sender_user, state_key: Some("".to_owned()),
&replacement_room, redacts: None,
&state_lock, },
)?; sender_user,
&replacement_room,
&state_lock,
)
.await?;
} }
// Moves any local aliases to the new room // Moves any local aliases to the new room
@ -713,19 +764,23 @@ pub async fn upgrade_room_route(
power_levels_event_content.invite = new_level; power_levels_event_content.invite = new_level;
// Modify the power levels in the old room to prevent sending of events and inviting new users // Modify the power levels in the old room to prevent sending of events and inviting new users
let _ = services().rooms.timeline.build_and_append_pdu( let _ = services()
PduBuilder { .rooms
event_type: RoomEventType::RoomPowerLevels, .timeline
content: to_raw_value(&power_levels_event_content) .build_and_append_pdu(
.expect("event is valid, we just created it"), PduBuilder {
unsigned: None, event_type: RoomEventType::RoomPowerLevels,
state_key: Some("".to_owned()), content: to_raw_value(&power_levels_event_content)
redacts: None, .expect("event is valid, we just created it"),
}, unsigned: None,
sender_user, state_key: Some("".to_owned()),
&body.room_id, redacts: None,
&state_lock, },
)?; sender_user,
&body.room_id,
&state_lock,
)
.await?;
drop(state_lock); drop(state_lock);

28
src/api/client_server/state.rs
View file

@ -287,18 +287,22 @@ async fn send_state_event_for_key_helper(
); );
let state_lock = mutex_state.lock().await; let state_lock = mutex_state.lock().await;
let event_id = services().rooms.timeline.build_and_append_pdu( let event_id = services()
PduBuilder { .rooms
event_type: event_type.to_string().into(), .timeline
content: serde_json::from_str(json.json().get()).expect("content is valid json"), .build_and_append_pdu(
unsigned: None, PduBuilder {
state_key: Some(state_key), event_type: event_type.to_string().into(),
redacts: None, content: serde_json::from_str(json.json().get()).expect("content is valid json"),
}, unsigned: None,
sender_user, state_key: Some(state_key),
room_id, redacts: None,
&state_lock, },
)?; sender_user,
room_id,
&state_lock,
)
.await?;
Ok(event_id) Ok(event_id)
} }

6
src/api/client_server/sync.rs
View file

@ -231,7 +231,7 @@ async fn sync_helper(
.entry(room_id.clone()) .entry(room_id.clone())
.or_default(), .or_default(),
); );
let insert_lock = mutex_insert.lock().unwrap(); let insert_lock = mutex_insert.lock().await;
drop(insert_lock); drop(insert_lock);
} }
@ -847,7 +847,7 @@ async fn sync_helper(
.entry(room_id.clone()) .entry(room_id.clone())
.or_default(), .or_default(),
); );
let insert_lock = mutex_insert.lock().unwrap(); let insert_lock = mutex_insert.lock().await;
drop(insert_lock); drop(insert_lock);
} }
@ -979,7 +979,7 @@ async fn sync_helper(
.entry(room_id.clone()) .entry(room_id.clone())
.or_default(), .or_default(),
); );
let insert_lock = mutex_insert.lock().unwrap(); let insert_lock = mutex_insert.lock().await;
drop(insert_lock); drop(insert_lock);
} }

20
src/api/server_server.rs
View file

@ -1615,14 +1615,18 @@ pub async fn create_invite_route(
.state_cache .state_cache
.server_in_room(services().globals.server_name(), &body.room_id)? .server_in_room(services().globals.server_name(), &body.room_id)?
{ {
services().rooms.state_cache.update_membership( services()
&body.room_id, .rooms
&invited_user, .state_cache
MembershipState::Invite, .update_membership(
&sender, &body.room_id,
Some(invite_state), &invited_user,
true, RoomMemberEventContent::new(MembershipState::Invite),
)?; &sender,
Some(invite_state),
true,
)
.await?;
} }
Ok(create_invite::v2::Response { Ok(create_invite::v2::Response {

477
src/service/admin/mod.rs
View file

@ -26,7 +26,7 @@ use ruma::{
EventId, OwnedRoomAliasId, RoomAliasId, RoomId, RoomVersionId, ServerName, UserId, EventId, OwnedRoomAliasId, RoomAliasId, RoomId, RoomVersionId, ServerName, UserId,
}; };
use serde_json::value::to_raw_value; use serde_json::value::to_raw_value;
use tokio::sync::{mpsc, Mutex, MutexGuard}; use tokio::sync::{mpsc, Mutex};
use crate::{ use crate::{
api::client_server::{leave_all_rooms, AUTO_GEN_PASSWORD_LENGTH}, api::client_server::{leave_all_rooms, AUTO_GEN_PASSWORD_LENGTH},
@ -206,26 +206,6 @@ impl Service {
.expect("Database data for admin room alias must be valid") .expect("Database data for admin room alias must be valid")
.expect("Admin room must exist"); .expect("Admin room must exist");
let send_message = |message: RoomMessageEventContent, mutex_lock: &MutexGuard<'_, ()>| {
services()
.rooms
.timeline
.build_and_append_pdu(
PduBuilder {
event_type: RoomEventType::RoomMessage,
content: to_raw_value(&message)
.expect("event is valid, we just created it"),
unsigned: None,
state_key: None,
redacts: None,
},
&conduit_user,
&conduit_room,
mutex_lock,
)
.unwrap();
};
loop { loop {
tokio::select! { tokio::select! {
Some(event) = receiver.recv() => { Some(event) = receiver.recv() => {
@ -245,7 +225,20 @@ impl Service {
let state_lock = mutex_state.lock().await; let state_lock = mutex_state.lock().await;
send_message(message_content, &state_lock); services().rooms.timeline.build_and_append_pdu(
PduBuilder {
event_type: RoomEventType::RoomMessage,
content: to_raw_value(&message_content)
.expect("event is valid, we just created it"),
unsigned: None,
state_key: None,
redacts: None,
},
&conduit_user,
&conduit_room,
&state_lock)
.await
.unwrap();
drop(state_lock); drop(state_lock);
} }
@ -853,164 +846,202 @@ impl Service {
content.room_version = services().globals.default_room_version(); content.room_version = services().globals.default_room_version();
// 1. The room create event // 1. The room create event
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomCreate, .timeline
content: to_raw_value(&content).expect("event is valid, we just created it"), .build_and_append_pdu(
unsigned: None, PduBuilder {
state_key: Some("".to_owned()), event_type: RoomEventType::RoomCreate,
redacts: None, content: to_raw_value(&content).expect("event is valid, we just created it"),
}, unsigned: None,
&conduit_user, state_key: Some("".to_owned()),
&room_id, redacts: None,
&state_lock, },
)?; &conduit_user,
&room_id,
&state_lock,
)
.await?;
// 2. Make conduit bot join // 2. Make conduit bot join
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomMember, .timeline
content: to_raw_value(&RoomMemberEventContent { .build_and_append_pdu(
membership: MembershipState::Join, PduBuilder {
displayname: None, event_type: RoomEventType::RoomMember,
avatar_url: None, content: to_raw_value(&RoomMemberEventContent {
is_direct: None, membership: MembershipState::Join,
third_party_invite: None, displayname: None,
blurhash: None, avatar_url: None,
reason: None, is_direct: None,
join_authorized_via_users_server: None, third_party_invite: None,
}) blurhash: None,
.expect("event is valid, we just created it"), reason: None,
unsigned: None, join_authorized_via_users_server: None,
state_key: Some(conduit_user.to_string()), })
redacts: None, .expect("event is valid, we just created it"),
}, unsigned: None,
&conduit_user, state_key: Some(conduit_user.to_string()),
&room_id, redacts: None,
&state_lock, },
)?; &conduit_user,
&room_id,
&state_lock,
)
.await?;
// 3. Power levels // 3. Power levels
let mut users = BTreeMap::new(); let mut users = BTreeMap::new();
users.insert(conduit_user.clone(), 100.into()); users.insert(conduit_user.clone(), 100.into());
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomPowerLevels, .timeline
content: to_raw_value(&RoomPowerLevelsEventContent { .build_and_append_pdu(
users, PduBuilder {
..Default::default() event_type: RoomEventType::RoomPowerLevels,
}) content: to_raw_value(&RoomPowerLevelsEventContent {
.expect("event is valid, we just created it"), users,
unsigned: None, ..Default::default()
state_key: Some("".to_owned()), })
redacts: None, .expect("event is valid, we just created it"),
}, unsigned: None,
&conduit_user, state_key: Some("".to_owned()),
&room_id, redacts: None,
&state_lock, },
)?; &conduit_user,
&room_id,
&state_lock,
)
.await?;
// 4.1 Join Rules // 4.1 Join Rules
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomJoinRules, .timeline
content: to_raw_value(&RoomJoinRulesEventContent::new(JoinRule::Invite)) .build_and_append_pdu(
.expect("event is valid, we just created it"), PduBuilder {
unsigned: None, event_type: RoomEventType::RoomJoinRules,
state_key: Some("".to_owned()), content: to_raw_value(&RoomJoinRulesEventContent::new(JoinRule::Invite))
redacts: None, .expect("event is valid, we just created it"),
}, unsigned: None,
&conduit_user, state_key: Some("".to_owned()),
&room_id, redacts: None,
&state_lock, },
)?; &conduit_user,
&room_id,
&state_lock,
)
.await?;
// 4.2 History Visibility // 4.2 History Visibility
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomHistoryVisibility, .timeline
content: to_raw_value(&RoomHistoryVisibilityEventContent::new( .build_and_append_pdu(
HistoryVisibility::Shared, PduBuilder {
)) event_type: RoomEventType::RoomHistoryVisibility,
.expect("event is valid, we just created it"), content: to_raw_value(&RoomHistoryVisibilityEventContent::new(
unsigned: None, HistoryVisibility::Shared,
state_key: Some("".to_owned()), ))
redacts: None, .expect("event is valid, we just created it"),
}, unsigned: None,
&conduit_user, state_key: Some("".to_owned()),
&room_id, redacts: None,
&state_lock, },
)?; &conduit_user,
&room_id,
&state_lock,
)
.await?;
// 4.3 Guest Access // 4.3 Guest Access
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomGuestAccess, .timeline
content: to_raw_value(&RoomGuestAccessEventContent::new(GuestAccess::Forbidden)) .build_and_append_pdu(
PduBuilder {
event_type: RoomEventType::RoomGuestAccess,
content: to_raw_value(&RoomGuestAccessEventContent::new(
GuestAccess::Forbidden,
))
.expect("event is valid, we just created it"), .expect("event is valid, we just created it"),
unsigned: None, unsigned: None,
state_key: Some("".to_owned()), state_key: Some("".to_owned()),
redacts: None, redacts: None,
}, },
&conduit_user, &conduit_user,
&room_id, &room_id,
&state_lock, &state_lock,
)?; )
.await?;
// 5. Events implied by name and topic // 5. Events implied by name and topic
let room_name = format!("{} Admin Room", services().globals.server_name()); let room_name = format!("{} Admin Room", services().globals.server_name());
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomName, .timeline
content: to_raw_value(&RoomNameEventContent::new(Some(room_name))) .build_and_append_pdu(
.expect("event is valid, we just created it"), PduBuilder {
unsigned: None, event_type: RoomEventType::RoomName,
state_key: Some("".to_owned()), content: to_raw_value(&RoomNameEventContent::new(Some(room_name)))
redacts: None, .expect("event is valid, we just created it"),
}, unsigned: None,
&conduit_user, state_key: Some("".to_owned()),
&room_id, redacts: None,
&state_lock, },
)?; &conduit_user,
&room_id,
&state_lock,
)
.await?;
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomTopic, .timeline
content: to_raw_value(&RoomTopicEventContent { .build_and_append_pdu(
topic: format!("Manage {}", services().globals.server_name()), PduBuilder {
}) event_type: RoomEventType::RoomTopic,
.expect("event is valid, we just created it"), content: to_raw_value(&RoomTopicEventContent {
unsigned: None, topic: format!("Manage {}", services().globals.server_name()),
state_key: Some("".to_owned()), })
redacts: None, .expect("event is valid, we just created it"),
}, unsigned: None,
&conduit_user, state_key: Some("".to_owned()),
&room_id, redacts: None,
&state_lock, },
)?; &conduit_user,
&room_id,
&state_lock,
)
.await?;
// 6. Room alias // 6. Room alias
let alias: OwnedRoomAliasId = format!("#admins:{}", services().globals.server_name()) let alias: OwnedRoomAliasId = format!("#admins:{}", services().globals.server_name())
.try_into() .try_into()
.expect("#admins:server_name is a valid alias name"); .expect("#admins:server_name is a valid alias name");
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomCanonicalAlias, .timeline
content: to_raw_value(&RoomCanonicalAliasEventContent { .build_and_append_pdu(
alias: Some(alias.clone()), PduBuilder {
alt_aliases: Vec::new(), event_type: RoomEventType::RoomCanonicalAlias,
}) content: to_raw_value(&RoomCanonicalAliasEventContent {
.expect("event is valid, we just created it"), alias: Some(alias.clone()),
unsigned: None, alt_aliases: Vec::new(),
state_key: Some("".to_owned()), })
redacts: None, .expect("event is valid, we just created it"),
}, unsigned: None,
&conduit_user, state_key: Some("".to_owned()),
&room_id, redacts: None,
&state_lock, },
)?; &conduit_user,
&room_id,
&state_lock,
)
.await?;
services().rooms.alias.set_alias(&alias, &room_id)?; services().rooms.alias.set_alias(&alias, &room_id)?;
@ -1052,72 +1083,84 @@ impl Service {
.expect("@conduit:server_name is valid"); .expect("@conduit:server_name is valid");
// Invite and join the real user // Invite and join the real user
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomMember, .timeline
content: to_raw_value(&RoomMemberEventContent { .build_and_append_pdu(
membership: MembershipState::Invite, PduBuilder {
displayname: None, event_type: RoomEventType::RoomMember,
avatar_url: None, content: to_raw_value(&RoomMemberEventContent {
is_direct: None, membership: MembershipState::Invite,
third_party_invite: None, displayname: None,
blurhash: None, avatar_url: None,
reason: None, is_direct: None,
join_authorized_via_users_server: None, third_party_invite: None,
}) blurhash: None,
.expect("event is valid, we just created it"), reason: None,
unsigned: None, join_authorized_via_users_server: None,
state_key: Some(user_id.to_string()), })
redacts: None, .expect("event is valid, we just created it"),
}, unsigned: None,
&conduit_user, state_key: Some(user_id.to_string()),
&room_id, redacts: None,
&state_lock, },
)?; &conduit_user,
services().rooms.timeline.build_and_append_pdu( &room_id,
PduBuilder { &state_lock,
event_type: RoomEventType::RoomMember, )
content: to_raw_value(&RoomMemberEventContent { .await?;
membership: MembershipState::Join, services()
displayname: Some(displayname), .rooms
avatar_url: None, .timeline
is_direct: None, .build_and_append_pdu(
third_party_invite: None, PduBuilder {
blurhash: None, event_type: RoomEventType::RoomMember,
reason: None, content: to_raw_value(&RoomMemberEventContent {
join_authorized_via_users_server: None, membership: MembershipState::Join,
}) displayname: Some(displayname),
.expect("event is valid, we just created it"), avatar_url: None,
unsigned: None, is_direct: None,
state_key: Some(user_id.to_string()), third_party_invite: None,
redacts: None, blurhash: None,
}, reason: None,
user_id, join_authorized_via_users_server: None,
&room_id, })
&state_lock, .expect("event is valid, we just created it"),
)?; unsigned: None,
state_key: Some(user_id.to_string()),
redacts: None,
},
user_id,
&room_id,
&state_lock,
)
.await?;
// Set power level // Set power level
let mut users = BTreeMap::new(); let mut users = BTreeMap::new();
users.insert(conduit_user.to_owned(), 100.into()); users.insert(conduit_user.to_owned(), 100.into());
users.insert(user_id.to_owned(), 100.into()); users.insert(user_id.to_owned(), 100.into());
services().rooms.timeline.build_and_append_pdu( services()
PduBuilder { .rooms
event_type: RoomEventType::RoomPowerLevels, .timeline
content: to_raw_value(&RoomPowerLevelsEventContent { .build_and_append_pdu(
users, PduBuilder {
..Default::default() event_type: RoomEventType::RoomPowerLevels,
}) content: to_raw_value(&RoomPowerLevelsEventContent {
.expect("event is valid, we just created it"), users,
unsigned: None, ..Default::default()
state_key: Some("".to_owned()), })
redacts: None, .expect("event is valid, we just created it"),
}, unsigned: None,
&conduit_user, state_key: Some("".to_owned()),
&room_id, redacts: None,
&state_lock, },
)?; &conduit_user,
&room_id,
&state_lock,
)
.await?;
// Send welcome message // Send welcome message
services().rooms.timeline.build_and_append_pdu( services().rooms.timeline.build_and_append_pdu(
@ -1135,7 +1178,7 @@ impl Service {
&conduit_user, &conduit_user,
&room_id, &room_id,
&state_lock, &state_lock,
)?; ).await?;
Ok(()) Ok(())
} }

2
src/service/globals/mod.rs
View file

@ -52,7 +52,7 @@ pub struct Service {
pub bad_signature_ratelimiter: Arc<RwLock<HashMap<Vec<String>, RateLimitState>>>, pub bad_signature_ratelimiter: Arc<RwLock<HashMap<Vec<String>, RateLimitState>>>,
pub servername_ratelimiter: Arc<RwLock<HashMap<OwnedServerName, Arc<Semaphore>>>>, pub servername_ratelimiter: Arc<RwLock<HashMap<OwnedServerName, Arc<Semaphore>>>>,
pub sync_receivers: RwLock<HashMap<(OwnedUserId, OwnedDeviceId), SyncHandle>>, pub sync_receivers: RwLock<HashMap<(OwnedUserId, OwnedDeviceId), SyncHandle>>,
pub roomid_mutex_insert: RwLock<HashMap<OwnedRoomId, Arc<Mutex<()>>>>, pub roomid_mutex_insert: RwLock<HashMap<OwnedRoomId, Arc<TokioMutex<()>>>>,
pub roomid_mutex_state: RwLock<HashMap<OwnedRoomId, Arc<TokioMutex<()>>>>, pub roomid_mutex_state: RwLock<HashMap<OwnedRoomId, Arc<TokioMutex<()>>>>,
pub roomid_mutex_federation: RwLock<HashMap<OwnedRoomId, Arc<TokioMutex<()>>>>, // this lock will be held longer pub roomid_mutex_federation: RwLock<HashMap<OwnedRoomId, Arc<TokioMutex<()>>>>, // this lock will be held longer
pub roomid_federationhandletime: RwLock<HashMap<OwnedRoomId, (OwnedEventId, Instant)>>, pub roomid_federationhandletime: RwLock<HashMap<OwnedRoomId, (OwnedEventId, Instant)>>,

40
src/service/rooms/event_handler/mod.rs
View file

@ -801,14 +801,18 @@ impl Service {
.map_err(|_e| Error::BadRequest(ErrorKind::InvalidParam, "Auth check failed."))?; .map_err(|_e| Error::BadRequest(ErrorKind::InvalidParam, "Auth check failed."))?;
if soft_fail { if soft_fail {
services().rooms.timeline.append_incoming_pdu( services()
&incoming_pdu, .rooms
val, .timeline
extremities.iter().map(|e| (**e).to_owned()).collect(), .append_incoming_pdu(
state_ids_compressed, &incoming_pdu,
soft_fail, val,
&state_lock, extremities.iter().map(|e| (**e).to_owned()).collect(),
)?; state_ids_compressed,
soft_fail,
&state_lock,
)
.await?;
// Soft fail, we keep the event as an outlier but don't add it to the timeline // Soft fail, we keep the event as an outlier but don't add it to the timeline
warn!("Event was soft failed: {:?}", incoming_pdu); warn!("Event was soft failed: {:?}", incoming_pdu);
@ -1004,14 +1008,18 @@ impl Service {
// We use the `state_at_event` instead of `state_after` so we accurately // We use the `state_at_event` instead of `state_after` so we accurately
// represent the state for this event. // represent the state for this event.
let pdu_id = services().rooms.timeline.append_incoming_pdu( let pdu_id = services()
&incoming_pdu, .rooms
val, .timeline
extremities.iter().map(|e| (**e).to_owned()).collect(), .append_incoming_pdu(
state_ids_compressed, &incoming_pdu,
soft_fail, val,
&state_lock, extremities.iter().map(|e| (**e).to_owned()).collect(),
)?; state_ids_compressed,
soft_fail,
&state_lock,
)
.await?;
info!("Appended incoming pdu"); info!("Appended incoming pdu");

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

@ -7,14 +7,13 @@ use std::{
pub use data::Data; pub use data::Data;
use ruma::{ use ruma::{
events::{ events::{
room::{create::RoomCreateEventContent, member::MembershipState}, room::{create::RoomCreateEventContent, member::RoomMemberEventContent},
AnyStrippedStateEvent, RoomEventType, StateEventType, AnyStrippedStateEvent, RoomEventType, StateEventType,
}, },
serde::Raw, serde::Raw,
state_res::{self, StateMap}, state_res::{self, StateMap},
EventId, OwnedEventId, RoomId, RoomVersionId, UserId, EventId, OwnedEventId, RoomId, RoomVersionId, UserId,
}; };
use serde::Deserialize;
use tokio::sync::MutexGuard; use tokio::sync::MutexGuard;
use tracing::warn; use tracing::warn;
@ -60,15 +59,11 @@ impl Service {
Err(_) => continue, Err(_) => continue,
}; };
#[derive(Deserialize)] let membership_event =
struct ExtractMembership { match serde_json::from_str::<RoomMemberEventContent>(pdu.content.get()) {
membership: MembershipState, Ok(e) => e,
} Err(_) => continue,
};
let membership = match serde_json::from_str::<ExtractMembership>(pdu.content.get()) {
Ok(e) => e.membership,
Err(_) => continue,
};
let state_key = match pdu.state_key { let state_key = match pdu.state_key {
Some(k) => k, Some(k) => k,
@ -80,14 +75,18 @@ impl Service {
Err(_) => continue, Err(_) => continue,
}; };
services().rooms.state_cache.update_membership( services()
room_id, .rooms
&user_id, .state_cache
membership, .update_membership(
&pdu.sender, room_id,
None, &user_id,
false, membership_event,
)?; &pdu.sender,
None,
false,
)
.await?;
} }
services().rooms.state_cache.update_joined_count(room_id)?; services().rooms.state_cache.update_joined_count(room_id)?;

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

@ -4,10 +4,14 @@ use std::{collections::HashSet, sync::Arc};
pub use data::Data; pub use data::Data;
use ruma::{ use ruma::{
api::federation::{self, query::get_profile_information::v1::ProfileField},
events::{ events::{
direct::DirectEvent, direct::DirectEvent,
ignored_user_list::IgnoredUserListEvent, ignored_user_list::IgnoredUserListEvent,
room::{create::RoomCreateEventContent, member::MembershipState}, room::{
create::RoomCreateEventContent,
member::{MembershipState, RoomMemberEventContent},
},
AnyStrippedStateEvent, AnySyncStateEvent, GlobalAccountDataEventType, AnyStrippedStateEvent, AnySyncStateEvent, GlobalAccountDataEventType,
RoomAccountDataEventType, StateEventType, RoomAccountDataEventType, StateEventType,
}, },
@ -24,19 +28,43 @@ pub struct Service {
impl Service { impl Service {
/// Update current membership data. /// Update current membership data.
#[tracing::instrument(skip(self, last_state))] #[tracing::instrument(skip(self, last_state))]
pub fn update_membership( pub async fn update_membership(
&self, &self,
room_id: &RoomId, room_id: &RoomId,
user_id: &UserId, user_id: &UserId,
membership: MembershipState, membership_event: RoomMemberEventContent,
sender: &UserId, sender: &UserId,
last_state: Option<Vec<Raw<AnyStrippedStateEvent>>>, last_state: Option<Vec<Raw<AnyStrippedStateEvent>>>,
update_joined_count: bool, update_joined_count: bool,
) -> Result<()> { ) -> Result<()> {
let membership = membership_event.membership;
// Keep track what remote users exist by adding them as "deactivated" users // Keep track what remote users exist by adding them as "deactivated" users
if user_id.server_name() != services().globals.server_name() { if user_id.server_name() != services().globals.server_name() {
services().users.create(user_id, None)?; services().users.create(user_id, None)?;
// TODO: displayname, avatar url // Try to update our local copy of the user if ours does not match
if ((services().users.displayname(user_id)? != membership_event.displayname)
|| (services().users.avatar_url(user_id)? != membership_event.avatar_url)
|| (services().users.blurhash(user_id)? != membership_event.blurhash))
&& (membership != MembershipState::Leave)
{
let response = services()
.sending
.send_federation_request(
user_id.server_name(),
federation::query::get_profile_information::v1::Request {
user_id: user_id.into(),
field: Some(ProfileField::AvatarUrl),
},
)
.await?;
let _ = services()
.users
.set_displayname(user_id, response.displayname.clone());
let _ = services()
.users
.set_avatar_url(user_id, response.avatar_url);
let _ = services().users.set_blurhash(user_id, response.blurhash);
};
} }
match &membership { match &membership {

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

@ -15,7 +15,8 @@ use ruma::{
events::{ events::{
push_rules::PushRulesEvent, push_rules::PushRulesEvent,
room::{ room::{
create::RoomCreateEventContent, member::MembershipState, create::RoomCreateEventContent,
member::{MembershipState, RoomMemberEventContent},
power_levels::RoomPowerLevelsEventContent, power_levels::RoomPowerLevelsEventContent,
}, },
GlobalAccountDataEventType, RoomEventType, StateEventType, GlobalAccountDataEventType, RoomEventType, StateEventType,
@ -145,7 +146,7 @@ impl Service {
/// ///
/// Returns pdu id /// Returns pdu id
#[tracing::instrument(skip(self, pdu, pdu_json, leaves))] #[tracing::instrument(skip(self, pdu, pdu_json, leaves))]
pub fn append_pdu<'a>( pub async fn append_pdu<'a>(
&self, &self,
pdu: &PduEvent, pdu: &PduEvent,
mut pdu_json: CanonicalJsonObject, mut pdu_json: CanonicalJsonObject,
@ -211,7 +212,7 @@ impl Service {
.entry(pdu.room_id.clone()) .entry(pdu.room_id.clone())
.or_default(), .or_default(),
); );
let insert_lock = mutex_insert.lock().unwrap(); let insert_lock = mutex_insert.lock().await;
let count1 = services().globals.next_count()?; let count1 = services().globals.next_count()?;
// Mark as read first so the sending client doesn't get a notification even if appending // Mark as read first so the sending client doesn't get a notification even if appending
@ -323,16 +324,11 @@ impl Service {
} }
RoomEventType::RoomMember => { RoomEventType::RoomMember => {
if let Some(state_key) = &pdu.state_key { if let Some(state_key) = &pdu.state_key {
#[derive(Deserialize)]
struct ExtractMembership {
membership: MembershipState,
}
// if the state_key fails // if the state_key fails
let target_user_id = UserId::parse(state_key.clone()) let target_user_id = UserId::parse(state_key.clone())
.expect("This state_key was previously validated"); .expect("This state_key was previously validated");
let content = serde_json::from_str::<ExtractMembership>(pdu.content.get()) let content = serde_json::from_str::<RoomMemberEventContent>(pdu.content.get())
.map_err(|_| Error::bad_database("Invalid content in pdu."))?; .map_err(|_| Error::bad_database("Invalid content in pdu."))?;
let invite_state = match content.membership { let invite_state = match content.membership {
@ -345,14 +341,18 @@ impl Service {
// Update our membership info, we do this here incase a user is invited // Update our membership info, we do this here incase a user is invited
// and immediately leaves we need the DB to record the invite event for auth // and immediately leaves we need the DB to record the invite event for auth
services().rooms.state_cache.update_membership( services()
&pdu.room_id, .rooms
&target_user_id, .state_cache
content.membership, .update_membership(
&pdu.sender, &pdu.room_id,
invite_state, &target_user_id,
true, content,
)?; &pdu.sender,
invite_state,
true,
)
.await?;
} }
} }
RoomEventType::RoomMessage => { RoomEventType::RoomMessage => {
@ -673,7 +673,7 @@ impl Service {
/// Creates a new persisted data unit and adds it to a room. This function takes a /// Creates a new persisted data unit and adds it to a room. This function takes a
/// roomid_mutex_state, meaning that only this function is able to mutate the room state. /// roomid_mutex_state, meaning that only this function is able to mutate the room state.
#[tracing::instrument(skip(self, state_lock))] #[tracing::instrument(skip(self, state_lock))]
pub fn build_and_append_pdu( pub async fn build_and_append_pdu(
&self, &self,
pdu_builder: PduBuilder, pdu_builder: PduBuilder,
sender: &UserId, sender: &UserId,
@ -687,14 +687,16 @@ impl Service {
// pdu without it's state. This is okay because append_pdu can't fail. // pdu without it's state. This is okay because append_pdu can't fail.
let statehashid = services().rooms.state.append_to_state(&pdu)?; let statehashid = services().rooms.state.append_to_state(&pdu)?;
let pdu_id = self.append_pdu( let pdu_id = self
&pdu, .append_pdu(
pdu_json, &pdu,
// Since this PDU references all pdu_leaves we can update the leaves pdu_json,
// of the room // Since this PDU references all pdu_leaves we can update the leaves
vec![(*pdu.event_id).to_owned()], // of the room
state_lock, vec![(*pdu.event_id).to_owned()],
)?; state_lock,
)
.await?;
// We set the room state after inserting the pdu, so that we never have a moment in time // We set the room state after inserting the pdu, so that we never have a moment in time
// where events in the current room state do not exist // where events in the current room state do not exist
@ -732,7 +734,7 @@ impl Service {
/// Append the incoming event setting the state snapshot to the state from the /// Append the incoming event setting the state snapshot to the state from the
/// server that sent the event. /// server that sent the event.
#[tracing::instrument(skip_all)] #[tracing::instrument(skip_all)]
pub fn append_incoming_pdu<'a>( pub async fn append_incoming_pdu<'a>(
&self, &self,
pdu: &PduEvent, pdu: &PduEvent,
pdu_json: CanonicalJsonObject, pdu_json: CanonicalJsonObject,
@ -762,11 +764,11 @@ impl Service {
return Ok(None); return Ok(None);
} }
let pdu_id = let pdu_id = services()
services() .rooms
.rooms .timeline
.timeline .append_pdu(pdu, pdu_json, new_room_leaves, state_lock)
.append_pdu(pdu, pdu_json, new_room_leaves, state_lock)?; .await?;
Ok(Some(pdu_id)) Ok(Some(pdu_id))
} }