oss/conduit
oss/conduit
1
0
Fork 0
mirror of https://gitlab.com/famedly/conduit.git synced 2025-07-22 17:18:35 +00:00
Code Wiki Activity

wip

Browse source
This commit is contained in:
mikoto 2024-05-07 07:42:10 +02:00
parent a0c4a639b2
commit ec48df78da
7 changed files with 193 additions and 145 deletions
Download patch file Download diff file Expand all files Collapse all files

4
Cargo.toml
View file

@ -78,7 +78,7 @@ http = "0.2.9"
# Used to find data directory for default db path
directories = "5"
# Used for ruma wrapper
serde_json = { version = "1.0.96", features = ["raw_value"] }
serde_json = { version = "1.0.96", features = ["raw_value", "preserve_order"] }
# Used for appservice registration files
serde_yaml = "0.9.21"
# Used for pdu definition
@ -157,8 +157,6 @@ tikv-jemallocator = { version = "0.5.0", features = [
sd-notify = { version = "0.4.1", optional = true }
url = "2.5.0"
[dependencies.rocksdb]
features = ["lz4", "multi-threaded-cf", "zstd"]
optional = true

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

@ -5,7 +5,7 @@ use crate::{
use ruma::{
api::client::{
filter::{FilterDefinition, LazyLoadOptions},
filter::{FilterDefinition, LazyLoadOptions, RoomFilter},
sync::sync_events::{
self,
v3::{
@ -19,7 +19,7 @@ use ruma::{
},
events::{
room::member::{MembershipState, RoomMemberEventContent},
StateEventType, TimelineEventType,
RoomAccountDataEventType, StateEventType, TimelineEventType,
},
serde::Raw,
uint, DeviceId, EventId, JsOption, OwnedDeviceId, OwnedUserId, RoomId, UInt, UserId,
@ -194,6 +194,8 @@ async fn sync_helper(
.unwrap_or_default(),
};
let event_fields = filter.event_fields.as_ref().map(Vec::as_slice);
let (lazy_load_enabled, lazy_load_send_redundant) = match filter.room.state.lazy_load_options {
LazyLoadOptions::Enabled {
include_redundant_members: redundant,
@ -241,6 +243,8 @@ async fn sync_helper(
next_batchcount,
lazy_load_enabled,
lazy_load_send_redundant,
&filter.room,
event_fields,
full_state,
&mut device_list_updates,
&mut left_encrypted_users,
@ -289,11 +293,14 @@ async fn sync_helper(
}
let mut left_rooms = BTreeMap::new();
let all_left_rooms: Vec<_> = services()
.rooms
.state_cache
.rooms_left(&sender_user)
.collect();
let all_left_rooms = match filter.room.include_leave {
false => Vec::with_capacity(0),
true => services()
.rooms
.state_cache
.rooms_left(&sender_user)
.collect(),
};
for result in all_left_rooms {
let (room_id, _) = result?;
@ -606,6 +613,8 @@ async fn load_joined_room(
next_batchcount: PduCount,
lazy_load_enabled: bool,
lazy_load_send_redundant: bool,
filter: &RoomFilter,
event_fields: &[String],
full_state: bool,
device_list_updates: &mut HashSet<OwnedUserId>,
left_encrypted_users: &mut HashSet<OwnedUserId>,
@ -1123,11 +1132,34 @@ async fn load_joined_room(
.account_data
.changes_since(Some(room_id), sender_user, since)?
.into_iter()
.filter_map(|(_, v)| {
// TODO: contains_url
.filter_map(|(et, v)| {
let et = et.to_string();
let (not_types, types) = (
&filter.account_data.not_types,
filter.account_data.types.as_ref(),
);
if not_types.contains(&et)
|| types.map(|v| !v.contains(&et)).unwrap_or_default()
{
return None;
}
serde_json::from_str(v.json().get())
.map_err(|_| Error::bad_database("Invalid account event in database."))
.ok()
})
.take(
filter
.account_data
.limit
.map(TryInto::try_into)
.map(Result::ok)
.flatten()
.unwrap_or(usize::MAX),
)
.collect(),
},
summary: RoomSummary {

6
src/database/key_value/account_data.rs
View file

@ -2,7 +2,7 @@ use std::collections::HashMap;
use ruma::{
api::client::error::ErrorKind,
events::{AnyEphemeralRoomEvent, RoomAccountDataEventType},
events::{AnyRoomAccountDataEvent, RoomAccountDataEventType},
serde::Raw,
RoomId, UserId,
};
@ -101,7 +101,7 @@ impl service::account_data::Data for KeyValueDatabase {
room_id: Option<&RoomId>,
user_id: &UserId,
since: u64,
) -> Result<HashMap<RoomAccountDataEventType, Raw<AnyEphemeralRoomEvent>>> {
) -> Result<HashMap<RoomAccountDataEventType, Raw<AnyRoomAccountDataEvent>>> {
let mut userdata = HashMap::new();
let mut prefix = room_id
@ -129,7 +129,7 @@ impl service::account_data::Data for KeyValueDatabase {
)?)
.map_err(|_| Error::bad_database("RoomUserData ID in db is invalid."))?,
),
serde_json::from_slice::<Raw<AnyEphemeralRoomEvent>>(&v).map_err(|_| {
serde_json::from_slice::<Raw<AnyRoomAccountDataEvent>>(&v).map_err(|_| {
Error::bad_database("Database contains invalid account data.")
})?,
))

4
src/service/account_data/data.rs
View file

@ -2,7 +2,7 @@ use std::collections::HashMap;
use crate::Result;
use ruma::{
events::{AnyEphemeralRoomEvent, RoomAccountDataEventType},
events::{AnyRoomAccountDataEvent, RoomAccountDataEventType},
serde::Raw,
RoomId, UserId,
};
@ -31,5 +31,5 @@ pub trait Data: Send + Sync {
room_id: Option<&RoomId>,
user_id: &UserId,
since: u64,
) -> Result<HashMap<RoomAccountDataEventType, Raw<AnyEphemeralRoomEvent>>>;
) -> Result<HashMap<RoomAccountDataEventType, Raw<AnyRoomAccountDataEvent>>>;
}

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

@ -3,7 +3,7 @@ mod data;
pub use data::Data;
use ruma::{
events::{AnyEphemeralRoomEvent, RoomAccountDataEventType},
events::{AnyRoomAccountDataEvent, AnyTimelineEvent, RoomAccountDataEventType},
serde::Raw,
RoomId, UserId,
};
@ -47,7 +47,7 @@ impl Service {
room_id: Option<&RoomId>,
user_id: &UserId,
since: u64,
) -> Result<HashMap<RoomAccountDataEventType, Raw<AnyEphemeralRoomEvent>>> {
) -> Result<HashMap<RoomAccountDataEventType, Raw<AnyRoomAccountDataEvent>>> {
self.db.changes_since(room_id, user_id, since)
}
}

1
src/service/pdu.rs
View file

@ -15,6 +15,7 @@ use serde::{Deserialize, Serialize};
use serde_json::{
json,
value::{to_raw_value, RawValue as RawJsonValue},
Map, Value,
};
use std::{cmp::Ordering, collections::BTreeMap, sync::Arc};
use tracing::warn;

271
src/utils/filter.rs
View file

@ -1,136 +1,153 @@
use std::collections::HashSet;
// pub fn filter_room_events<'i, I: Iterator<Item = &'i Raw<AnyTimelineEvent>>>(
// events: I,
// sender_user: &UserId,
// sender_device: &DeviceId,
// room_id: Option<&RoomId>,
// filter: RoomEventFilter,
// ) -> crate::Result<Box<dyn Iterator<Item = &'i Raw<AnyTimelineEvent>>>> {
// let (lazy_load_enabled, lazy_load_send_redundant) = match &filter.lazy_load_options {
// LazyLoadOptions::Enabled {
// include_redundant_members,
// } => (true, *include_redundant_members),
// _ => (false, false),
// };
use ruma::{
api::client::filter::{LazyLoadOptions, RoomEventFilter},
events::AnyTimelineEvent,
serde::Raw,
DeviceId, OwnedEventId, OwnedRoomId, OwnedUserId, RoomId, UserId,
};
// let it = Box::new(
// events
// .filter(|event| match &filter.rooms {
// None => true,
// Some(rooms) => rooms.iter().any(|r| {
// r.as_str()
// == event
// .get_field::<String>("room_id")
// .expect("room events should deserialize")
// .expect("room events should have a room_id")
// }),
// })
// .filter(|event| match &filter.not_rooms[..] {
// [] => true,
// not_rooms => not_rooms.iter().all(|r| {
// r.as_str()
// != event
// .get_field::<String>("room_id")
// .expect("room events should deserialize")
// .expect("room events should have a room_id")
// }),
// })
// .filter(|event| match &filter.senders {
// None => true,
// Some(rooms) => rooms.iter().any(|r| {
// r.as_str()
// == event
// .get_field::<String>("sender")
// .expect("room events should deserialize")
// .expect("room events should have a sender")
// }),
// })
// .filter(|event| match &filter.not_senders[..] {
// [] => true,
// not_senders => not_senders.iter().all(|r| {
// r.as_str()
// != event
// .get_field::<String>("sender")
// .expect("room events should deserialize")
// .expect("room events should have a sender")
// }),
// })
// .filter(|event| match &filter.types {
// None => true,
// Some(types) => types.iter().any(|t| {
// t.as_str()
// == event
// .get_field::<String>("type")
// .expect("room events should deserialize")
// .expect("room events should have a type")
// }),
// })
// .filter(|event| match &filter.not_types[..] {
// [] => true,
// not_types => not_types.iter().all(|t| {
// t.as_str()
// != event
// .get_field::<String>("type")
// .expect("room events should deserialize")
// .expect("room events should have a type")
// }),
// })
// .filter(|event| {
// let room_id = event
// .get_field::<OwnedRoomId>("room_id")
// .expect("room events should deserialize")
// .expect("room events should have a room_id");
// let event_id = event
// .get_field::<OwnedEventId>("event_id")
// .expect("room events should deserialize")
// .expect("room events should have an event_id");
use crate::services;
// services()
// .rooms
// .state_accessor
// .user_can_see_event(sender_user, &room_id, &event_id)
// .unwrap_or(false)
// }),
// );
pub fn filter_room_events<'i, I: Iterator<Item = &'i Raw<AnyTimelineEvent>>>(
events: I,
sender_user: &UserId,
sender_device: &DeviceId,
room_id: Option<&RoomId>,
filter: RoomEventFilter,
) -> crate::Result<Box<dyn Iterator<Item = &'i Raw<AnyTimelineEvent>>>> {
let (lazy_load_enabled, lazy_load_send_redundant) = match &filter.lazy_load_options {
LazyLoadOptions::Enabled {
include_redundant_members,
} => (true, *include_redundant_members),
_ => (false, false),
};
// let memberships = it
// .map(|event| {
// let room_id = event
// .get_field::<OwnedRoomId>("room_id")
// .expect("room events should deserialize")
// .expect("room events should have a room_id");
// let sender = event
// .get_field::<OwnedUserId>("sender")
// .expect("room events should deserialize")
// .expect("room events should have a sender");
let it = Box::new(
events
.filter(|event| match &filter.rooms {
None => true,
Some(rooms) => rooms.iter().any(|r| {
r.as_str()
== event
.get_field::<String>("room_id")
.expect("room events should deserialize")
.expect("room events should have a room_id")
}),
})
.filter(|event| match &filter.not_rooms[..] {
[] => true,
not_rooms => not_rooms.iter().all(|r| {
r.as_str()
!= event
.get_field::<String>("room_id")
.expect("room events should deserialize")
.expect("room events should have a room_id")
}),
})
.filter(|event| match &filter.senders {
None => true,
Some(rooms) => rooms.iter().any(|r| {
r.as_str()
== event
.get_field::<String>("sender")
.expect("room events should deserialize")
.expect("room events should have a sender")
}),
})
.filter(|event| match &filter.not_senders[..] {
[] => true,
not_senders => not_senders.iter().all(|r| {
r.as_str()
!= event
.get_field::<String>("sender")
.expect("room events should deserialize")
.expect("room events should have a sender")
}),
})
.filter(|event| match &filter.types {
None => true,
Some(types) => types.iter().any(|t| {
t.as_str()
== event
.get_field::<String>("type")
.expect("room events should deserialize")
.expect("room events should have a type")
}),
})
.filter(|event| match &filter.not_types[..] {
[] => true,
not_types => not_types.iter().all(|t| {
t.as_str()
!= event
.get_field::<String>("type")
.expect("room events should deserialize")
.expect("room events should have a type")
}),
})
.filter(|event| {
let room_id = event
.get_field::<OwnedRoomId>("room_id")
.expect("room events should deserialize")
.expect("room events should have a room_id");
let event_id = event
.get_field::<OwnedEventId>("event_id")
.expect("room events should deserialize")
.expect("room events should have an event_id");
// (room_id, sender)
// })
// .flat_map(|(room_id, sender)| {
// services()
// .rooms
// .lazy_loading
// .lazy_load_was_sent_before(sender_user, sender_device, &room_id, &sender)
// .map(|b| {
// if !b || lazy_load_send_redundant {
// Some(sender)
// } else {
// None
// }
// })
// .transpose()
// })
// .collect::<crate::Result<HashSet<_>>>()?;
services()
.rooms
.state_accessor
.user_can_see_event(sender_user, &room_id, &event_id)
.unwrap_or(false)
}),
);
// Ok(it)
// }
let memberships = it
.map(|event| {
let room_id = event
.get_field::<OwnedRoomId>("room_id")
.expect("room events should deserialize")
.expect("room events should have a room_id");
let sender = event
.get_field::<OwnedUserId>("sender")
.expect("room events should deserialize")
.expect("room events should have a sender");
use serde_json::Value;
(room_id, sender)
})
.flat_map(|(room_id, sender)| {
services()
.rooms
.lazy_loading
.lazy_load_was_sent_before(sender_user, sender_device, &room_id, &sender)
.map(|b| {
if !b || lazy_load_send_redundant {
Some(sender)
} else {
None
}
})
.transpose()
})
.collect::<crate::Result<HashSet<_>>>()?;
pub fn event_fields<I: Iterator<Item = String>>(mut json: Value, event_fields: I) -> Value {
let inner = json
.as_object_mut()
.expect("PduEvent should always be an object");
Ok(it)
// TODO: testing this properly
for field in event_fields {
let mut paths = field.split('.').peekable();
let mut parent = &mut *inner;
while let Some(key) = paths.next() {
if paths.peek().is_none() {
parent.remove(key);
} else {
parent = parent
.get_mut(key)
.map(Value::as_object_mut)
.flatten()
.unwrap();
}
}
}
json
}