magmaus3/continuwuity
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix every clippy warning possible, remove io_uring as default feature

Browse source 
this project's codebase is so horrendous, im shocked that no one has ran
clippy at all. it had ~200 total lint warnings, some with performance
issues and unsoundness, and the rest just very ugly codebase. i have sat
down and fixed as many of these as possible and i am exhausted.
i haven't fixed some extremely complex ones, but i brought it down from
~200 to ~30.

i have also removed io_uring as a default feature due to it falling
under the same category as linux eBPF: major kernel attack surface for
minimal performance gains. this also makes it impossible to cross-compile
from macOS to Linux because io_uring does not exist in Darwin land.
there are far better ways to achieve better performance than io_uring on
the codebase level.

Signed-off-by: strawberry <june@girlboss.ceo>
This commit is contained in:
strawberry 2023-11-27 00:39:50 -05:00
parent 19d1b484e0
commit 54a3f47851
33 changed files with 312 additions and 314 deletions
Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -1362,6 +1362,7 @@ dependencies = [
"libc", "libc",
"libz-sys", "libz-sys",
"lz4-sys", "lz4-sys",
"pkg-config",
"zstd-sys", "zstd-sys",
] ]

3
Cargo.toml
View file

@ -117,7 +117,7 @@ js_option = "0.1"
nix = { version = "0.26.2", features = ["resource"] } nix = { version = "0.26.2", features = ["resource"] }
[features] [features]
default = ["conduit_bin", "backend_sqlite", "backend_rocksdb", "systemd", "zstd_compression"] default = ["conduit_bin", "backend_rocksdb", "systemd", "zstd_compression"]
#backend_sled = ["sled"] #backend_sled = ["sled"]
backend_persy = ["persy", "parking_lot"] backend_persy = ["persy", "parking_lot"]
backend_sqlite = ["sqlite"] backend_sqlite = ["sqlite"]
@ -132,6 +132,7 @@ zstd_compression = []
#brotli_compression = ["tower-http/compression-br"] #brotli_compression = ["tower-http/compression-br"]
#compression = ["tower-http/compression-full"] #compression = ["tower-http/compression-full"]
sha256_media = [] sha256_media = []
io_uring = ["rocksdb/io-uring"]
[[bin]] [[bin]]
name = "conduit" name = "conduit"

4
src/api/client_server/keys.rs
View file

@ -367,7 +367,7 @@ pub(crate) async fn get_keys_helper<F: Fn(&UserId) -> bool>(
.bad_query_ratelimiter .bad_query_ratelimiter
.read() .read()
.unwrap() .unwrap()
.get(&*server) .get(server)
{ {
// Exponential backoff // Exponential backoff
let mut min_elapsed_duration = Duration::from_secs(30) * (*tries) * (*tries); let mut min_elapsed_duration = Duration::from_secs(30) * (*tries) * (*tries);
@ -468,7 +468,7 @@ fn add_unsigned_device_display_name(
} else { } else {
unsigned_object.insert( unsigned_object.insert(
"device_display_name".to_owned(), "device_display_name".to_owned(),
Some(metadata.device_id.as_str().to_string()).into(), Some(metadata.device_id.as_str().to_owned()).into(),
); );
} }
} }

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

@ -412,7 +412,7 @@ pub async fn get_member_events_route(
if !services() if !services()
.rooms .rooms
.state_accessor .state_accessor
.user_can_see_state_events(&sender_user, &body.room_id)? .user_can_see_state_events(sender_user, &body.room_id)?
{ {
return Err(Error::BadRequest( return Err(Error::BadRequest(
ErrorKind::Forbidden, ErrorKind::Forbidden,
@ -447,7 +447,7 @@ pub async fn joined_members_route(
if !services() if !services()
.rooms .rooms
.state_accessor .state_accessor
.user_can_see_state_events(&sender_user, &body.room_id)? .user_can_see_state_events(sender_user, &body.room_id)?
{ {
return Err(Error::BadRequest( return Err(Error::BadRequest(
ErrorKind::Forbidden, ErrorKind::Forbidden,

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

@ -128,7 +128,7 @@ pub async fn get_message_events_route(
let to = body let to = body
.to .to
.as_ref() .as_ref()
.and_then(|t| PduCount::try_from_string(&t).ok()); .and_then(|t| PduCount::try_from_string(t).ok());
services().rooms.lazy_loading.lazy_load_confirm_delivery( services().rooms.lazy_loading.lazy_load_confirm_delivery(
sender_user, sender_user,

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

@ -23,7 +23,7 @@ pub async fn get_relating_events_with_rel_type_and_event_type_route(
let to = body let to = body
.to .to
.as_ref() .as_ref()
.and_then(|t| PduCount::try_from_string(&t).ok()); .and_then(|t| PduCount::try_from_string(t).ok());
// Use limit or else 10, with maximum 100 // Use limit or else 10, with maximum 100
let limit = body let limit = body
@ -73,7 +73,7 @@ pub async fn get_relating_events_with_rel_type_route(
let to = body let to = body
.to .to
.as_ref() .as_ref()
.and_then(|t| PduCount::try_from_string(&t).ok()); .and_then(|t| PduCount::try_from_string(t).ok());
// Use limit or else 10, with maximum 100 // Use limit or else 10, with maximum 100
let limit = body let limit = body
@ -121,7 +121,7 @@ pub async fn get_relating_events_route(
let to = body let to = body
.to .to
.as_ref() .as_ref()
.and_then(|t| PduCount::try_from_string(&t).ok()); .and_then(|t| PduCount::try_from_string(t).ok());
// Use limit or else 10, with maximum 100 // Use limit or else 10, with maximum 100
let limit = body let limit = body

11
src/api/client_server/session.rs
View file

@ -117,12 +117,10 @@ pub async fn login_route(body: Ruma<login::v3::Request>) -> Result<login::v3::Re
} else { } else {
return Err(Error::BadRequest(ErrorKind::Forbidden, "Bad login type.")); return Err(Error::BadRequest(ErrorKind::Forbidden, "Bad login type."));
}; };
let user_id =
UserId::parse_with_server_name(username, services().globals.server_name()) UserId::parse_with_server_name(username, services().globals.server_name()).map_err(
.map_err(|_| { |_| Error::BadRequest(ErrorKind::InvalidUsername, "Username is invalid."),
Error::BadRequest(ErrorKind::InvalidUsername, "Username is invalid.") )?
})?;
user_id
} }
_ => { _ => {
warn!("Unsupported or unknown login type: {:?}", &body.login_info); warn!("Unsupported or unknown login type: {:?}", &body.login_info);
@ -163,6 +161,7 @@ pub async fn login_route(body: Ruma<login::v3::Request>) -> Result<login::v3::Re
info!("{} logged in", user_id); info!("{} logged in", user_id);
#[allow(deprecated)]
Ok(login::v3::Response { Ok(login::v3::Response {
user_id, user_id,
access_token: token, access_token: token,

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

@ -85,7 +85,7 @@ pub async fn get_state_events_route(
if !services() if !services()
.rooms .rooms
.state_accessor .state_accessor
.user_can_see_state_events(&sender_user, &body.room_id)? .user_can_see_state_events(sender_user, &body.room_id)?
{ {
return Err(Error::BadRequest( return Err(Error::BadRequest(
ErrorKind::Forbidden, ErrorKind::Forbidden,
@ -118,7 +118,7 @@ pub async fn get_state_events_for_key_route(
if !services() if !services()
.rooms .rooms
.state_accessor .state_accessor
.user_can_see_state_events(&sender_user, &body.room_id)? .user_can_see_state_events(sender_user, &body.room_id)?
{ {
return Err(Error::BadRequest( return Err(Error::BadRequest(
ErrorKind::Forbidden, ErrorKind::Forbidden,
@ -157,7 +157,7 @@ pub async fn get_state_events_for_empty_key_route(
if !services() if !services()
.rooms .rooms
.state_accessor .state_accessor
.user_can_see_state_events(&sender_user, &body.room_id)? .user_can_see_state_events(sender_user, &body.room_id)?
{ {
return Err(Error::BadRequest( return Err(Error::BadRequest(
ErrorKind::Forbidden, ErrorKind::Forbidden,

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

@ -24,6 +24,7 @@ use ruma::{
uint, DeviceId, OwnedDeviceId, OwnedUserId, RoomId, UInt, UserId, uint, DeviceId, OwnedDeviceId, OwnedUserId, RoomId, UInt, UserId,
}; };
use std::{ use std::{
cmp::Ordering,
collections::{hash_map::Entry, BTreeMap, BTreeSet, HashMap, HashSet}, collections::{hash_map::Entry, BTreeMap, BTreeSet, HashMap, HashSet},
sync::Arc, sync::Arc,
time::Duration, time::Duration,
@ -573,6 +574,7 @@ async fn process_room_presence_updates(
Ok(()) Ok(())
} }
#[allow(clippy::too_many_arguments)]
async fn load_joined_room( async fn load_joined_room(
sender_user: &UserId, sender_user: &UserId,
sender_device: &DeviceId, sender_device: &DeviceId,
@ -609,7 +611,7 @@ async fn load_joined_room(
|| services() || services()
.rooms .rooms
.user .user
.last_notification_read(&sender_user, &room_id)? .last_notification_read(sender_user, room_id)?
> since; > since;
let mut timeline_users = HashSet::new(); let mut timeline_users = HashSet::new();
@ -618,16 +620,16 @@ async fn load_joined_room(
} }
services().rooms.lazy_loading.lazy_load_confirm_delivery( services().rooms.lazy_loading.lazy_load_confirm_delivery(
&sender_user, sender_user,
&sender_device, sender_device,
&room_id, room_id,
sincecount, sincecount,
)?; )?;
// Database queries: // Database queries:
let current_shortstatehash = let current_shortstatehash =
if let Some(s) = services().rooms.state.get_room_shortstatehash(&room_id)? { if let Some(s) = services().rooms.state.get_room_shortstatehash(room_id)? {
s s
} else { } else {
error!("Room {} has no state", room_id); error!("Room {} has no state", room_id);
@ -637,7 +639,7 @@ async fn load_joined_room(
let since_shortstatehash = services() let since_shortstatehash = services()
.rooms .rooms
.user .user
.get_token_shortstatehash(&room_id, since)?; .get_token_shortstatehash(room_id, since)?;
let (heroes, joined_member_count, invited_member_count, joined_since_last_sync, state_events) = let (heroes, joined_member_count, invited_member_count, joined_since_last_sync, state_events) =
if timeline_pdus.is_empty() && since_shortstatehash == Some(current_shortstatehash) { if timeline_pdus.is_empty() && since_shortstatehash == Some(current_shortstatehash) {
@ -649,12 +651,12 @@ async fn load_joined_room(
let joined_member_count = services() let joined_member_count = services()
.rooms .rooms
.state_cache .state_cache
.room_joined_count(&room_id)? .room_joined_count(room_id)?
.unwrap_or(0); .unwrap_or(0);
let invited_member_count = services() let invited_member_count = services()
.rooms .rooms
.state_cache .state_cache
.room_invited_count(&room_id)? .room_invited_count(room_id)?
.unwrap_or(0); .unwrap_or(0);
// Recalculate heroes (first 5 members) // Recalculate heroes (first 5 members)
@ -667,7 +669,7 @@ async fn load_joined_room(
for hero in services() for hero in services()
.rooms .rooms
.timeline .timeline
.all_pdus(&sender_user, &room_id)? .all_pdus(sender_user, room_id)?
.filter_map(|pdu| pdu.ok()) // Ignore all broken pdus .filter_map(|pdu| pdu.ok()) // Ignore all broken pdus
.filter(|(_, pdu)| pdu.kind == TimelineEventType::RoomMember) .filter(|(_, pdu)| pdu.kind == TimelineEventType::RoomMember)
.map(|(_, pdu)| { .map(|(_, pdu)| {
@ -688,11 +690,11 @@ async fn load_joined_room(
) && (services() ) && (services()
.rooms .rooms
.state_cache .state_cache
.is_joined(&user_id, &room_id)? .is_joined(&user_id, room_id)?
|| services() || services()
.rooms .rooms
.state_cache .state_cache
.is_invited(&user_id, &room_id)?) .is_invited(&user_id, room_id)?)
{ {
Ok::<_, Error>(Some(state_key.clone())) Ok::<_, Error>(Some(state_key.clone()))
} else { } else {
@ -808,17 +810,17 @@ async fn load_joined_room(
// Reset lazy loading because this is an initial sync // Reset lazy loading because this is an initial sync
services().rooms.lazy_loading.lazy_load_reset( services().rooms.lazy_loading.lazy_load_reset(
&sender_user, sender_user,
&sender_device, sender_device,
&room_id, room_id,
)?; )?;
// The state_events above should contain all timeline_users, let's mark them as lazy // The state_events above should contain all timeline_users, let's mark them as lazy
// loaded. // loaded.
services().rooms.lazy_loading.lazy_load_mark_sent( services().rooms.lazy_loading.lazy_load_mark_sent(
&sender_user, sender_user,
&sender_device, sender_device,
&room_id, room_id,
lazy_loaded, lazy_loaded,
next_batchcount, next_batchcount,
); );
@ -885,14 +887,14 @@ async fn load_joined_room(
} }
if !services().rooms.lazy_loading.lazy_load_was_sent_before( if !services().rooms.lazy_loading.lazy_load_was_sent_before(
&sender_user, sender_user,
&sender_device, sender_device,
&room_id, room_id,
&event.sender, &event.sender,
)? || lazy_load_send_redundant )? || lazy_load_send_redundant
{ {
if let Some(member_event) = services().rooms.state_accessor.room_state_get( if let Some(member_event) = services().rooms.state_accessor.room_state_get(
&room_id, room_id,
&StateEventType::RoomMember, &StateEventType::RoomMember,
event.sender.as_str(), event.sender.as_str(),
)? { )? {
@ -903,9 +905,9 @@ async fn load_joined_room(
} }
services().rooms.lazy_loading.lazy_load_mark_sent( services().rooms.lazy_loading.lazy_load_mark_sent(
&sender_user, sender_user,
&sender_device, sender_device,
&room_id, room_id,
lazy_loaded, lazy_loaded,
next_batchcount, next_batchcount,
); );
@ -953,7 +955,7 @@ async fn load_joined_room(
match new_membership { match new_membership {
MembershipState::Join => { MembershipState::Join => {
// A new user joined an encrypted room // A new user joined an encrypted room
if !share_encrypted_room(&sender_user, &user_id, &room_id)? { if !share_encrypted_room(sender_user, &user_id, room_id)? {
device_list_updates.insert(user_id); device_list_updates.insert(user_id);
} }
} }
@ -973,15 +975,15 @@ async fn load_joined_room(
services() services()
.rooms .rooms
.state_cache .state_cache
.room_members(&room_id) .room_members(room_id)
.flatten() .flatten()
.filter(|user_id| { .filter(|user_id| {
// Don't send key updates from the sender to the sender // Don't send key updates from the sender to the sender
&sender_user != user_id sender_user != user_id
}) })
.filter(|user_id| { .filter(|user_id| {
// Only send keys if the sender doesn't share an encrypted room with the target already // Only send keys if the sender doesn't share an encrypted room with the target already
!share_encrypted_room(&sender_user, user_id, &room_id) !share_encrypted_room(sender_user, user_id, room_id)
.unwrap_or(false) .unwrap_or(false)
}), }),
); );
@ -1016,7 +1018,7 @@ async fn load_joined_room(
services() services()
.rooms .rooms
.user .user
.notification_count(&sender_user, &room_id)? .notification_count(sender_user, room_id)?
.try_into() .try_into()
.expect("notification count can't go that high"), .expect("notification count can't go that high"),
) )
@ -1029,7 +1031,7 @@ async fn load_joined_room(
services() services()
.rooms .rooms
.user .user
.highlight_count(&sender_user, &room_id)? .highlight_count(sender_user, room_id)?
.try_into() .try_into()
.expect("highlight count can't go that high"), .expect("highlight count can't go that high"),
) )
@ -1058,15 +1060,15 @@ async fn load_joined_room(
.rooms .rooms
.edus .edus
.read_receipt .read_receipt
.readreceipts_since(&room_id, since) .readreceipts_since(room_id, since)
.filter_map(|r| r.ok()) // Filter out buggy events .filter_map(|r| r.ok()) // Filter out buggy events
.map(|(_, _, v)| v) .map(|(_, _, v)| v)
.collect(); .collect();
if services().rooms.edus.typing.last_typing_update(&room_id)? > since { if services().rooms.edus.typing.last_typing_update(room_id)? > since {
edus.push( edus.push(
serde_json::from_str( serde_json::from_str(
&serde_json::to_string(&services().rooms.edus.typing.typings_all(&room_id)?) &serde_json::to_string(&services().rooms.edus.typing.typings_all(room_id)?)
.expect("event is valid, we just created it"), .expect("event is valid, we just created it"),
) )
.expect("event is valid, we just created it"), .expect("event is valid, we just created it"),
@ -1075,7 +1077,7 @@ async fn load_joined_room(
// Save the state after this sync so we can send the correct state diff next sync // Save the state after this sync so we can send the correct state diff next sync
services().rooms.user.associate_token_shortstatehash( services().rooms.user.associate_token_shortstatehash(
&room_id, room_id,
next_batch, next_batch,
current_shortstatehash, current_shortstatehash,
)?; )?;
@ -1084,7 +1086,7 @@ async fn load_joined_room(
account_data: RoomAccountData { account_data: RoomAccountData {
events: services() events: services()
.account_data .account_data
.changes_since(Some(&room_id), &sender_user, since)? .changes_since(Some(room_id), sender_user, since)?
.into_iter() .into_iter()
.filter_map(|(_, v)| { .filter_map(|(_, v)| {
serde_json::from_str(v.json().get()) serde_json::from_str(v.json().get())
@ -1129,13 +1131,13 @@ fn load_timeline(
if services() if services()
.rooms .rooms
.timeline .timeline
.last_timeline_count(&sender_user, &room_id)? .last_timeline_count(sender_user, room_id)?
> roomsincecount > roomsincecount
{ {
let mut non_timeline_pdus = services() let mut non_timeline_pdus = services()
.rooms .rooms
.timeline .timeline
.pdus_until(&sender_user, &room_id, PduCount::max())? .pdus_until(sender_user, room_id, PduCount::max())?
.filter_map(|r| { .filter_map(|r| {
// Filter out buggy events // Filter out buggy events
if r.is_err() { if r.is_err() {
@ -1250,7 +1252,7 @@ pub async fn sync_events_v4_route(
for room_id in &all_joined_rooms { for room_id in &all_joined_rooms {
let current_shortstatehash = let current_shortstatehash =
if let Some(s) = services().rooms.state.get_room_shortstatehash(&room_id)? { if let Some(s) = services().rooms.state.get_room_shortstatehash(room_id)? {
s s
} else { } else {
error!("Room {} has no state", room_id); error!("Room {} has no state", room_id);
@ -1260,7 +1262,7 @@ pub async fn sync_events_v4_route(
let since_shortstatehash = services() let since_shortstatehash = services()
.rooms .rooms
.user .user
.get_token_shortstatehash(&room_id, globalsince)?; .get_token_shortstatehash(room_id, globalsince)?;
let since_sender_member: Option<RoomMemberEventContent> = since_shortstatehash let since_sender_member: Option<RoomMemberEventContent> = since_shortstatehash
.and_then(|shortstatehash| { .and_then(|shortstatehash| {
@ -1349,7 +1351,7 @@ pub async fn sync_events_v4_route(
if !share_encrypted_room( if !share_encrypted_room(
&sender_user, &sender_user,
&user_id, &user_id,
&room_id, room_id,
)? { )? {
device_list_changes.insert(user_id); device_list_changes.insert(user_id);
} }
@ -1370,7 +1372,7 @@ pub async fn sync_events_v4_route(
services() services()
.rooms .rooms
.state_cache .state_cache
.room_members(&room_id) .room_members(room_id)
.flatten() .flatten()
.filter(|user_id| { .filter(|user_id| {
// Don't send key updates from the sender to the sender // Don't send key updates from the sender to the sender
@ -1378,7 +1380,7 @@ pub async fn sync_events_v4_route(
}) })
.filter(|user_id| { .filter(|user_id| {
// Only send keys if the sender doesn't share an encrypted room with the target already // Only send keys if the sender doesn't share an encrypted room with the target already
!share_encrypted_room(&sender_user, user_id, &room_id) !share_encrypted_room(&sender_user, user_id, room_id)
.unwrap_or(false) .unwrap_or(false)
}), }),
); );
@ -1469,7 +1471,7 @@ pub async fn sync_events_v4_route(
} }
sync_events::v4::SyncOp { sync_events::v4::SyncOp {
op: SlidingOp::Sync, op: SlidingOp::Sync,
range: Some(r.clone()), range: Some(r),
index: None, index: None,
room_ids, room_ids,
room_id: None, room_id: None,
@ -1541,7 +1543,7 @@ pub async fn sync_events_v4_route(
let roomsincecount = PduCount::Normal(*roomsince); let roomsincecount = PduCount::Normal(*roomsince);
let (timeline_pdus, limited) = let (timeline_pdus, limited) =
load_timeline(&sender_user, &room_id, roomsincecount, *timeline_limit)?; load_timeline(&sender_user, room_id, roomsincecount, *timeline_limit)?;
if roomsince != &0 && timeline_pdus.is_empty() { if roomsince != &0 && timeline_pdus.is_empty() {
continue; continue;
@ -1577,10 +1579,10 @@ pub async fn sync_events_v4_route(
services() services()
.rooms .rooms
.state_accessor .state_accessor
.room_state_get(&room_id, &state.0, &state.1) .room_state_get(room_id, &state.0, &state.1)
}) })
.filter_map(|r| r.ok()) .filter_map(|r| r.ok())
.filter_map(|o| o) .flatten()
.map(|state| state.to_sync_state_event()) .map(|state| state.to_sync_state_event())
.collect(); .collect();
@ -1588,7 +1590,7 @@ pub async fn sync_events_v4_route(
let heroes = services() let heroes = services()
.rooms .rooms
.state_cache .state_cache
.room_members(&room_id) .room_members(room_id)
.filter_map(|r| r.ok()) .filter_map(|r| r.ok())
.filter(|member| member != &sender_user) .filter(|member| member != &sender_user)
.map(|member| { .map(|member| {
@ -1596,7 +1598,7 @@ pub async fn sync_events_v4_route(
services() services()
.rooms .rooms
.state_accessor .state_accessor
.get_member(&room_id, &member)? .get_member(room_id, &member)?
.map(|memberevent| { .map(|memberevent| {
( (
memberevent memberevent
@ -1608,24 +1610,24 @@ pub async fn sync_events_v4_route(
) )
}) })
.filter_map(|r| r.ok()) .filter_map(|r| r.ok())
.filter_map(|o| o) .flatten()
.take(5) .take(5)
.collect::<Vec<_>>(); .collect::<Vec<_>>();
let name = if heroes.len() > 1 { let name = match heroes.len().cmp(&(1_usize)) {
let last = heroes[0].0.clone(); Ordering::Greater => {
Some( let last = heroes[0].0.clone();
heroes[1..] Some(
.iter() heroes[1..]
.map(|h| h.0.clone()) .iter()
.collect::<Vec<_>>() .map(|h| h.0.clone())
.join(", ") .collect::<Vec<_>>()
+ " and " .join(", ")
+ &last, + " and "
) + &last,
} else if heroes.len() == 1 { )
Some(heroes[0].0.clone()) }
} else { Ordering::Equal => Some(heroes[0].0.clone()),
None Ordering::Less => None,
}; };
let avatar = if heroes.len() == 1 { let avatar = if heroes.len() == 1 {
@ -1637,15 +1639,11 @@ pub async fn sync_events_v4_route(
rooms.insert( rooms.insert(
room_id.clone(), room_id.clone(),
sync_events::v4::SlidingSyncRoom { sync_events::v4::SlidingSyncRoom {
name: services() name: services().rooms.state_accessor.get_name(room_id)?.or(name),
.rooms
.state_accessor
.get_name(&room_id)?
.or_else(|| name),
avatar: services() avatar: services()
.rooms .rooms
.state_accessor .state_accessor
.get_avatar(&room_id)? .get_avatar(room_id)?
.map_or(avatar, |a| a.url), .map_or(avatar, |a| a.url),
initial: Some(roomsince == &0), initial: Some(roomsince == &0),
is_dm: None, is_dm: None,
@ -1655,7 +1653,7 @@ pub async fn sync_events_v4_route(
services() services()
.rooms .rooms
.user .user
.highlight_count(&sender_user, &room_id)? .highlight_count(&sender_user, room_id)?
.try_into() .try_into()
.expect("notification count can't go that high"), .expect("notification count can't go that high"),
), ),
@ -1663,7 +1661,7 @@ pub async fn sync_events_v4_route(
services() services()
.rooms .rooms
.user .user
.notification_count(&sender_user, &room_id)? .notification_count(&sender_user, room_id)?
.try_into() .try_into()
.expect("notification count can't go that high"), .expect("notification count can't go that high"),
), ),
@ -1676,7 +1674,7 @@ pub async fn sync_events_v4_route(
(services() (services()
.rooms .rooms
.state_cache .state_cache
.room_joined_count(&room_id)? .room_joined_count(room_id)?
.unwrap_or(0) as u32) .unwrap_or(0) as u32)
.into(), .into(),
), ),
@ -1684,7 +1682,7 @@ pub async fn sync_events_v4_route(
(services() (services()
.rooms .rooms
.state_cache .state_cache
.room_invited_count(&room_id)? .room_invited_count(room_id)?
.unwrap_or(0) as u32) .unwrap_or(0) as u32)
.into(), .into(),
), ),

17
src/api/server_server.rs
View file

@ -1,4 +1,5 @@
#![allow(deprecated)] #![allow(deprecated)]
// Conduit implements the older APIs
use crate::{ use crate::{
api::client_server::{self, claim_keys_helper, get_keys_helper}, api::client_server::{self, claim_keys_helper, get_keys_helper},
@ -520,7 +521,7 @@ async fn query_srv_record(hostname: &'_ str) -> Option<FedDest> {
services() services()
.globals .globals
.dns_resolver() .dns_resolver()
.srv_lookup(format!("{}", hostname)) .srv_lookup(hostname.to_owned())
.await .await
} }
@ -724,7 +725,7 @@ pub fn parse_incoming_pdu(
let room_version_id = services().rooms.state.get_room_version(&room_id)?; let room_version_id = services().rooms.state.get_room_version(&room_id)?;
let (event_id, value) = match gen_event_id_canonical_json(&pdu, &room_version_id) { let (event_id, value) = match gen_event_id_canonical_json(pdu, &room_version_id) {
Ok(t) => t, Ok(t) => t,
Err(_) => { Err(_) => {
// Event could not be converted to canonical json // Event could not be converted to canonical json
@ -783,7 +784,7 @@ pub async fn send_transaction_message_route(
continue; continue;
} }
let r = parse_incoming_pdu(&pdu); let r = parse_incoming_pdu(pdu);
let (event_id, value, room_id) = match r { let (event_id, value, room_id) = match r {
Ok(t) => t, Ok(t) => t,
Err(e) => { Err(e) => {
@ -1088,7 +1089,7 @@ pub async fn get_event_route(
if !services().rooms.state_accessor.server_can_see_event( if !services().rooms.state_accessor.server_can_see_event(
sender_servername, sender_servername,
&room_id, room_id,
&body.event_id, &body.event_id,
)? { )? {
return Err(Error::BadRequest( return Err(Error::BadRequest(
@ -1154,7 +1155,7 @@ pub async fn get_backfill_route(
let all_events = services() let all_events = services()
.rooms .rooms
.timeline .timeline
.pdus_until(&user_id!("@doesntmatter:conduit.rs"), &body.room_id, until)? .pdus_until(user_id!("@doesntmatter:conduit.rs"), &body.room_id, until)?
.take(limit.try_into().unwrap()); .take(limit.try_into().unwrap());
let events = all_events let events = all_events
@ -1171,7 +1172,7 @@ pub async fn get_backfill_route(
}) })
.map(|(_, pdu)| services().rooms.timeline.get_pdu_json(&pdu.event_id)) .map(|(_, pdu)| services().rooms.timeline.get_pdu_json(&pdu.event_id))
.filter_map(|r| r.ok().flatten()) .filter_map(|r| r.ok().flatten())
.map(|pdu| PduEvent::convert_to_outgoing_federation_event(pdu)) .map(PduEvent::convert_to_outgoing_federation_event)
.collect(); .collect();
Ok(get_backfill::v1::Response { Ok(get_backfill::v1::Response {
@ -1917,7 +1918,7 @@ pub async fn get_devices_route(
.all_devices_metadata(&body.user_id) .all_devices_metadata(&body.user_id)
.filter_map(|r| r.ok()) .filter_map(|r| r.ok())
.filter_map(|metadata| { .filter_map(|metadata| {
let device_id_string = metadata.device_id.as_str().to_string(); let device_id_string = metadata.device_id.as_str().to_owned();
let device_display_name = match services().globals.allow_device_name_federation() { let device_display_name = match services().globals.allow_device_name_federation() {
true => metadata.display_name, true => metadata.display_name,
false => Some(device_id_string.to_string()), false => Some(device_id_string.to_string()),
@ -1928,7 +1929,7 @@ pub async fn get_devices_route(
.get_device_keys(&body.user_id, &metadata.device_id) .get_device_keys(&body.user_id, &metadata.device_id)
.ok()??, .ok()??,
device_id: metadata.device_id, device_id: metadata.device_id,
device_display_name: device_display_name, device_display_name,
}) })
}) })
.collect(), .collect(),

9
src/config/proxy.rs
View file

@ -27,10 +27,12 @@ use crate::Result;
/// If a domain matches both the exclude and include list, the proxy will only be used if it was /// If a domain matches both the exclude and include list, the proxy will only be used if it was
/// included because of a more specific rule than it was excluded. In the above example, the proxy /// included because of a more specific rule than it was excluded. In the above example, the proxy
/// would be used for `ordinary.onion`, `matrix.myspecial.onion`, but not `hello.myspecial.onion`. /// would be used for `ordinary.onion`, `matrix.myspecial.onion`, but not `hello.myspecial.onion`.
#[derive(Clone, Debug, Deserialize)] #[derive(Clone, Default, Debug, Deserialize)]
#[serde(rename_all = "snake_case")] #[serde(rename_all = "snake_case")]
pub enum ProxyConfig { pub enum ProxyConfig {
#[default]
None, None,
Global { Global {
#[serde(deserialize_with = "crate::utils::deserialize_from_str")] #[serde(deserialize_with = "crate::utils::deserialize_from_str")]
url: Url, url: Url,
@ -48,11 +50,6 @@ impl ProxyConfig {
}) })
} }
} }
impl Default for ProxyConfig {
fn default() -> Self {
ProxyConfig::None
}
}
#[derive(Clone, Debug, Deserialize)] #[derive(Clone, Debug, Deserialize)]
pub struct PartialProxyConfig { pub struct PartialProxyConfig {

4
src/database/key_value/globals.rs
View file

@ -256,8 +256,8 @@ lasttimelinecount_cache: {lasttimelinecount_cache}\n"
.. ..
} = new_keys; } = new_keys;
keys.verify_keys.extend(verify_keys.into_iter()); keys.verify_keys.extend(verify_keys);
keys.old_verify_keys.extend(old_verify_keys.into_iter()); keys.old_verify_keys.extend(old_verify_keys);
self.server_signingkeys.insert( self.server_signingkeys.insert(
origin.as_bytes(), origin.as_bytes(),

6
src/database/key_value/rooms/state_accessor.rs
View file

@ -20,7 +20,7 @@ impl service::rooms::state_accessor::Data for KeyValueDatabase {
let parsed = services() let parsed = services()
.rooms .rooms
.state_compressor .state_compressor
.parse_compressed_state_event(&compressed)?; .parse_compressed_state_event(compressed)?;
result.insert(parsed.0, parsed.1); result.insert(parsed.0, parsed.1);
i += 1; i += 1;
@ -49,7 +49,7 @@ impl service::rooms::state_accessor::Data for KeyValueDatabase {
let (_, eventid) = services() let (_, eventid) = services()
.rooms .rooms
.state_compressor .state_compressor
.parse_compressed_state_event(&compressed)?; .parse_compressed_state_event(compressed)?;
if let Some(pdu) = services().rooms.timeline.get_pdu(&eventid)? { if let Some(pdu) = services().rooms.timeline.get_pdu(&eventid)? {
result.insert( result.insert(
( (
@ -101,7 +101,7 @@ impl service::rooms::state_accessor::Data for KeyValueDatabase {
services() services()
.rooms .rooms
.state_compressor .state_compressor
.parse_compressed_state_event(&compressed) .parse_compressed_state_event(compressed)
.ok() .ok()
.map(|(_, id)| id) .map(|(_, id)| id)
})) }))

4
src/database/key_value/rooms/threads.rs
View file

@ -52,13 +52,13 @@ impl service::rooms::threads::Data for KeyValueDatabase {
.collect::<Vec<_>>() .collect::<Vec<_>>()
.join(&[0xff][..]); .join(&[0xff][..]);
self.threadid_userids.insert(&root_id, &users)?; self.threadid_userids.insert(root_id, &users)?;
Ok(()) Ok(())
} }
fn get_participants(&self, root_id: &[u8]) -> Result<Option<Vec<OwnedUserId>>> { fn get_participants(&self, root_id: &[u8]) -> Result<Option<Vec<OwnedUserId>>> {
if let Some(users) = self.threadid_userids.get(&root_id)? { if let Some(users) = self.threadid_userids.get(root_id)? {
Ok(Some( Ok(Some(
users users
.split(|b| *b == 0xff) .split(|b| *b == 0xff)

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

@ -39,11 +39,10 @@ impl service::rooms::timeline::Data for KeyValueDatabase {
/// Returns the `count` of this pdu's id. /// Returns the `count` of this pdu's id.
fn get_pdu_count(&self, event_id: &EventId) -> Result<Option<PduCount>> { fn get_pdu_count(&self, event_id: &EventId) -> Result<Option<PduCount>> {
Ok(self self.eventid_pduid
.eventid_pduid
.get(event_id.as_bytes())? .get(event_id.as_bytes())?
.map(|pdu_id| pdu_count(&pdu_id)) .map(|pdu_id| pdu_count(&pdu_id))
.transpose()?) .transpose()
} }
/// Returns the json of a pdu. /// Returns the json of a pdu.
@ -80,7 +79,7 @@ impl service::rooms::timeline::Data for KeyValueDatabase {
/// Returns the pdu's id. /// Returns the pdu's id.
fn get_pdu_id(&self, event_id: &EventId) -> Result<Option<Vec<u8>>> { fn get_pdu_id(&self, event_id: &EventId) -> Result<Option<Vec<u8>>> {
Ok(self.eventid_pduid.get(event_id.as_bytes())?) self.eventid_pduid.get(event_id.as_bytes())
} }
/// Returns the pdu. /// Returns the pdu.
@ -230,7 +229,7 @@ impl service::rooms::timeline::Data for KeyValueDatabase {
room_id: &RoomId, room_id: &RoomId,
until: PduCount, until: PduCount,
) -> Result<Box<dyn Iterator<Item = Result<(PduCount, PduEvent)>> + 'a>> { ) -> Result<Box<dyn Iterator<Item = Result<(PduCount, PduEvent)>> + 'a>> {
let (prefix, current) = count_to_id(&room_id, until, 1, true)?; let (prefix, current) = count_to_id(room_id, until, 1, true)?;
let user_id = user_id.to_owned(); let user_id = user_id.to_owned();
@ -257,7 +256,7 @@ impl service::rooms::timeline::Data for KeyValueDatabase {
room_id: &RoomId, room_id: &RoomId,
from: PduCount, from: PduCount,
) -> Result<Box<dyn Iterator<Item = Result<(PduCount, PduEvent)>> + 'a>> { ) -> Result<Box<dyn Iterator<Item = Result<(PduCount, PduEvent)>> + 'a>> {
let (prefix, current) = count_to_id(&room_id, from, 1, false)?; let (prefix, current) = count_to_id(room_id, from, 1, false)?;
let user_id = user_id.to_owned(); let user_id = user_id.to_owned();

11
src/database/mod.rs
View file

@ -427,14 +427,12 @@ impl KeyValueDatabase {
} }
// If the database has any data, perform data migrations before starting // If the database has any data, perform data migrations before starting
let latest_database_version: u64;
// do not increment the db version if the user is not using sha256_media // do not increment the db version if the user is not using sha256_media
if cfg!(feature = "sha256_media") { let latest_database_version = if cfg!(feature = "sha256_media") {
latest_database_version = 14; 14
} else { } else {
latest_database_version = 13; 13
} };
if services().users.count()? > 0 { if services().users.count()? > 0 {
// MIGRATIONS // MIGRATIONS
@ -487,6 +485,7 @@ impl KeyValueDatabase {
continue; continue;
} }
#[allow(deprecated)]
let path = services().globals.get_media_file(&key); let path = services().globals.get_media_file(&key);
let mut file = fs::File::create(path)?; let mut file = fs::File::create(path)?;
file.write_all(&content)?; file.write_all(&content)?;

1
src/lib.rs
View file

@ -4,7 +4,6 @@
clippy::cloned_instead_of_copied, clippy::cloned_instead_of_copied,
clippy::str_to_string clippy::str_to_string
)] )]
#![allow(clippy::suspicious_else_formatting)]
#![deny(clippy::dbg_macro)] #![deny(clippy::dbg_macro)]
pub mod api; pub mod api;

19
src/main.rs
View file

@ -5,7 +5,6 @@
clippy::str_to_string, clippy::str_to_string,
clippy::future_not_send clippy::future_not_send
)] )]
#![allow(clippy::suspicious_else_formatting)]
#![deny(clippy::dbg_macro)] #![deny(clippy::dbg_macro)]
use std::{ use std::{
@ -17,7 +16,7 @@ use axum::{
extract::{DefaultBodyLimit, FromRequestParts, MatchedPath}, extract::{DefaultBodyLimit, FromRequestParts, MatchedPath},
response::IntoResponse, response::IntoResponse,
routing::{get, on, MethodFilter}, routing::{get, on, MethodFilter},
Json, Router, Router,
}; };
use axum_server::{bind, bind_rustls, tls_rustls::RustlsConfig, Handle as ServerHandle}; use axum_server::{bind, bind_rustls, tls_rustls::RustlsConfig, Handle as ServerHandle};
use conduit::api::{client_server, server_server}; use conduit::api::{client_server, server_server};
@ -38,7 +37,6 @@ use ruma::api::{
}, },
IncomingRequest, IncomingRequest,
}; };
use serde::Deserialize;
use tokio::{net::UnixListener, signal, sync::oneshot}; use tokio::{net::UnixListener, signal, sync::oneshot};
use tower::ServiceBuilder; use tower::ServiceBuilder;
use tower_http::{ use tower_http::{
@ -147,7 +145,7 @@ async fn main() {
maximize_fd_limit().expect("should be able to increase the soft limit to the hard limit"); maximize_fd_limit().expect("should be able to increase the soft limit to the hard limit");
config.warn_deprecated(); config.warn_deprecated();
if let Err(_) = config.error_dual_listening(raw_config) { if config.error_dual_listening(raw_config).is_err() {
return; return;
}; };
@ -213,16 +211,15 @@ async fn run_server() -> io::Result<()> {
.expect("failed to convert max request size"), .expect("failed to convert max request size"),
)); ));
let app: axum::routing::IntoMakeService<Router>; let app = if cfg!(feature = "zstd_compression") && config.zstd_compression {
if cfg!(feature = "zstd_compression") && config.zstd_compression {
debug!("zstd body compression is enabled"); debug!("zstd body compression is enabled");
app = routes() routes()
.layer(middlewares.compression()) .layer(middlewares.compression())
.into_make_service(); .into_make_service()
} else { } else {
app = routes().layer(middlewares).into_make_service(); routes().layer(middlewares).into_make_service()
} };
let handle = ServerHandle::new(); let handle = ServerHandle::new();
let (tx, rx) = oneshot::channel::<()>(); let (tx, rx) = oneshot::channel::<()>();

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

@ -5,6 +5,8 @@ use std::{
time::Instant, time::Instant,
}; };
use std::fmt::Write;
use clap::{Parser, Subcommand}; use clap::{Parser, Subcommand};
use regex::Regex; use regex::Regex;
use ruma::{ use ruma::{
@ -27,7 +29,7 @@ use ruma::{
EventId, OwnedRoomAliasId, OwnedRoomId, RoomAliasId, RoomId, RoomVersionId, ServerName, UserId, EventId, OwnedRoomAliasId, OwnedRoomId, 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},
@ -194,19 +196,19 @@ enum RoomAliasCommand {
room_id: Box<RoomId>, room_id: Box<RoomId>,
/// The alias localpart to use (`alias`, not `#alias:servername.tld`) /// The alias localpart to use (`alias`, not `#alias:servername.tld`)
room_alias_localpart: Box<String>, room_alias_localpart: String,
}, },
/// Remove an alias /// Remove an alias
Remove { Remove {
/// The alias localpart to remove (`alias`, not `#alias:servername.tld`) /// The alias localpart to remove (`alias`, not `#alias:servername.tld`)
room_alias_localpart: Box<String>, room_alias_localpart: String,
}, },
/// Show which room is using an alias /// Show which room is using an alias
Which { Which {
/// The alias localpart to look up (`alias`, not `#alias:servername.tld`) /// The alias localpart to look up (`alias`, not `#alias:servername.tld`)
room_alias_localpart: Box<String>, room_alias_localpart: String,
}, },
/// List aliases currently being used /// List aliases currently being used
@ -802,13 +804,12 @@ impl Service {
"<table><caption>Room list - page {page}</caption>\n<tr><th>id</th>\t<th>members</th>\t<th>name</th></tr>\n{}</table>", "<table><caption>Room list - page {page}</caption>\n<tr><th>id</th>\t<th>members</th>\t<th>name</th></tr>\n{}</table>",
rooms rooms
.iter() .iter()
.map(|(id, members, name)| format!( .fold(String::new(), |mut output, (id, members, name)| {
"<tr><td>{}</td>\t<td>{}</td>\t<td>{}</td></tr>\n", writeln!(output, "<tr><td>{}</td>\t<td>{}</td>\t<td>{}</td></tr>", escape_html(id.as_ref()),
escape_html(&id.to_string()),
members, members,
escape_html(name), escape_html(name)).unwrap();
)) output
.collect::<String>() })
); );
RoomMessageEventContent::text_html(output_plain, output_html) RoomMessageEventContent::text_html(output_plain, output_html)
} }
@ -886,7 +887,7 @@ impl Service {
Ok(None) => { Ok(None) => {
RoomMessageEventContent::text_plain("Alias isn't in use.") RoomMessageEventContent::text_plain("Alias isn't in use.")
} }
Err(err) => RoomMessageEventContent::text_plain(&format!( Err(err) => RoomMessageEventContent::text_plain(format!(
"Unable to lookup alias: {}", "Unable to lookup alias: {}",
err err
)), )),
@ -904,27 +905,29 @@ impl Service {
.collect(); .collect();
match aliases { match aliases {
Ok(aliases) => { Ok(aliases) => {
let plain_list: String = aliases let plain_list: String =
.iter() aliases.iter().fold(String::new(), |mut output, alias| {
.map(|alias| format!("- {}\n", alias)) writeln!(output, "- {}", alias).unwrap();
.collect(); output
});
let html_list: String = aliases let html_list: String =
.iter() aliases.iter().fold(String::new(), |mut output, alias| {
.map(|alias| { writeln!(
format!( output,
"<li>{}</li>\n", "<li>{}</li>",
escape_html(&alias.to_string()) escape_html(alias.as_ref())
) )
}) .unwrap();
.collect(); output
});
let plain = format!("Aliases for {}:\n{}", room_id, plain_list); let plain = format!("Aliases for {}:\n{}", room_id, plain_list);
let html = let html =
format!("Aliases for {}:\n<ul>{}</ul>", room_id, html_list); format!("Aliases for {}:\n<ul>{}</ul>", room_id, html_list);
RoomMessageEventContent::text_html(plain, html) RoomMessageEventContent::text_html(plain, html)
} }
Err(err) => RoomMessageEventContent::text_plain(&format!( Err(err) => RoomMessageEventContent::text_plain(format!(
"Unable to list aliases: {}", "Unable to list aliases: {}",
err err
)), )),
@ -936,30 +939,39 @@ impl Service {
match aliases { match aliases {
Ok(aliases) => { Ok(aliases) => {
let server_name = services().globals.server_name(); let server_name = services().globals.server_name();
let plain_list: String = aliases let plain_list: String = aliases.iter().fold(
.iter() String::new(),
.map(|(id, alias)| { |mut output, (alias, id)| {
format!("- #{}:{} -> {}\n", alias, server_name, id) writeln!(
}) output,
.collect(); "- #{}:{} -> {}",
alias, server_name, id
let html_list: String = aliases
.iter()
.map(|(id, alias)| {
format!(
"<li>#{}:{} -> {}</li>\n",
escape_html(&alias.to_string()),
server_name,
escape_html(&id.to_string())
) )
}) .unwrap();
.collect(); output
},
);
let html_list: String = aliases.iter().fold(
String::new(),
|mut output, (alias, id)| {
writeln!(
output,
"<li>#{}:{} -> {}</li>",
escape_html(alias.as_ref()),
server_name,
escape_html(id.as_ref())
)
.unwrap();
output
},
);
let plain = format!("Aliases:\n{}", plain_list); let plain = format!("Aliases:\n{}", plain_list);
let html = format!("Aliases:\n<ul>{}</ul>", html_list); let html = format!("Aliases:\n<ul>{}</ul>", html_list);
RoomMessageEventContent::text_html(plain, html) RoomMessageEventContent::text_html(plain, html)
} }
Err(err) => RoomMessageEventContent::text_plain(&format!( Err(err) => RoomMessageEventContent::text_plain(format!(
"Unable to list aliases: {}", "Unable to list aliases: {}",
err err
)), )),
@ -971,7 +983,7 @@ impl Service {
RoomDirectoryCommand::Publish { room_id } => { RoomDirectoryCommand::Publish { room_id } => {
match services().rooms.directory.set_public(&room_id) { match services().rooms.directory.set_public(&room_id) {
Ok(()) => RoomMessageEventContent::text_plain("Room published"), Ok(()) => RoomMessageEventContent::text_plain("Room published"),
Err(err) => RoomMessageEventContent::text_plain(&format!( Err(err) => RoomMessageEventContent::text_plain(format!(
"Unable to update room: {}", "Unable to update room: {}",
err err
)), )),
@ -980,7 +992,7 @@ impl Service {
RoomDirectoryCommand::Unpublish { room_id } => { RoomDirectoryCommand::Unpublish { room_id } => {
match services().rooms.directory.set_not_public(&room_id) { match services().rooms.directory.set_not_public(&room_id) {
Ok(()) => RoomMessageEventContent::text_plain("Room unpublished"), Ok(()) => RoomMessageEventContent::text_plain("Room unpublished"),
Err(err) => RoomMessageEventContent::text_plain(&format!( Err(err) => RoomMessageEventContent::text_plain(format!(
"Unable to update room: {}", "Unable to update room: {}",
err err
)), )),
@ -1023,13 +1035,10 @@ impl Service {
"<table><caption>Room directory - page {page}</caption>\n<tr><th>id</th>\t<th>members</th>\t<th>name</th></tr>\n{}</table>", "<table><caption>Room directory - page {page}</caption>\n<tr><th>id</th>\t<th>members</th>\t<th>name</th></tr>\n{}</table>",
rooms rooms
.iter() .iter()
.map(|(id, members, name)| format!( .fold(String::new(), |mut output, (id, members, name)| {
"<tr><td>{}</td>\t<td>{}</td>\t<td>{}</td></tr>\n", writeln!(output, "<tr><td>{}</td>\t<td>{}</td>\t<td>{}</td></tr>", escape_html(id.as_ref()), members, escape_html(name.as_ref())).unwrap();
escape_html(&id.to_string()), output
members, })
escape_html(name),
))
.collect::<String>()
); );
RoomMessageEventContent::text_html(output_plain, output_html) RoomMessageEventContent::text_html(output_plain, output_html)
} }

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

@ -164,7 +164,7 @@ impl Service {
.dns_resolver(Arc::new(Resolver::new(Box::new(move |domain| { .dns_resolver(Arc::new(Resolver::new(Box::new(move |domain| {
let read_guard = name_override.read().unwrap(); let read_guard = name_override.read().unwrap();
let (override_name, port) = read_guard.get(domain)?; let (override_name, port) = read_guard.get(domain)?;
let first_name = override_name.get(0)?; let first_name = override_name.first()?;
Some(SocketAddr::new(*first_name, *port)) Some(SocketAddr::new(*first_name, *port))
})))) }))))
.build()?; .build()?;

69
src/service/media/mod.rs
View file

@ -35,19 +35,19 @@ impl Service {
.db .db
.create_file_metadata(mxc, 0, 0, content_disposition, content_type)?; .create_file_metadata(mxc, 0, 0, content_disposition, content_type)?;
let path: std::path::PathBuf; let path = if cfg!(feature = "sha256_media") {
if cfg!(feature = "sha256_media") { services().globals.get_media_file_new(&key)
path = services().globals.get_media_file_new(&key);
} else { } else {
path = services().globals.get_media_file(&key); #[allow(deprecated)]
} services().globals.get_media_file(&key)
};
let mut f = File::create(path).await?; let mut f = File::create(path).await?;
f.write_all(file).await?; f.write_all(file).await?;
Ok(()) Ok(())
} }
/// Uploads or replaces a file thumbnail. /// Uploads or replaces a file thumbnail.
#[allow(clippy::too_many_arguments)]
pub async fn upload_thumbnail( pub async fn upload_thumbnail(
&self, &self,
mxc: String, mxc: String,
@ -61,12 +61,13 @@ impl Service {
self.db self.db
.create_file_metadata(mxc, width, height, content_disposition, content_type)?; .create_file_metadata(mxc, width, height, content_disposition, content_type)?;
let path: std::path::PathBuf; let path = if cfg!(feature = "sha256_media") {
if cfg!(feature = "sha256_media") { services().globals.get_media_file_new(&key)
path = services().globals.get_media_file_new(&key);
} else { } else {
path = services().globals.get_media_file(&key); #[allow(deprecated)]
} services().globals.get_media_file(&key)
};
let mut f = File::create(path).await?; let mut f = File::create(path).await?;
f.write_all(file).await?; f.write_all(file).await?;
@ -78,12 +79,13 @@ impl Service {
if let Ok((content_disposition, content_type, key)) = if let Ok((content_disposition, content_type, key)) =
self.db.search_file_metadata(mxc, 0, 0) self.db.search_file_metadata(mxc, 0, 0)
{ {
let path: std::path::PathBuf; let path = if cfg!(feature = "sha256_media") {
if cfg!(feature = "sha256_media") { services().globals.get_media_file_new(&key)
path = services().globals.get_media_file_new(&key);
} else { } else {
path = services().globals.get_media_file(&key); #[allow(deprecated)]
} services().globals.get_media_file(&key)
};
let mut file = Vec::new(); let mut file = Vec::new();
BufReader::new(File::open(path).await?) BufReader::new(File::open(path).await?)
.read_to_end(&mut file) .read_to_end(&mut file)
@ -136,12 +138,13 @@ impl Service {
self.db.search_file_metadata(mxc.clone(), width, height) self.db.search_file_metadata(mxc.clone(), width, height)
{ {
// Using saved thumbnail // Using saved thumbnail
let path: std::path::PathBuf; let path = if cfg!(feature = "sha256_media") {
if cfg!(feature = "sha256_media") { services().globals.get_media_file_new(&key)
path = services().globals.get_media_file_new(&key);
} else { } else {
path = services().globals.get_media_file(&key); #[allow(deprecated)]
} services().globals.get_media_file(&key)
};
let mut file = Vec::new(); let mut file = Vec::new();
File::open(path).await?.read_to_end(&mut file).await?; File::open(path).await?.read_to_end(&mut file).await?;
@ -154,12 +157,13 @@ impl Service {
self.db.search_file_metadata(mxc.clone(), 0, 0) self.db.search_file_metadata(mxc.clone(), 0, 0)
{ {
// Generate a thumbnail // Generate a thumbnail
let path: std::path::PathBuf; let path = if cfg!(feature = "sha256_media") {
if cfg!(feature = "sha256_media") { services().globals.get_media_file_new(&key)
path = services().globals.get_media_file_new(&key);
} else { } else {
path = services().globals.get_media_file(&key); #[allow(deprecated)]
} services().globals.get_media_file(&key)
};
let mut file = Vec::new(); let mut file = Vec::new();
File::open(path).await?.read_to_end(&mut file).await?; File::open(path).await?.read_to_end(&mut file).await?;
@ -229,12 +233,13 @@ impl Service {
content_type.as_deref(), content_type.as_deref(),
)?; )?;
let path: std::path::PathBuf; let path = if cfg!(feature = "sha256_media") {
if cfg!(feature = "sha256_media") { services().globals.get_media_file_new(&thumbnail_key)
path = services().globals.get_media_file_new(&thumbnail_key);
} else { } else {
path = services().globals.get_media_file(&thumbnail_key); #[allow(deprecated)]
} services().globals.get_media_file(&thumbnail_key)
};
let mut f = File::create(path).await?; let mut f = File::create(path).await?;
f.write_all(&thumbnail_bytes).await?; f.write_all(&thumbnail_bytes).await?;
@ -337,7 +342,7 @@ mod tests {
// r.push(base64::encode_config(key, base64::URL_SAFE_NO_PAD)); // r.push(base64::encode_config(key, base64::URL_SAFE_NO_PAD));
// use the sha256 hash of the key as the file name instead of the key itself // use the sha256 hash of the key as the file name instead of the key itself
// this is because the base64 encoded key can be longer than 255 characters. // this is because the base64 encoded key can be longer than 255 characters.
r.push(general_purpose::URL_SAFE_NO_PAD.encode(sha2::Sha256::digest(&key))); r.push(general_purpose::URL_SAFE_NO_PAD.encode(sha2::Sha256::digest(key)));
// Check that the file path is not longer than 255 characters // Check that the file path is not longer than 255 characters
// (255 is the maximum length of a file path on most file systems) // (255 is the maximum length of a file path on most file systems)
assert!( assert!(

3
src/service/pdu.rs
View file

@ -383,10 +383,9 @@ impl PartialEq for PduEvent {
self.event_id == other.event_id self.event_id == other.event_id
} }
} }
#[allow(clippy::non_canonical_partial_ord_impl)]
impl PartialOrd for PduEvent { impl PartialOrd for PduEvent {
fn partial_cmp(&self, other: &Self) -> Option<Ordering> { fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
self.event_id.partial_cmp(&other.event_id) Some(self.cmp(other))
} }
} }
impl Ord for PduEvent { impl Ord for PduEvent {

2
src/service/rooms/edus/presence/mod.rs
View file

@ -182,7 +182,7 @@ fn process_presence_timer(user_id: OwnedUserId) -> Result<()> {
} }
} }
let new_state = match (&presence_state, last_active_ago.map(|ago| u64::from(ago))) { let new_state = match (&presence_state, last_active_ago.map(u64::from)) {
(PresenceState::Online, Some(ago)) if ago >= idle_timeout => { (PresenceState::Online, Some(ago)) if ago >= idle_timeout => {
Some(PresenceState::Unavailable) Some(PresenceState::Unavailable)
} }

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

@ -92,7 +92,7 @@ impl Service {
)); ));
} }
services().rooms.event_handler.acl_check(origin, &room_id)?; services().rooms.event_handler.acl_check(origin, room_id)?;
// 1. Skip the PDU if we already have it as a timeline event // 1. Skip the PDU if we already have it as a timeline event
if let Some(pdu_id) = services().rooms.timeline.get_pdu_id(event_id)? { if let Some(pdu_id) = services().rooms.timeline.get_pdu_id(event_id)? {
@ -277,6 +277,7 @@ impl Service {
r r
} }
#[allow(clippy::too_many_arguments)]
fn handle_outlier_pdu<'a>( fn handle_outlier_pdu<'a>(
&'a self, &'a self,
origin: &'a ServerName, origin: &'a ServerName,

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

@ -40,6 +40,7 @@ impl Service {
} }
} }
#[allow(clippy::too_many_arguments)]
pub fn paginate_relations_with_filter( pub fn paginate_relations_with_filter(
&self, &self,
sender_user: &UserId, sender_user: &UserId,
@ -82,7 +83,7 @@ impl Service {
services() services()
.rooms .rooms
.state_accessor .state_accessor
.user_can_see_event(sender_user, &room_id, &pdu.event_id) .user_can_see_event(sender_user, room_id, &pdu.event_id)
.unwrap_or(false) .unwrap_or(false)
}) })
.take_while(|&(k, _)| Some(k) != to) // Stop at `to` .take_while(|&(k, _)| Some(k) != to) // Stop at `to`
@ -106,7 +107,7 @@ impl Service {
let events_before: Vec<_> = services() let events_before: Vec<_> = services()
.rooms .rooms
.pdu_metadata .pdu_metadata
.relations_until(sender_user, &room_id, target, from)? .relations_until(sender_user, room_id, target, from)?
.filter(|r| { .filter(|r| {
r.as_ref().map_or(true, |(_, pdu)| { r.as_ref().map_or(true, |(_, pdu)| {
filter_event_type.as_ref().map_or(true, |t| &pdu.kind == t) filter_event_type.as_ref().map_or(true, |t| &pdu.kind == t)
@ -129,7 +130,7 @@ impl Service {
services() services()
.rooms .rooms
.state_accessor .state_accessor
.user_can_see_event(sender_user, &room_id, &pdu.event_id) .user_can_see_event(sender_user, room_id, &pdu.event_id)
.unwrap_or(false) .unwrap_or(false)
}) })
.take_while(|&(k, _)| Some(k) != to) // Stop at `to` .take_while(|&(k, _)| Some(k) != to) // Stop at `to`

45
src/service/rooms/spaces/mod.rs
View file

@ -134,7 +134,7 @@ impl Service {
if serde_json::from_str::<SpaceChildEventContent>(pdu.content.get()) if serde_json::from_str::<SpaceChildEventContent>(pdu.content.get())
.ok() .ok()
.and_then(|c| Some(c.via)) .map(|c| c.via)
.map_or(true, |v| v.is_empty()) .map_or(true, |v| v.is_empty())
{ {
continue; continue;
@ -199,7 +199,7 @@ impl Service {
if let Ok(response) = services() if let Ok(response) = services()
.sending .sending
.send_federation_request( .send_federation_request(
&server, server,
federation::space::get_hierarchy::v1::Request { federation::space::get_hierarchy::v1::Request {
room_id: current_room.to_owned(), room_id: current_room.to_owned(),
suggested_only, suggested_only,
@ -237,7 +237,7 @@ impl Service {
.room .room
.allowed_room_ids .allowed_room_ids
.into_iter() .into_iter()
.map(|room| AllowRule::room_membership(room)) .map(AllowRule::room_membership)
.collect(), .collect(),
}) })
} }
@ -247,7 +247,7 @@ impl Service {
.room .room
.allowed_room_ids .allowed_room_ids
.into_iter() .into_iter()
.map(|room| AllowRule::room_membership(room)) .map(AllowRule::room_membership)
.collect(), .collect(),
}) })
} }
@ -315,7 +315,7 @@ impl Service {
canonical_alias: services() canonical_alias: services()
.rooms .rooms
.state_accessor .state_accessor
.room_state_get(&room_id, &StateEventType::RoomCanonicalAlias, "")? .room_state_get(room_id, &StateEventType::RoomCanonicalAlias, "")?
.map_or(Ok(None), |s| { .map_or(Ok(None), |s| {
serde_json::from_str(s.content.get()) serde_json::from_str(s.content.get())
.map(|c: RoomCanonicalAliasEventContent| c.alias) .map(|c: RoomCanonicalAliasEventContent| c.alias)
@ -323,11 +323,11 @@ impl Service {
Error::bad_database("Invalid canonical alias event in database.") Error::bad_database("Invalid canonical alias event in database.")
}) })
})?, })?,
name: services().rooms.state_accessor.get_name(&room_id)?, name: services().rooms.state_accessor.get_name(room_id)?,
num_joined_members: services() num_joined_members: services()
.rooms .rooms
.state_cache .state_cache
.room_joined_count(&room_id)? .room_joined_count(room_id)?
.unwrap_or_else(|| { .unwrap_or_else(|| {
warn!("Room {} has no member count", room_id); warn!("Room {} has no member count", room_id);
0 0
@ -338,7 +338,7 @@ impl Service {
topic: services() topic: services()
.rooms .rooms
.state_accessor .state_accessor
.room_state_get(&room_id, &StateEventType::RoomTopic, "")? .room_state_get(room_id, &StateEventType::RoomTopic, "")?
.map_or(Ok(None), |s| { .map_or(Ok(None), |s| {
serde_json::from_str(s.content.get()) serde_json::from_str(s.content.get())
.map(|c: RoomTopicEventContent| Some(c.topic)) .map(|c: RoomTopicEventContent| Some(c.topic))
@ -350,7 +350,7 @@ impl Service {
world_readable: services() world_readable: services()
.rooms .rooms
.state_accessor .state_accessor
.room_state_get(&room_id, &StateEventType::RoomHistoryVisibility, "")? .room_state_get(room_id, &StateEventType::RoomHistoryVisibility, "")?
.map_or(Ok(false), |s| { .map_or(Ok(false), |s| {
serde_json::from_str(s.content.get()) serde_json::from_str(s.content.get())
.map(|c: RoomHistoryVisibilityEventContent| { .map(|c: RoomHistoryVisibilityEventContent| {
@ -365,7 +365,7 @@ impl Service {
guest_can_join: services() guest_can_join: services()
.rooms .rooms
.state_accessor .state_accessor
.room_state_get(&room_id, &StateEventType::RoomGuestAccess, "")? .room_state_get(room_id, &StateEventType::RoomGuestAccess, "")?
.map_or(Ok(false), |s| { .map_or(Ok(false), |s| {
serde_json::from_str(s.content.get()) serde_json::from_str(s.content.get())
.map(|c: RoomGuestAccessEventContent| { .map(|c: RoomGuestAccessEventContent| {
@ -378,7 +378,7 @@ impl Service {
avatar_url: services() avatar_url: services()
.rooms .rooms
.state_accessor .state_accessor
.room_state_get(&room_id, &StateEventType::RoomAvatar, "")? .room_state_get(room_id, &StateEventType::RoomAvatar, "")?
.map(|s| { .map(|s| {
serde_json::from_str(s.content.get()) serde_json::from_str(s.content.get())
.map(|c: RoomAvatarEventContent| c.url) .map(|c: RoomAvatarEventContent| c.url)
@ -391,7 +391,7 @@ impl Service {
let join_rule = services() let join_rule = services()
.rooms .rooms
.state_accessor .state_accessor
.room_state_get(&room_id, &StateEventType::RoomJoinRules, "")? .room_state_get(room_id, &StateEventType::RoomJoinRules, "")?
.map(|s| { .map(|s| {
serde_json::from_str(s.content.get()) serde_json::from_str(s.content.get())
.map(|c: RoomJoinRulesEventContent| c.join_rule) .map(|c: RoomJoinRulesEventContent| c.join_rule)
@ -417,7 +417,7 @@ impl Service {
room_type: services() room_type: services()
.rooms .rooms
.state_accessor .state_accessor
.room_state_get(&room_id, &StateEventType::RoomCreate, "")? .room_state_get(room_id, &StateEventType::RoomCreate, "")?
.map(|s| { .map(|s| {
serde_json::from_str::<RoomCreateEventContent>(s.content.get()).map_err(|e| { serde_json::from_str::<RoomCreateEventContent>(s.content.get()).map_err(|e| {
error!("Invalid room create event in database: {}", e); error!("Invalid room create event in database: {}", e);
@ -457,7 +457,7 @@ impl Service {
SpaceRoomJoinRule::Invite => services() SpaceRoomJoinRule::Invite => services()
.rooms .rooms
.state_cache .state_cache
.is_joined(sender_user, &room_id)?, .is_joined(sender_user, room_id)?,
_ => false, _ => false,
}; };
@ -481,17 +481,14 @@ impl Service {
match join_rule { match join_rule {
JoinRule::Restricted(r) => { JoinRule::Restricted(r) => {
for rule in &r.allow { for rule in &r.allow {
match rule { if let join_rules::AllowRule::RoomMembership(rm) = rule {
join_rules::AllowRule::RoomMembership(rm) => { if let Ok(true) = services()
if let Ok(true) = services() .rooms
.rooms .state_cache
.state_cache .is_joined(sender_user, &rm.room_id)
.is_joined(sender_user, &rm.room_id) {
{ return Ok(true);
return Ok(true);
}
} }
_ => {}
} }
} }

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

@ -41,7 +41,7 @@ impl Service {
services() services()
.rooms .rooms
.state_compressor .state_compressor
.parse_compressed_state_event(&new) .parse_compressed_state_event(new)
.ok() .ok()
.map(|(_, id)| id) .map(|(_, id)| id)
}) { }) {
@ -409,7 +409,7 @@ impl Service {
services() services()
.rooms .rooms
.state_compressor .state_compressor
.parse_compressed_state_event(&compressed) .parse_compressed_state_event(compressed)
.ok() .ok()
}) })
.filter_map(|(shortstatekey, event_id)| { .filter_map(|(shortstatekey, event_id)| {

16
src/service/rooms/state_accessor/mod.rs
View file

@ -180,7 +180,7 @@ impl Service {
return Ok(*visibility); return Ok(*visibility);
} }
let currently_member = services().rooms.state_cache.is_joined(&user_id, &room_id)?; let currently_member = services().rooms.state_cache.is_joined(user_id, room_id)?;
let history_visibility = self let history_visibility = self
.state_get(shortstatehash, &StateEventType::RoomHistoryVisibility, "")? .state_get(shortstatehash, &StateEventType::RoomHistoryVisibility, "")?
@ -197,11 +197,11 @@ impl Service {
HistoryVisibility::Shared => currently_member, HistoryVisibility::Shared => currently_member,
HistoryVisibility::Invited => { HistoryVisibility::Invited => {
// Allow if any member on requesting server was AT LEAST invited, else deny // Allow if any member on requesting server was AT LEAST invited, else deny
self.user_was_invited(shortstatehash, &user_id) self.user_was_invited(shortstatehash, user_id)
} }
HistoryVisibility::Joined => { HistoryVisibility::Joined => {
// Allow if any member on requested server was joined, else deny // Allow if any member on requested server was joined, else deny
self.user_was_joined(shortstatehash, &user_id) self.user_was_joined(shortstatehash, user_id)
} }
_ => { _ => {
error!("Unknown history visibility {history_visibility}"); error!("Unknown history visibility {history_visibility}");
@ -221,10 +221,10 @@ impl Service {
/// the room's history_visibility at that event's state. /// the room's history_visibility at that event's state.
#[tracing::instrument(skip(self, user_id, room_id))] #[tracing::instrument(skip(self, user_id, room_id))]
pub fn user_can_see_state_events(&self, user_id: &UserId, room_id: &RoomId) -> Result<bool> { pub fn user_can_see_state_events(&self, user_id: &UserId, room_id: &RoomId) -> Result<bool> {
let currently_member = services().rooms.state_cache.is_joined(&user_id, &room_id)?; let currently_member = services().rooms.state_cache.is_joined(user_id, room_id)?;
let history_visibility = self let history_visibility = self
.room_state_get(&room_id, &StateEventType::RoomHistoryVisibility, "")? .room_state_get(room_id, &StateEventType::RoomHistoryVisibility, "")?
.map_or(Ok(HistoryVisibility::Shared), |s| { .map_or(Ok(HistoryVisibility::Shared), |s| {
serde_json::from_str(s.content.get()) serde_json::from_str(s.content.get())
.map(|c: RoomHistoryVisibilityEventContent| c.history_visibility) .map(|c: RoomHistoryVisibilityEventContent| c.history_visibility)
@ -276,7 +276,7 @@ impl Service {
services() services()
.rooms .rooms
.state_accessor .state_accessor
.room_state_get(&room_id, &StateEventType::RoomName, "")? .room_state_get(room_id, &StateEventType::RoomName, "")?
.map_or(Ok(None), |s| { .map_or(Ok(None), |s| {
serde_json::from_str(s.content.get()) serde_json::from_str(s.content.get())
.map(|c: RoomNameEventContent| Some(c.name)) .map(|c: RoomNameEventContent| Some(c.name))
@ -294,7 +294,7 @@ impl Service {
services() services()
.rooms .rooms
.state_accessor .state_accessor
.room_state_get(&room_id, &StateEventType::RoomAvatar, "")? .room_state_get(room_id, &StateEventType::RoomAvatar, "")?
.map_or(Ok(None), |s| { .map_or(Ok(None), |s| {
serde_json::from_str(s.content.get()) serde_json::from_str(s.content.get())
.map_err(|_| Error::bad_database("Invalid room avatar event in database.")) .map_err(|_| Error::bad_database("Invalid room avatar event in database."))
@ -309,7 +309,7 @@ impl Service {
services() services()
.rooms .rooms
.state_accessor .state_accessor
.room_state_get(&room_id, &StateEventType::RoomMember, user_id.as_str())? .room_state_get(room_id, &StateEventType::RoomMember, user_id.as_str())?
.map_or(Ok(None), |s| { .map_or(Ok(None), |s| {
serde_json::from_str(s.content.get()) serde_json::from_str(s.content.get())
.map_err(|_| Error::bad_database("Invalid room member event in database.")) .map_err(|_| Error::bad_database("Invalid room member event in database."))

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

@ -164,7 +164,7 @@ impl Service {
for removed in statediffremoved.iter() { for removed in statediffremoved.iter() {
if !parent_new.remove(removed) { if !parent_new.remove(removed) {
// It was not added in the parent and we removed it // It was not added in the parent and we removed it
parent_removed.insert(removed.clone()); parent_removed.insert(*removed);
} }
// Else it was added in the parent and we removed it again. We can forget this change // Else it was added in the parent and we removed it again. We can forget this change
} }
@ -172,7 +172,7 @@ impl Service {
for new in statediffnew.iter() { for new in statediffnew.iter() {
if !parent_removed.remove(new) { if !parent_removed.remove(new) {
// It was not touched in the parent and we added it // It was not touched in the parent and we added it
parent_new.insert(new.clone()); parent_new.insert(*new);
} }
// Else it was removed in the parent and we added it again. We can forget this change // Else it was removed in the parent and we added it again. We can forget this change
} }
@ -217,7 +217,7 @@ impl Service {
for removed in statediffremoved.iter() { for removed in statediffremoved.iter() {
if !parent_new.remove(removed) { if !parent_new.remove(removed) {
// It was not added in the parent and we removed it // It was not added in the parent and we removed it
parent_removed.insert(removed.clone()); parent_removed.insert(*removed);
} }
// Else it was added in the parent and we removed it again. We can forget this change // Else it was added in the parent and we removed it again. We can forget this change
} }
@ -225,7 +225,7 @@ impl Service {
for new in statediffnew.iter() { for new in statediffnew.iter() {
if !parent_removed.remove(new) { if !parent_removed.remove(new) {
// It was not touched in the parent and we added it // It was not touched in the parent and we added it
parent_new.insert(new.clone()); parent_new.insert(*new);
} }
// Else it was removed in the parent and we added it again. We can forget this change // Else it was removed in the parent and we added it again. We can forget this change
} }

4
src/service/rooms/threads/mod.rs
View file

@ -26,7 +26,7 @@ impl Service {
self.db.threads_until(user_id, room_id, until, include) self.db.threads_until(user_id, room_id, until, include)
} }
pub fn add_to_thread<'a>(&'a self, root_event_id: &EventId, pdu: &PduEvent) -> Result<()> { pub fn add_to_thread(&self, root_event_id: &EventId, pdu: &PduEvent) -> Result<()> {
let root_id = &services() let root_id = &services()
.rooms .rooms
.timeline .timeline
@ -103,7 +103,7 @@ impl Service {
} }
let mut users = Vec::new(); let mut users = Vec::new();
if let Some(userids) = self.db.get_participants(&root_id)? { if let Some(userids) = self.db.get_participants(root_id)? {
users.extend_from_slice(&userids); users.extend_from_slice(&userids);
users.push(pdu.sender.clone()); users.push(pdu.sender.clone());
} else { } else {

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

@ -58,8 +58,8 @@ impl PduCount {
} }
pub fn try_from_string(token: &str) -> Result<Self> { pub fn try_from_string(token: &str) -> Result<Self> {
if token.starts_with('-') { if let Some(stripped_token) = token.strip_prefix('-') {
token[1..].parse().map(PduCount::Backfilled) stripped_token.parse().map(PduCount::Backfilled)
} else { } else {
token.parse().map(PduCount::Normal) token.parse().map(PduCount::Normal)
} }
@ -90,18 +90,6 @@ impl Ord for PduCount {
} }
} }
} }
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn comparisons() {
assert!(PduCount::Normal(1) < PduCount::Normal(2));
assert!(PduCount::Backfilled(2) < PduCount::Backfilled(1));
assert!(PduCount::Normal(1) > PduCount::Backfilled(1));
assert!(PduCount::Backfilled(1) < PduCount::Normal(1));
}
}
pub struct Service { pub struct Service {
pub db: &'static dyn Data, pub db: &'static dyn Data,
@ -112,7 +100,7 @@ pub struct Service {
impl Service { impl Service {
#[tracing::instrument(skip(self))] #[tracing::instrument(skip(self))]
pub fn first_pdu_in_room(&self, room_id: &RoomId) -> Result<Option<Arc<PduEvent>>> { pub fn first_pdu_in_room(&self, room_id: &RoomId) -> Result<Option<Arc<PduEvent>>> {
self.all_pdus(&user_id!("@doesntmatter:conduit.rs"), &room_id)? self.all_pdus(user_id!("@doesntmatter:conduit.rs"), room_id)?
.next() .next()
.map(|o| o.map(|(_, p)| Arc::new(p))) .map(|o| o.map(|(_, p)| Arc::new(p)))
.transpose() .transpose()
@ -475,7 +463,7 @@ impl Service {
let to_conduit = body.starts_with(&format!("{server_user}: ")) let to_conduit = body.starts_with(&format!("{server_user}: "))
|| body.starts_with(&format!("{server_user} ")) || body.starts_with(&format!("{server_user} "))
|| body == format!("{server_user}:") || body == format!("{server_user}:")
|| body == format!("{server_user}"); || body == server_user;
// This will evaluate to false if the emergency password is set up so that // This will evaluate to false if the emergency password is set up so that
// the administrator can execute commands as conduit // the administrator can execute commands as conduit
@ -867,7 +855,7 @@ impl Service {
.map_err(|_| Error::bad_database("Invalid content in pdu."))?; .map_err(|_| Error::bad_database("Invalid content in pdu."))?;
if content.membership == MembershipState::Leave { if content.membership == MembershipState::Leave {
if target == &server_user { if target == server_user {
warn!("Conduit user cannot leave from admins room"); warn!("Conduit user cannot leave from admins room");
return Err(Error::BadRequest( return Err(Error::BadRequest(
ErrorKind::Forbidden, ErrorKind::Forbidden,
@ -893,7 +881,7 @@ impl Service {
} }
if content.membership == MembershipState::Ban && pdu.state_key().is_some() { if content.membership == MembershipState::Ban && pdu.state_key().is_some() {
if target == &server_user { if target == server_user {
warn!("Conduit user cannot be banned in admins room"); warn!("Conduit user cannot be banned in admins room");
return Err(Error::BadRequest( return Err(Error::BadRequest(
ErrorKind::Forbidden, ErrorKind::Forbidden,
@ -1067,7 +1055,7 @@ impl Service {
#[tracing::instrument(skip(self, room_id))] #[tracing::instrument(skip(self, room_id))]
pub async fn backfill_if_required(&self, room_id: &RoomId, from: PduCount) -> Result<()> { pub async fn backfill_if_required(&self, room_id: &RoomId, from: PduCount) -> Result<()> {
let first_pdu = self let first_pdu = self
.all_pdus(&user_id!("@doesntmatter:conduit.rs"), &room_id)? .all_pdus(user_id!("@doesntmatter:conduit.rs"), room_id)?
.next() .next()
.expect("Room is not empty")?; .expect("Room is not empty")?;
@ -1079,7 +1067,7 @@ impl Service {
let power_levels: RoomPowerLevelsEventContent = services() let power_levels: RoomPowerLevelsEventContent = services()
.rooms .rooms
.state_accessor .state_accessor
.room_state_get(&room_id, &StateEventType::RoomPowerLevels, "")? .room_state_get(room_id, &StateEventType::RoomPowerLevels, "")?
.map(|ev| { .map(|ev| {
serde_json::from_str(ev.content.get()) serde_json::from_str(ev.content.get())
.map_err(|_| Error::bad_database("invalid m.room.power_levels event")) .map_err(|_| Error::bad_database("invalid m.room.power_levels event"))
@ -1110,11 +1098,9 @@ impl Service {
.await; .await;
match response { match response {
Ok(response) => { Ok(response) => {
let mut pub_key_map = RwLock::new(BTreeMap::new()); let pub_key_map = RwLock::new(BTreeMap::new());
for pdu in response.pdus { for pdu in response.pdus {
if let Err(e) = self if let Err(e) = self.backfill_pdu(backfill_server, pdu, &pub_key_map).await
.backfill_pdu(backfill_server, pdu, &mut pub_key_map)
.await
{ {
warn!("Failed to add backfilled pdu: {e}"); warn!("Failed to add backfilled pdu: {e}");
} }
@ -1161,13 +1147,13 @@ impl Service {
services() services()
.rooms .rooms
.event_handler .event_handler
.fetch_required_signing_keys([&value], &pub_key_map) .fetch_required_signing_keys([&value], pub_key_map)
.await?; .await?;
services() services()
.rooms .rooms
.event_handler .event_handler
.handle_incoming_pdu(origin, &event_id, &room_id, value, false, &pub_key_map) .handle_incoming_pdu(origin, &event_id, &room_id, value, false, pub_key_map)
.await?; .await?;
let value = self.get_pdu_json(&event_id)?.expect("We just created it"); let value = self.get_pdu_json(&event_id)?.expect("We just created it");
@ -1200,24 +1186,21 @@ impl Service {
drop(insert_lock); drop(insert_lock);
match pdu.kind { if pdu.kind == TimelineEventType::RoomMessage {
TimelineEventType::RoomMessage => { #[derive(Deserialize)]
#[derive(Deserialize)] struct ExtractBody {
struct ExtractBody { body: Option<String>,
body: Option<String>, }
}
let content = serde_json::from_str::<ExtractBody>(pdu.content.get())
let content = serde_json::from_str::<ExtractBody>(pdu.content.get()) .map_err(|_| Error::bad_database("Invalid content in pdu."))?;
.map_err(|_| Error::bad_database("Invalid content in pdu."))?;
if let Some(body) = content.body {
if let Some(body) = content.body { services()
services() .rooms
.rooms .search
.search .index_pdu(shortroomid, &pdu_id, &body)?;
.index_pdu(shortroomid, &pdu_id, &body)?;
}
} }
_ => {}
} }
drop(mutex_lock); drop(mutex_lock);
@ -1225,3 +1208,15 @@ impl Service {
Ok(()) Ok(())
} }
} }
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn comparisons() {
assert!(PduCount::Normal(1) < PduCount::Normal(2));
assert!(PduCount::Backfilled(2) < PduCount::Backfilled(1));
assert!(PduCount::Normal(1) > PduCount::Backfilled(1));
assert!(PduCount::Backfilled(1) < PduCount::Normal(1));
}
}

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

@ -132,7 +132,7 @@ impl Service {
for (key, outgoing_kind, event) in self.db.active_requests().filter_map(|r| r.ok()) { for (key, outgoing_kind, event) in self.db.active_requests().filter_map(|r| r.ok()) {
let entry = initial_transactions let entry = initial_transactions
.entry(outgoing_kind.clone()) .entry(outgoing_kind.clone())
.or_insert_with(Vec::new); .or_default();
if entry.len() > 30 { if entry.len() > 30 {
warn!( warn!(

4
src/service/users/mod.rs
View file

@ -139,10 +139,10 @@ impl Service {
cached cached
.subscriptions .subscriptions
.extend(request.room_subscriptions.clone().into_iter()); .extend(request.room_subscriptions.clone());
request request
.room_subscriptions .room_subscriptions
.extend(cached.subscriptions.clone().into_iter()); .extend(cached.subscriptions.clone());
request.extensions.e2ee.enabled = request request.extensions.e2ee.enabled = request
.extensions .extensions