nero/tsuki
1
0
Fork 0
Code Issues 2 Pull requests Projects Releases Packages Wiki Activity Actions

Fixed more compile time errors

Browse source
This commit is contained in:
Timo Kösters 2022-09-07 13:25:51 +02:00 • committed by Nyaaori
parent 785ddfc4aa
commit bd8b616ca0
No known key found for this signature in database
GPG key ID: E7819C3ED4D1F82E
103 changed files with 1617 additions and 2749 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -1,145 +1,32 @@
use crate::{utils, Error, Result};
use ruma::{
api::client::error::ErrorKind,
events::{AnyEphemeralRoomEvent, RoomAccountDataEventType},
serde::Raw,
RoomId, UserId,
};
use serde::{de::DeserializeOwned, Serialize};
use std::{collections::HashMap, sync::Arc};
use std::collections::HashMap;
impl AccountData {
use ruma::{UserId, RoomId, events::{RoomAccountDataEventType, AnyEphemeralRoomEvent}, serde::Raw};
use serde::{Serialize, de::DeserializeOwned};
use crate::Result;
pub trait Data {
/// Places one event in the account data of the user and removes the previous entry.
#[tracing::instrument(skip(self, room_id, user_id, event_type, data))]
pub fn update<T: Serialize>(
fn update<T: Serialize>(
&self,
room_id: Option<&RoomId>,
user_id: &UserId,
event_type: RoomAccountDataEventType,
data: &T,
) -> Result<()> {
let mut prefix = room_id
.map(|r| r.to_string())
.unwrap_or_default()
.as_bytes()
.to_vec();
prefix.push(0xff);
prefix.extend_from_slice(user_id.as_bytes());
prefix.push(0xff);
let mut roomuserdataid = prefix.clone();
roomuserdataid.extend_from_slice(&services().globals.next_count()?.to_be_bytes());
roomuserdataid.push(0xff);
roomuserdataid.extend_from_slice(event_type.to_string().as_bytes());
let mut key = prefix;
key.extend_from_slice(event_type.to_string().as_bytes());
let json = serde_json::to_value(data).expect("all types here can be serialized"); // TODO: maybe add error handling
if json.get("type").is_none() || json.get("content").is_none() {
return Err(Error::BadRequest(
ErrorKind::InvalidParam,
"Account data doesn't have all required fields.",
));
}
self.roomuserdataid_accountdata.insert(
&roomuserdataid,
&serde_json::to_vec(&json).expect("to_vec always works on json values"),
)?;
let prev = self.roomusertype_roomuserdataid.get(&key)?;
self.roomusertype_roomuserdataid
.insert(&key, &roomuserdataid)?;
// Remove old entry
if let Some(prev) = prev {
self.roomuserdataid_accountdata.remove(&prev)?;
}
Ok(())
}
) -> Result<()>;
/// Searches the account data for a specific kind.
#[tracing::instrument(skip(self, room_id, user_id, kind))]
pub fn get<T: DeserializeOwned>(
fn get<T: DeserializeOwned>(
&self,
room_id: Option<&RoomId>,
user_id: &UserId,
kind: RoomAccountDataEventType,
) -> Result<Option<T>> {
let mut key = room_id
.map(|r| r.to_string())
.unwrap_or_default()
.as_bytes()
.to_vec();
key.push(0xff);
key.extend_from_slice(user_id.as_bytes());
key.push(0xff);
key.extend_from_slice(kind.to_string().as_bytes());
self.roomusertype_roomuserdataid
.get(&key)?
.and_then(|roomuserdataid| {
self.roomuserdataid_accountdata
.get(&roomuserdataid)
.transpose()
})
.transpose()?
.map(|data| {
serde_json::from_slice(&data)
.map_err(|_| Error::bad_database("could not deserialize"))
})
.transpose()
}
) -> Result<Option<T>>;
/// Returns all changes to the account data that happened after `since`.
#[tracing::instrument(skip(self, room_id, user_id, since))]
pub fn changes_since(
fn changes_since(
&self,
room_id: Option<&RoomId>,
user_id: &UserId,
since: u64,
) -> Result<HashMap<RoomAccountDataEventType, Raw<AnyEphemeralRoomEvent>>> {
let mut userdata = HashMap::new();
let mut prefix = room_id
.map(|r| r.to_string())
.unwrap_or_default()
.as_bytes()
.to_vec();
prefix.push(0xff);
prefix.extend_from_slice(user_id.as_bytes());
prefix.push(0xff);
// Skip the data that's exactly at since, because we sent that last time
let mut first_possible = prefix.clone();
first_possible.extend_from_slice(&(since + 1).to_be_bytes());
for r in self
.roomuserdataid_accountdata
.iter_from(&first_possible, false)
.take_while(move |(k, _)| k.starts_with(&prefix))
.map(|(k, v)| {
Ok::<_, Error>((
RoomAccountDataEventType::try_from(
utils::string_from_bytes(k.rsplit(|&b| b == 0xff).next().ok_or_else(
|| Error::bad_database("RoomUserData ID in db is invalid."),
)?)
.map_err(|_| Error::bad_database("RoomUserData ID in db is invalid."))?,
)
.map_err(|_| Error::bad_database("RoomUserData ID in db is invalid."))?,
serde_json::from_slice::<Raw<AnyEphemeralRoomEvent>>(&v).map_err(|_| {
Error::bad_database("Database contains invalid account data.")
})?,
))
})
{
let (kind, data) = r?;
userdata.insert(kind, data);
}
Ok(userdata)
}
) -> Result<HashMap<RoomAccountDataEventType, Raw<AnyEphemeralRoomEvent>>>;
}

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

@ -1,14 +1,27 @@
use crate::{utils, Error, Result};
mod data;
pub use data::Data;
use ruma::{
api::client::error::ErrorKind,
api::client::{
error::ErrorKind,
},
events::{AnyEphemeralRoomEvent, RoomAccountDataEventType},
serde::Raw,
RoomId, UserId,
signatures::CanonicalJsonValue,
DeviceId, RoomId, UserId,
};
use serde::{de::DeserializeOwned, Serialize};
use std::{collections::HashMap, sync::Arc};
use tracing::error;
impl AccountData {
use crate::{service::*, services, utils, Error, Result};
pub struct Service<D: Data> {
db: D,
}
impl<D: Data> Service<D> {
/// Places one event in the account data of the user and removes the previous entry.
#[tracing::instrument(skip(self, room_id, user_id, event_type, data))]
pub fn update<T: Serialize>(

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

File diff suppressed because it is too large Load diff

5
src/service/appservice/data.rs
View file

@ -1,5 +1,6 @@
use crate::Result;
pub trait Data {
type Iter: Iterator;
/// Registers an appservice and returns the ID to the caller
fn register_appservice(&self, yaml: serde_yaml::Value) -> Result<String>;
@ -12,7 +13,7 @@ pub trait Data {
fn get_registration(&self, id: &str) -> Result<Option<serde_yaml::Value>>;
fn iter_ids(&self) -> Result<Self::Iter<Item = Result<String>>>;
fn iter_ids(&self) -> Result<Box<dyn Iterator<Item = Result<String>>>>;
fn all(&self) -> Result<Vec<(String, serde_yaml::Value)>>;
}

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

@ -1,13 +1,13 @@
mod data;
pub use data::Data;
use crate::service::*;
use crate::Result;
pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
impl<D: Data> Service<D> {
/// Registers an appservice and returns the ID to the caller
pub fn register_appservice(&self, yaml: serde_yaml::Value) -> Result<String> {
self.db.register_appservice(yaml)

8
src/service/globals/data.rs Normal file
View file

@ -0,0 +1,8 @@
use ruma::signatures::Ed25519KeyPair;
use crate::Result;
pub trait Data {
fn load_keypair(&self) -> Result<Ed25519KeyPair>;
fn remove_keypair(&self) -> Result<()>;
}

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

@ -3,7 +3,7 @@ pub use data::Data;
use crate::service::*;
use crate::{database::Config, server_server::FedDest, utils, Error, Result};
use crate::{Config, utils, Error, Result};
use ruma::{
api::{
client::sync::sync_events,
@ -25,8 +25,6 @@ use tokio::sync::{broadcast, watch::Receiver, Mutex as TokioMutex, Semaphore};
use tracing::error;
use trust_dns_resolver::TokioAsyncResolver;
use super::abstraction::Tree;
pub const COUNTER: &[u8] = b"c";
type WellKnownMap = HashMap<Box<ServerName>, (FedDest, String)>;
@ -93,47 +91,18 @@ impl Default for RotationHandler {
}
impl Service<_> {
impl<D: Data> Service<D> {
pub fn load(
globals: Arc<dyn Tree>,
server_signingkeys: Arc<dyn Tree>,
db: D,
config: Config,
) -> Result<Self> {
let keypair_bytes = globals.get(b"keypair")?.map_or_else(
|| {
let keypair = utils::generate_keypair();
globals.insert(b"keypair", &keypair)?;
Ok::<_, Error>(keypair)
},
|s| Ok(s.to_vec()),
)?;
let mut parts = keypair_bytes.splitn(2, |&b| b == 0xff);
let keypair = utils::string_from_bytes(
// 1. version
parts
.next()
.expect("splitn always returns at least one element"),
)
.map_err(|_| Error::bad_database("Invalid version bytes in keypair."))
.and_then(|version| {
// 2. key
parts
.next()
.ok_or_else(|| Error::bad_database("Invalid keypair format in database."))
.map(|key| (version, key))
})
.and_then(|(version, key)| {
ruma::signatures::Ed25519KeyPair::from_der(key, version)
.map_err(|_| Error::bad_database("Private or public keys are invalid."))
});
let keypair = db.load_keypair();
let keypair = match keypair {
Ok(k) => k,
Err(e) => {
error!("Keypair invalid. Deleting...");
globals.remove(b"keypair")?;
db.remove_keypair();
return Err(e);
}
};
@ -167,7 +136,7 @@ impl Service<_> {
let unstable_room_versions = vec![RoomVersionId::V3, RoomVersionId::V4, RoomVersionId::V5];
let mut s = Self {
globals,
db,
config,
keypair: Arc::new(keypair),
dns_resolver: TokioAsyncResolver::tokio_from_system_conf().map_err(|e| {
@ -181,7 +150,6 @@ impl Service<_> {
tls_name_override,
federation_client,
default_client,
server_signingkeys,
jwt_decoding_key,
stable_room_versions,
unstable_room_versions,

346
src/service/key_backups/data.rs
View file

@ -1,371 +1,85 @@
use crate::{utils, Error, Result, services};
use ruma::{
api::client::{
backup::{BackupAlgorithm, KeyBackupData, RoomKeyBackup},
error::ErrorKind,
},
serde::Raw,
RoomId, UserId,
};
use std::{collections::BTreeMap, sync::Arc};
use std::collections::BTreeMap;
impl KeyBackups {
pub fn create_backup(
use ruma::{api::client::backup::{BackupAlgorithm, RoomKeyBackup, KeyBackupData}, serde::Raw, UserId, RoomId};
use crate::Result;
pub trait Data {
fn create_backup(
&self,
user_id: &UserId,
backup_metadata: &Raw<BackupAlgorithm>,
) -> Result<String> {
let version = services().globals.next_count()?.to_string();
) -> Result<String>;
let mut key = user_id.as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(version.as_bytes());
fn delete_backup(&self, user_id: &UserId, version: &str) -> Result<()>;
self.backupid_algorithm.insert(
&key,
&serde_json::to_vec(backup_metadata).expect("BackupAlgorithm::to_vec always works"),
)?;
self.backupid_etag
.insert(&key, &services().globals.next_count()?.to_be_bytes())?;
Ok(version)
}
pub fn delete_backup(&self, user_id: &UserId, version: &str) -> Result<()> {
let mut key = user_id.as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(version.as_bytes());
self.backupid_algorithm.remove(&key)?;
self.backupid_etag.remove(&key)?;
key.push(0xff);
for (outdated_key, _) in self.backupkeyid_backup.scan_prefix(key) {
self.backupkeyid_backup.remove(&outdated_key)?;
}
Ok(())
}
pub fn update_backup(
fn update_backup(
&self,
user_id: &UserId,
version: &str,
backup_metadata: &Raw<BackupAlgorithm>,
) -> Result<String> {
let mut key = user_id.as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(version.as_bytes());
) -> Result<String>;
if self.backupid_algorithm.get(&key)?.is_none() {
return Err(Error::BadRequest(
ErrorKind::NotFound,
"Tried to update nonexistent backup.",
));
}
fn get_latest_backup_version(&self, user_id: &UserId) -> Result<Option<String>>;
self.backupid_algorithm
.insert(&key, backup_metadata.json().get().as_bytes())?;
self.backupid_etag
.insert(&key, &services().globals.next_count()?.to_be_bytes())?;
Ok(version.to_owned())
}
pub fn get_latest_backup_version(&self, user_id: &UserId) -> Result<Option<String>> {
let mut prefix = user_id.as_bytes().to_vec();
prefix.push(0xff);
let mut last_possible_key = prefix.clone();
last_possible_key.extend_from_slice(&u64::MAX.to_be_bytes());
self.backupid_algorithm
.iter_from(&last_possible_key, true)
.take_while(move |(k, _)| k.starts_with(&prefix))
.next()
.map(|(key, _)| {
utils::string_from_bytes(
key.rsplit(|&b| b == 0xff)
.next()
.expect("rsplit always returns an element"),
)
.map_err(|_| Error::bad_database("backupid_algorithm key is invalid."))
})
.transpose()
}
pub fn get_latest_backup(
fn get_latest_backup(
&self,
user_id: &UserId,
) -> Result<Option<(String, Raw<BackupAlgorithm>)>> {
let mut prefix = user_id.as_bytes().to_vec();
prefix.push(0xff);
let mut last_possible_key = prefix.clone();
last_possible_key.extend_from_slice(&u64::MAX.to_be_bytes());
) -> Result<Option<(String, Raw<BackupAlgorithm>)>>;
self.backupid_algorithm
.iter_from(&last_possible_key, true)
.take_while(move |(k, _)| k.starts_with(&prefix))
.next()
.map(|(key, value)| {
let version = utils::string_from_bytes(
key.rsplit(|&b| b == 0xff)
.next()
.expect("rsplit always returns an element"),
)
.map_err(|_| Error::bad_database("backupid_algorithm key is invalid."))?;
Ok((
version,
serde_json::from_slice(&value).map_err(|_| {
Error::bad_database("Algorithm in backupid_algorithm is invalid.")
})?,
))
})
.transpose()
}
pub fn get_backup(
fn get_backup(
&self,
user_id: &UserId,
version: &str,
) -> Result<Option<Raw<BackupAlgorithm>>> {
let mut key = user_id.as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(version.as_bytes());
) -> Result<Option<Raw<BackupAlgorithm>>>;
self.backupid_algorithm
.get(&key)?
.map_or(Ok(None), |bytes| {
serde_json::from_slice(&bytes)
.map_err(|_| Error::bad_database("Algorithm in backupid_algorithm is invalid."))
})
}
pub fn add_key(
fn add_key(
&self,
user_id: &UserId,
version: &str,
room_id: &RoomId,
session_id: &str,
key_data: &Raw<KeyBackupData>,
) -> Result<()> {
let mut key = user_id.as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(version.as_bytes());
) -> Result<()>;
if self.backupid_algorithm.get(&key)?.is_none() {
return Err(Error::BadRequest(
ErrorKind::NotFound,
"Tried to update nonexistent backup.",
));
}
fn count_keys(&self, user_id: &UserId, version: &str) -> Result<usize>;
self.backupid_etag
.insert(&key, &services().globals.next_count()?.to_be_bytes())?;
fn get_etag(&self, user_id: &UserId, version: &str) -> Result<String>;
key.push(0xff);
key.extend_from_slice(room_id.as_bytes());
key.push(0xff);
key.extend_from_slice(session_id.as_bytes());
self.backupkeyid_backup
.insert(&key, key_data.json().get().as_bytes())?;
Ok(())
}
pub fn count_keys(&self, user_id: &UserId, version: &str) -> Result<usize> {
let mut prefix = user_id.as_bytes().to_vec();
prefix.push(0xff);
prefix.extend_from_slice(version.as_bytes());
Ok(self.backupkeyid_backup.scan_prefix(prefix).count())
}
pub fn get_etag(&self, user_id: &UserId, version: &str) -> Result<String> {
let mut key = user_id.as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(version.as_bytes());
Ok(utils::u64_from_bytes(
&self
.backupid_etag
.get(&key)?
.ok_or_else(|| Error::bad_database("Backup has no etag."))?,
)
.map_err(|_| Error::bad_database("etag in backupid_etag invalid."))?
.to_string())
}
pub fn get_all(
fn get_all(
&self,
user_id: &UserId,
version: &str,
) -> Result<BTreeMap<Box<RoomId>, RoomKeyBackup>> {
let mut prefix = user_id.as_bytes().to_vec();
prefix.push(0xff);
prefix.extend_from_slice(version.as_bytes());
prefix.push(0xff);
) -> Result<BTreeMap<Box<RoomId>, RoomKeyBackup>>;
let mut rooms = BTreeMap::<Box<RoomId>, RoomKeyBackup>::new();
for result in self
.backupkeyid_backup
.scan_prefix(prefix)
.map(|(key, value)| {
let mut parts = key.rsplit(|&b| b == 0xff);
let session_id =
utils::string_from_bytes(parts.next().ok_or_else(|| {
Error::bad_database("backupkeyid_backup key is invalid.")
})?)
.map_err(|_| {
Error::bad_database("backupkeyid_backup session_id is invalid.")
})?;
let room_id = RoomId::parse(
utils::string_from_bytes(parts.next().ok_or_else(|| {
Error::bad_database("backupkeyid_backup key is invalid.")
})?)
.map_err(|_| Error::bad_database("backupkeyid_backup room_id is invalid."))?,
)
.map_err(|_| {
Error::bad_database("backupkeyid_backup room_id is invalid room id.")
})?;
let key_data = serde_json::from_slice(&value).map_err(|_| {
Error::bad_database("KeyBackupData in backupkeyid_backup is invalid.")
})?;
Ok::<_, Error>((room_id, session_id, key_data))
})
{
let (room_id, session_id, key_data) = result?;
rooms
.entry(room_id)
.or_insert_with(|| RoomKeyBackup {
sessions: BTreeMap::new(),
})
.sessions
.insert(session_id, key_data);
}
Ok(rooms)
}
pub fn get_room(
fn get_room(
&self,
user_id: &UserId,
version: &str,
room_id: &RoomId,
) -> Result<BTreeMap<String, Raw<KeyBackupData>>> {
let mut prefix = user_id.as_bytes().to_vec();
prefix.push(0xff);
prefix.extend_from_slice(version.as_bytes());
prefix.push(0xff);
prefix.extend_from_slice(room_id.as_bytes());
prefix.push(0xff);
) -> Result<BTreeMap<String, Raw<KeyBackupData>>>;
Ok(self
.backupkeyid_backup
.scan_prefix(prefix)
.map(|(key, value)| {
let mut parts = key.rsplit(|&b| b == 0xff);
let session_id =
utils::string_from_bytes(parts.next().ok_or_else(|| {
Error::bad_database("backupkeyid_backup key is invalid.")
})?)
.map_err(|_| {
Error::bad_database("backupkeyid_backup session_id is invalid.")
})?;
let key_data = serde_json::from_slice(&value).map_err(|_| {
Error::bad_database("KeyBackupData in backupkeyid_backup is invalid.")
})?;
Ok::<_, Error>((session_id, key_data))
})
.filter_map(|r| r.ok())
.collect())
}
pub fn get_session(
fn get_session(
&self,
user_id: &UserId,
version: &str,
room_id: &RoomId,
session_id: &str,
) -> Result<Option<Raw<KeyBackupData>>> {
let mut key = user_id.as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(version.as_bytes());
key.push(0xff);
key.extend_from_slice(room_id.as_bytes());
key.push(0xff);
key.extend_from_slice(session_id.as_bytes());
) -> Result<Option<Raw<KeyBackupData>>>;
self.backupkeyid_backup
.get(&key)?
.map(|value| {
serde_json::from_slice(&value).map_err(|_| {
Error::bad_database("KeyBackupData in backupkeyid_backup is invalid.")
})
})
.transpose()
}
fn delete_all_keys(&self, user_id: &UserId, version: &str) -> Result<()>;
pub fn delete_all_keys(&self, user_id: &UserId, version: &str) -> Result<()> {
let mut key = user_id.as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(version.as_bytes());
key.push(0xff);
for (outdated_key, _) in self.backupkeyid_backup.scan_prefix(key) {
self.backupkeyid_backup.remove(&outdated_key)?;
}
Ok(())
}
pub fn delete_room_keys(
fn delete_room_keys(
&self,
user_id: &UserId,
version: &str,
room_id: &RoomId,
) -> Result<()> {
let mut key = user_id.as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(version.as_bytes());
key.push(0xff);
key.extend_from_slice(room_id.as_bytes());
key.push(0xff);
) -> Result<()>;
for (outdated_key, _) in self.backupkeyid_backup.scan_prefix(key) {
self.backupkeyid_backup.remove(&outdated_key)?;
}
Ok(())
}
pub fn delete_room_key(
fn delete_room_key(
&self,
user_id: &UserId,
version: &str,
room_id: &RoomId,
session_id: &str,
) -> Result<()> {
let mut key = user_id.as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(version.as_bytes());
key.push(0xff);
key.extend_from_slice(room_id.as_bytes());
key.push(0xff);
key.extend_from_slice(session_id.as_bytes());
for (outdated_key, _) in self.backupkeyid_backup.scan_prefix(key) {
self.backupkeyid_backup.remove(&outdated_key)?;
}
Ok(())
}
) -> Result<()>;
}

9
src/service/key_backups/mod.rs
View file

@ -1,3 +1,6 @@
mod data;
pub use data::Data;
use crate::{utils, Error, Result, services};
use ruma::{
api::client::{
@ -9,7 +12,11 @@ use ruma::{
};
use std::{collections::BTreeMap, sync::Arc};
impl KeyBackups {
pub struct Service<D: Data> {
db: D,
}
impl<D: Data> Service<D> {
pub fn create_backup(
&self,
user_id: &UserId,

8
src/service/media/data.rs Normal file
View file

@ -0,0 +1,8 @@
use crate::Result;
pub trait Data {
fn create_file_metadata(&self, mxc: String, width: u32, height: u32, content_disposition: &Option<&str>, content_type: &Option<&str>) -> Result<Vec<u8>>;
/// Returns content_disposition, content_type and the metadata key.
fn search_file_metadata(&self, mxc: String, width: u32, height: u32) -> Result<(Option<String>, Option<String>, Vec<u8>)>;
}

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

@ -1,7 +1,8 @@
use image::{imageops::FilterType, GenericImageView};
mod data;
pub use data::Data;
use super::abstraction::Tree;
use crate::{utils, Error, Result};
use image::{imageops::FilterType, GenericImageView};
use crate::{utils, Error, Result, services};
use std::{mem, sync::Arc};
use tokio::{
fs::File,
@ -14,44 +15,25 @@ pub struct FileMeta {
pub file: Vec<u8>,
}
pub struct Media {
pub(super) mediaid_file: Arc<dyn Tree>, // MediaId = MXC + WidthHeight + ContentDisposition + ContentType
pub struct Service<D: Data> {
db: D,
}
impl Media {
impl<D: Data> Service<D> {
/// Uploads a file.
pub async fn create(
&self,
mxc: String,
globals: &Globals,
content_disposition: &Option<&str>,
content_type: &Option<&str>,
file: &[u8],
) -> Result<()> {
let mut key = mxc.as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(&0_u32.to_be_bytes()); // Width = 0 if it's not a thumbnail
key.extend_from_slice(&0_u32.to_be_bytes()); // Height = 0 if it's not a thumbnail
key.push(0xff);
key.extend_from_slice(
content_disposition
.as_ref()
.map(|f| f.as_bytes())
.unwrap_or_default(),
);
key.push(0xff);
key.extend_from_slice(
content_type
.as_ref()
.map(|c| c.as_bytes())
.unwrap_or_default(),
);
// Width, Height = 0 if it's not a thumbnail
let key = self.db.create_file_metadata(mxc, 0, 0, content_disposition, content_type);
let path = globals.get_media_file(&key);
let path = services().globals.get_media_file(&key);
let mut f = File::create(path).await?;
f.write_all(file).await?;
self.mediaid_file.insert(&key, &[])?;
Ok(())
}
@ -60,80 +42,28 @@ impl Media {
pub async fn upload_thumbnail(
&self,
mxc: String,
globals: &Globals,
content_disposition: &Option<String>,
content_type: &Option<String>,
width: u32,
height: u32,
file: &[u8],
) -> Result<()> {
let mut key = mxc.as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(&width.to_be_bytes());
key.extend_from_slice(&height.to_be_bytes());
key.push(0xff);
key.extend_from_slice(
content_disposition
.as_ref()
.map(|f| f.as_bytes())
.unwrap_or_default(),
);
key.push(0xff);
key.extend_from_slice(
content_type
.as_ref()
.map(|c| c.as_bytes())
.unwrap_or_default(),
);
let key = self.db.create_file_metadata(mxc, width, height, content_disposition, content_type);
let path = globals.get_media_file(&key);
let path = services().globals.get_media_file(&key);
let mut f = File::create(path).await?;
f.write_all(file).await?;
self.mediaid_file.insert(&key, &[])?;
Ok(())
}
/// Downloads a file.
pub async fn get(&self, globals: &Globals, mxc: &str) -> Result<Option<FileMeta>> {
let mut prefix = mxc.as_bytes().to_vec();
prefix.push(0xff);
prefix.extend_from_slice(&0_u32.to_be_bytes()); // Width = 0 if it's not a thumbnail
prefix.extend_from_slice(&0_u32.to_be_bytes()); // Height = 0 if it's not a thumbnail
prefix.push(0xff);
let first = self.mediaid_file.scan_prefix(prefix).next();
if let Some((key, _)) = first {
let path = globals.get_media_file(&key);
pub async fn get(&self, mxc: String) -> Result<Option<FileMeta>> {
if let Ok((content_disposition, content_type, key)) = self.db.search_file_metadata(mxc, 0, 0) {
let path = services().globals.get_media_file(&key);
let mut file = Vec::new();
File::open(path).await?.read_to_end(&mut file).await?;
let mut parts = key.rsplit(|&b| b == 0xff);
let content_type = parts
.next()
.map(|bytes| {
utils::string_from_bytes(bytes).map_err(|_| {
Error::bad_database("Content type in mediaid_file is invalid unicode.")
})
})
.transpose()?;
let content_disposition_bytes = parts
.next()
.ok_or_else(|| Error::bad_database("Media ID in db is invalid."))?;
let content_disposition = if content_disposition_bytes.is_empty() {
None
} else {
Some(
utils::string_from_bytes(content_disposition_bytes).map_err(|_| {
Error::bad_database(
"Content Disposition in mediaid_file is invalid unicode.",
)
})?,
)
};
Ok(Some(FileMeta {
content_disposition,
@ -170,8 +100,7 @@ impl Media {
/// For width,height <= 96 the server uses another thumbnailing algorithm which crops the image afterwards.
pub async fn get_thumbnail(
&self,
mxc: &str,
globals: &Globals,
mxc: String,
width: u32,
height: u32,
) -> Result<Option<FileMeta>> {
@ -179,89 +108,23 @@ impl Media {
.thumbnail_properties(width, height)
.unwrap_or((0, 0, false)); // 0, 0 because that's the original file
let mut main_prefix = mxc.as_bytes().to_vec();
main_prefix.push(0xff);
let mut thumbnail_prefix = main_prefix.clone();
thumbnail_prefix.extend_from_slice(&width.to_be_bytes());
thumbnail_prefix.extend_from_slice(&height.to_be_bytes());
thumbnail_prefix.push(0xff);
let mut original_prefix = main_prefix;
original_prefix.extend_from_slice(&0_u32.to_be_bytes()); // Width = 0 if it's not a thumbnail
original_prefix.extend_from_slice(&0_u32.to_be_bytes()); // Height = 0 if it's not a thumbnail
original_prefix.push(0xff);
let first_thumbnailprefix = self.mediaid_file.scan_prefix(thumbnail_prefix).next();
let first_originalprefix = self.mediaid_file.scan_prefix(original_prefix).next();
if let Some((key, _)) = first_thumbnailprefix {
if let Ok((content_disposition, content_type, key)) = self.db.search_file_metadata(mxc, width, height) {
// Using saved thumbnail
let path = globals.get_media_file(&key);
let path = services().globals.get_media_file(&key);
let mut file = Vec::new();
File::open(path).await?.read_to_end(&mut file).await?;
let mut parts = key.rsplit(|&b| b == 0xff);
let content_type = parts
.next()
.map(|bytes| {
utils::string_from_bytes(bytes).map_err(|_| {
Error::bad_database("Content type in mediaid_file is invalid unicode.")
})
})
.transpose()?;
let content_disposition_bytes = parts
.next()
.ok_or_else(|| Error::bad_database("Media ID in db is invalid."))?;
let content_disposition = if content_disposition_bytes.is_empty() {
None
} else {
Some(
utils::string_from_bytes(content_disposition_bytes).map_err(|_| {
Error::bad_database("Content Disposition in db is invalid.")
})?,
)
};
Ok(Some(FileMeta {
content_disposition,
content_type,
file: file.to_vec(),
}))
} else if let Some((key, _)) = first_originalprefix {
} else if let Ok((content_disposition, content_type, key)) = self.db.search_file_metadata(mxc, 0, 0) {
// Generate a thumbnail
let path = globals.get_media_file(&key);
let path = services().globals.get_media_file(&key);
let mut file = Vec::new();
File::open(path).await?.read_to_end(&mut file).await?;
let mut parts = key.rsplit(|&b| b == 0xff);
let content_type = parts
.next()
.map(|bytes| {
utils::string_from_bytes(bytes).map_err(|_| {
Error::bad_database("Content type in mediaid_file is invalid unicode.")
})
})
.transpose()?;
let content_disposition_bytes = parts
.next()
.ok_or_else(|| Error::bad_database("Media ID in db is invalid."))?;
let content_disposition = if content_disposition_bytes.is_empty() {
None
} else {
Some(
utils::string_from_bytes(content_disposition_bytes).map_err(|_| {
Error::bad_database(
"Content Disposition in mediaid_file is invalid unicode.",
)
})?,
)
};
if let Ok(image) = image::load_from_memory(&file) {
let original_width = image.width();
let original_height = image.height();
@ -317,26 +180,12 @@ impl Media {
thumbnail.write_to(&mut thumbnail_bytes, image::ImageOutputFormat::Png)?;
// Save thumbnail in database so we don't have to generate it again next time
let mut thumbnail_key = key.to_vec();
let width_index = thumbnail_key
.iter()
.position(|&b| b == 0xff)
.ok_or_else(|| Error::bad_database("Media in db is invalid."))?
+ 1;
let mut widthheight = width.to_be_bytes().to_vec();
widthheight.extend_from_slice(&height.to_be_bytes());
let thumbnail_key = self.db.create_file_metadata(mxc, width, height, content_disposition, content_type)?;
thumbnail_key.splice(
width_index..width_index + 2 * mem::size_of::<u32>(),
widthheight,
);
let path = globals.get_media_file(&thumbnail_key);
let path = services().globals.get_media_file(&thumbnail_key);
let mut f = File::create(path).await?;
f.write_all(&thumbnail_bytes).await?;
self.mediaid_file.insert(&thumbnail_key, &[])?;
Ok(Some(FileMeta {
content_disposition,
content_type,

27
src/service/mod.rs
View file

@ -1,28 +1,29 @@
pub mod pdu;
pub mod appservice;
pub mod pusher;
pub mod rooms;
pub mod transaction_ids;
pub mod uiaa;
pub mod users;
pub mod account_data;
pub mod admin;
pub mod appservice;
pub mod globals;
pub mod key_backups;
pub mod media;
pub mod pdu;
pub mod pusher;
pub mod rooms;
pub mod sending;
pub mod transaction_ids;
pub mod uiaa;
pub mod users;
pub struct Services<D> {
pub struct Services<D: appservice::Data + pusher::Data + rooms::Data + transaction_ids::Data + uiaa::Data + users::Data + account_data::Data + globals::Data + key_backups::Data + media::Data>
{
pub appservice: appservice::Service<D>,
pub pusher: pusher::Service<D>,
pub rooms: rooms::Service<D>,
pub transaction_ids: transaction_ids::Service<D>,
pub uiaa: uiaa::Service<D>,
pub users: users::Service<D>,
//pub account_data: account_data::Service<D>,
//pub admin: admin::Service<D>,
pub account_data: account_data::Service<D>,
pub admin: admin::Service,
pub globals: globals::Service<D>,
//pub key_backups: key_backups::Service<D>,
//pub media: media::Service<D>,
//pub sending: sending::Service<D>,
pub key_backups: key_backups::Service<D>,
pub media: media::Service<D>,
pub sending: sending::Service,
}

4
src/service/pdu.rs
View file

@ -1,4 +1,4 @@
use crate::{Database, Error, services};
use crate::{Error, services};
use ruma::{
events::{
room::member::RoomMemberEventContent, AnyEphemeralRoomEvent, AnyRoomEvent, AnyStateEvent,
@ -357,7 +357,7 @@ pub(crate) fn gen_event_id_canonical_json(
Ok((event_id, value))
}
/// Build the start of a PDU in order to add it to the `Database`.
/// Build the start of a PDU in order to add it to the Database.
#[derive(Debug, Deserialize)]
pub struct PduBuilder {
#[serde(rename = "type")]

3
src/service/pusher/data.rs
View file

@ -1,4 +1,5 @@
use ruma::{UserId, api::client::push::{set_pusher, get_pushers}};
use crate::Result;
pub trait Data {
fn set_pusher(&self, sender: &UserId, pusher: set_pusher::v3::Pusher) -> Result<()>;
@ -10,5 +11,5 @@ pub trait Data {
fn get_pusher_senderkeys<'a>(
&'a self,
sender: &UserId,
) -> impl Iterator<Item = Vec<u8>> + 'a;
) -> Box<dyn Iterator<Item = Vec<u8>>>;
}

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

@ -1,7 +1,7 @@
mod data;
pub use data::Data;
use crate::{services, Error, PduEvent};
use crate::{services, Error, PduEvent, Result};
use bytes::BytesMut;
use ruma::{
api::{
@ -27,7 +27,7 @@ pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
impl<D: Data> Service<D> {
pub fn set_pusher(&self, sender: &UserId, pusher: set_pusher::v3::Pusher) -> Result<()> {
self.db.set_pusher(sender, pusher)
}

11
src/service/rooms/alias/data.rs
View file

@ -1,24 +1,29 @@
use ruma::{RoomId, RoomAliasId};
use crate::Result;
pub trait Data {
/// Creates or updates the alias to the given room id.
fn set_alias(
&self,
alias: &RoomAliasId,
room_id: &RoomId
) -> Result<()>;
/// Forgets about an alias. Returns an error if the alias did not exist.
fn remove_alias(
&self,
alias: &RoomAliasId,
) -> Result<()>;
/// Looks up the roomid for the given alias.
fn resolve_local_alias(
&self,
alias: &RoomAliasId,
) -> Result<()>;
) -> Result<Option<Box<RoomId>>>;
/// Returns all local aliases that point to the given room
fn local_aliases_for_room(
alias: &RoomAliasId,
) -> Result<()>;
&self,
room_id: &RoomId,
) -> Result<Box<dyn Iterator<Item=String>>>;
}

6
src/service/rooms/alias/mod.rs
View file

@ -1,12 +1,14 @@
mod data;
pub use data::Data;
use ruma::{RoomAliasId, RoomId};
use crate::Result;
pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
impl<D: Data> Service<D> {
#[tracing::instrument(skip(self))]
pub fn set_alias(
&self,
@ -26,7 +28,7 @@ impl Service<_> {
#[tracing::instrument(skip(self))]
pub fn resolve_local_alias(&self, alias: &RoomAliasId) -> Result<Option<Box<RoomId>>> {
self.db.resolve_local_alias(alias: &RoomAliasId)
self.db.resolve_local_alias(alias)
}
#[tracing::instrument(skip(self))]

5
src/service/rooms/auth_chain/data.rs
View file

@ -1,6 +1,7 @@
use std::collections::HashSet;
use crate::Result;
pub trait Data {
fn get_cached_eventid_authchain<'a>() -> Result<HashSet<u64>>;
fn cache_eventid_authchain<'a>(shorteventid: u64, auth_chain: &HashSet<u64>) -> Result<HashSet<u64>>;
fn get_cached_eventid_authchain(&self, shorteventid: u64) -> Result<HashSet<u64>>;
fn cache_eventid_authchain(&self, shorteventid: u64, auth_chain: &HashSet<u64>) -> Result<()>;
}

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

@ -3,13 +3,13 @@ use std::{sync::Arc, collections::HashSet};
pub use data::Data;
use crate::service::*;
use crate::Result;
pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
impl<D: Data> Service<D> {
#[tracing::instrument(skip(self))]
pub fn get_cached_eventid_authchain<'a>(
&'a self,

9
src/service/rooms/directory/data.rs
View file

@ -1,15 +1,16 @@
use ruma::RoomId;
use crate::Result;
pub trait Data {
/// Adds the room to the public room directory
fn set_public(room_id: &RoomId) -> Result<()>;
fn set_public(&self, room_id: &RoomId) -> Result<()>;
/// Removes the room from the public room directory.
fn set_not_public(room_id: &RoomId) -> Result<()>;
fn set_not_public(&self, room_id: &RoomId) -> Result<()>;
/// Returns true if the room is in the public room directory.
fn is_public_room(room_id: &RoomId) -> Result<bool>;
fn is_public_room(&self, room_id: &RoomId) -> Result<bool>;
/// Returns the unsorted public room directory
fn public_rooms() -> impl Iterator<Item = Result<Box<RoomId>>> + '_;
fn public_rooms(&self) -> Box<dyn Iterator<Item = Result<Box<RoomId>>>>;
}

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

@ -2,13 +2,13 @@ mod data;
pub use data::Data;
use ruma::RoomId;
use crate::service::*;
use crate::Result;
pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
impl<D: Data> Service<D> {
#[tracing::instrument(skip(self))]
pub fn set_public(&self, room_id: &RoomId) -> Result<()> {
self.db.set_public(room_id)

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

@ -2,7 +2,9 @@ pub mod presence;
pub mod read_receipt;
pub mod typing;
pub struct Service<D> {
pub trait Data: presence::Data + read_receipt::Data + typing::Data {}
pub struct Service<D: Data> {
presence: presence::Service<D>,
read_receipt: read_receipt::Service<D>,
typing: typing::Service<D>,

1
src/service/rooms/edus/presence/data.rs
View file

@ -1,6 +1,7 @@
use std::collections::HashMap;
use ruma::{UserId, RoomId, events::presence::PresenceEvent};
use crate::Result;
pub trait Data {
/// Adds a presence event which will be saved until a new event replaces it.

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

@ -4,13 +4,13 @@ use std::collections::HashMap;
pub use data::Data;
use ruma::{RoomId, UserId, events::presence::PresenceEvent};
use crate::service::*;
use crate::Result;
pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
impl<D: Data> Service<D> {
/// Adds a presence event which will be saved until a new event replaces it.
///
/// Note: This method takes a RoomId because presence updates are always bound to rooms to

5
src/service/rooms/edus/read_receipt/data.rs
View file

@ -1,4 +1,5 @@
use ruma::{RoomId, events::receipt::ReceiptEvent, UserId, serde::Raw};
use crate::Result;
pub trait Data {
/// Replaces the previous read receipt.
@ -14,13 +15,13 @@ pub trait Data {
&self,
room_id: &RoomId,
since: u64,
) -> impl Iterator<
) -> Box<dyn Iterator<
Item = Result<(
Box<UserId>,
u64,
Raw<ruma::events::AnySyncEphemeralRoomEvent>,
)>,
>;
>>;
/// Sets a private read marker at `count`.
fn private_read_set(&self, room_id: &RoomId, user_id: &UserId, count: u64) -> Result<()>;

4
src/service/rooms/edus/read_receipt/mod.rs
View file

@ -1,12 +1,14 @@
mod data;
pub use data::Data;
use ruma::{RoomId, UserId, events::receipt::ReceiptEvent, serde::Raw};
use crate::Result;
pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
impl<D: Data> Service<D> {
/// Replaces the previous read receipt.
pub fn readreceipt_update(
&self,

4
src/service/rooms/edus/typing/data.rs
View file

@ -1,5 +1,5 @@
use std::collections::HashSet;
use crate::Result;
use ruma::{UserId, RoomId};
pub trait Data {
@ -14,5 +14,5 @@ pub trait Data {
fn last_typing_update(&self, room_id: &RoomId) -> Result<u64>;
/// Returns all user ids currently typing.
fn typings_all(&self, room_id: &RoomId) -> Result<HashSet<UserId>>;
fn typings_all(&self, room_id: &RoomId) -> Result<HashSet<Box<UserId>>>;
}

6
src/service/rooms/edus/typing/mod.rs
View file

@ -1,14 +1,14 @@
mod data;
pub use data::Data;
use ruma::{UserId, RoomId};
use ruma::{UserId, RoomId, events::SyncEphemeralRoomEvent};
use crate::service::*;
use crate::Result;
pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
impl<D: Data> Service<D> {
/// Sets a user as typing until the timeout timestamp is reached or roomtyping_remove is
/// called.
pub fn typing_add(&self, user_id: &UserId, room_id: &RoomId, timeout: u64) -> Result<()> {

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

@ -250,7 +250,7 @@ impl Service {
// We go through all the signatures we see on the value and fetch the corresponding signing
// keys
self.fetch_required_signing_keys(&value, pub_key_map, db)
self.fetch_required_signing_keys(&value, pub_key_map)
.await?;
// 2. Check signatures, otherwise drop
@ -1153,6 +1153,11 @@ impl Service {
let mut eventid_info = HashMap::new();
let mut todo_outlier_stack: Vec<Arc<EventId>> = initial_set;
let first_pdu_in_room = services()
.rooms
.first_pdu_in_room(room_id)?
.ok_or_else(|| Error::bad_database("Failed to find first pdu in db."))?;
let mut amount = 0;
while let Some(prev_event_id) = todo_outlier_stack.pop() {

1
src/service/rooms/lazy_loading/data.rs
View file

@ -1,4 +1,5 @@
use ruma::{RoomId, DeviceId, UserId};
use crate::Result;
pub trait Data {
fn lazy_load_was_sent_before(

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

@ -4,13 +4,13 @@ use std::collections::HashSet;
pub use data::Data;
use ruma::{DeviceId, UserId, RoomId};
use crate::service::*;
use crate::Result;
pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
impl<D: Data> Service<D> {
#[tracing::instrument(skip(self))]
pub fn lazy_load_was_sent_before(
&self,

1
src/service/rooms/metadata/data.rs
View file

@ -1,4 +1,5 @@
use ruma::RoomId;
use crate::Result;
pub trait Data {
fn exists(&self, room_id: &RoomId) -> Result<bool>;

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

@ -2,13 +2,13 @@ mod data;
pub use data::Data;
use ruma::RoomId;
use crate::service::*;
use crate::Result;
pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
impl<D: Data> Service<D> {
/// Checks if a room exists.
#[tracing::instrument(skip(self))]
pub fn exists(&self, room_id: &RoomId) -> Result<bool> {

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

@ -16,7 +16,9 @@ pub mod state_compressor;
pub mod timeline;
pub mod user;
pub struct Service<D> {
pub trait Data: alias::Data + auth_chain::Data + directory::Data + edus::Data + lazy_loading::Data + metadata::Data + outlier::Data + pdu_metadata::Data + search::Data + short::Data + state::Data + state_accessor::Data + state_cache::Data + state_compressor::Data + timeline::Data + user::Data {}
pub struct Service<D: Data> {
pub alias: alias::Service<D>,
pub auth_chain: auth_chain::Service<D>,
pub directory: directory::Service<D>,

4
src/service/rooms/outlier/data.rs
View file

@ -1,6 +1,6 @@
use ruma::{EventId, signatures::CanonicalJsonObject};
use ruma::{signatures::CanonicalJsonObject, EventId};
use crate::PduEvent;
use crate::{PduEvent, Result};
pub trait Data {
fn get_outlier_pdu_json(&self, event_id: &EventId) -> Result<Option<CanonicalJsonObject>>;

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

@ -2,13 +2,13 @@ mod data;
pub use data::Data;
use ruma::{EventId, signatures::CanonicalJsonObject};
use crate::{service::*, PduEvent};
use crate::{Result, PduEvent};
pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
impl<D: Data> Service<D> {
/// Returns the pdu from the outlier tree.
pub fn get_outlier_pdu_json(&self, event_id: &EventId) -> Result<Option<CanonicalJsonObject>> {
self.db.get_outlier_pdu_json(event_id)

1
src/service/rooms/pdu_metadata/data.rs
View file

@ -1,6 +1,7 @@
use std::sync::Arc;
use ruma::{EventId, RoomId};
use crate::Result;
pub trait Data {
fn mark_as_referenced(&self, room_id: &RoomId, event_ids: &[Arc<EventId>]) -> Result<()>;

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

@ -4,13 +4,13 @@ use std::sync::Arc;
pub use data::Data;
use ruma::{RoomId, EventId};
use crate::service::*;
use crate::Result;
pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
impl<D: Data> Service<D> {
#[tracing::instrument(skip(self, room_id, event_ids))]
pub fn mark_as_referenced(&self, room_id: &RoomId, event_ids: &[Arc<EventId>]) -> Result<()> {
self.db.mark_as_referenced(room_id, event_ids)

5
src/service/rooms/search/data.rs
View file

@ -1,11 +1,12 @@
use ruma::RoomId;
use crate::Result;
pub trait Data {
fn index_pdu<'a>(&self, room_id: &RoomId, pdu_id: u64, message_body: String) -> Result<()>;
fn index_pdu<'a>(&self, shortroomid: u64, pdu_id: u64, message_body: String) -> Result<()>;
fn search_pdus<'a>(
&'a self,
room_id: &RoomId,
search_string: &str,
) -> Result<Option<(impl Iterator<Item = Vec<u8>> + 'a, Vec<String>)>>;
) -> Result<Option<(Box<dyn Iterator<Item = Vec<u8>>>, Vec<String>)>>;
}

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

@ -1,12 +1,14 @@
mod data;
pub use data::Data;
use crate::Result;
use ruma::RoomId;
pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
impl<D: Data> Service<D> {
#[tracing::instrument(skip(self))]
pub fn search_pdus<'a>(
&'a self,

2
src/service/rooms/short/data.rs Normal file
View file

@ -0,0 +1,2 @@
pub trait Data {
}

19
src/service/rooms/short/mod.rs
View file

@ -2,19 +2,18 @@ mod data;
use std::sync::Arc;
pub use data::Data;
use ruma::{EventId, events::StateEventType};
use ruma::{EventId, events::StateEventType, RoomId};
use crate::{service::*, Error, utils};
use crate::{Result, Error, utils, services};
pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
impl<D: Data> Service<D> {
pub fn get_or_create_shorteventid(
&self,
event_id: &EventId,
globals: &super::globals::Globals,
) -> Result<u64> {
if let Some(short) = self.eventidshort_cache.lock().unwrap().get_mut(event_id) {
return Ok(*short);
@ -24,7 +23,7 @@ impl Service<_> {
Some(shorteventid) => utils::u64_from_bytes(&shorteventid)
.map_err(|_| Error::bad_database("Invalid shorteventid in db."))?,
None => {
let shorteventid = globals.next_count()?;
let shorteventid = services().globals.next_count()?;
self.eventid_shorteventid
.insert(event_id.as_bytes(), &shorteventid.to_be_bytes())?;
self.shorteventid_eventid
@ -82,7 +81,6 @@ impl Service<_> {
&self,
event_type: &StateEventType,
state_key: &str,
globals: &super::globals::Globals,
) -> Result<u64> {
if let Some(short) = self
.statekeyshort_cache
@ -101,7 +99,7 @@ impl Service<_> {
Some(shortstatekey) => utils::u64_from_bytes(&shortstatekey)
.map_err(|_| Error::bad_database("Invalid shortstatekey in db."))?,
None => {
let shortstatekey = globals.next_count()?;
let shortstatekey = services().globals.next_count()?;
self.statekey_shortstatekey
.insert(&statekey, &shortstatekey.to_be_bytes())?;
self.shortstatekey_statekey
@ -190,7 +188,7 @@ impl Service<_> {
/// Returns (shortstatehash, already_existed)
fn get_or_create_shortstatehash(
&self,
state_hash: &StateHashId,
state_hash: &[u8],
) -> Result<(u64, bool)> {
Ok(match self.statehash_shortstatehash.get(state_hash)? {
Some(shortstatehash) => (
@ -199,7 +197,7 @@ impl Service<_> {
true,
),
None => {
let shortstatehash = globals.next_count()?;
let shortstatehash = services().globals.next_count()?;
self.statehash_shortstatehash
.insert(state_hash, &shortstatehash.to_be_bytes())?;
(shortstatehash, false)
@ -220,13 +218,12 @@ impl Service<_> {
pub fn get_or_create_shortroomid(
&self,
room_id: &RoomId,
globals: &super::globals::Globals,
) -> Result<u64> {
Ok(match self.roomid_shortroomid.get(room_id.as_bytes())? {
Some(short) => utils::u64_from_bytes(&short)
.map_err(|_| Error::bad_database("Invalid shortroomid in db."))?,
None => {
let short = globals.next_count()?;
let short = services().globals.next_count()?;
self.roomid_shortroomid
.insert(room_id.as_bytes(), &short.to_be_bytes())?;
short

22
src/service/rooms/state/data.rs
View file

@ -1,30 +1,28 @@
use std::sync::Arc;
use std::{sync::MutexGuard, collections::HashSet};
use std::fmt::Debug;
use crate::Result;
use ruma::{EventId, RoomId};
pub trait Data {
/// Returns the last state hash key added to the db for the given room.
fn get_room_shortstatehash(room_id: &RoomId);
fn get_room_shortstatehash(&self, room_id: &RoomId) -> Result<Option<u64>>;
/// Update the current state of the room.
fn set_room_state(room_id: &RoomId, new_shortstatehash: u64,
_mutex_lock: &MutexGuard<'_, StateLock>, // Take mutex guard to make sure users get the room state mutex
);
fn set_room_state(&self, room_id: &RoomId, new_shortstatehash: u64,
_mutex_lock: &MutexGuard<'_, ()>, // Take mutex guard to make sure users get the room state mutex
) -> Result<()>;
/// Associates a state with an event.
fn set_event_state(shorteventid: u64, shortstatehash: u64) -> Result<()>;
fn set_event_state(&self, shorteventid: u64, shortstatehash: u64) -> Result<()>;
/// Returns all events we would send as the prev_events of the next event.
fn get_forward_extremities(room_id: &RoomId) -> Result<HashSet<Arc<EventId>>>;
fn get_forward_extremities(&self, room_id: &RoomId) -> Result<HashSet<Arc<EventId>>>;
/// Replace the forward extremities of the room.
fn set_forward_extremities(
fn set_forward_extremities<'a>(&self,
room_id: &RoomId,
event_ids: impl IntoIterator<Item = &'_ EventId> + Debug,
_mutex_lock: &MutexGuard<'_, StateLock>, // Take mutex guard to make sure users get the room state mutex
event_ids: impl IntoIterator<Item = &'a EventId> + Debug,
_mutex_lock: &MutexGuard<'_, ()>, // Take mutex guard to make sure users get the room state mutex
) -> Result<()>;
}
pub struct StateLock;

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

@ -6,13 +6,15 @@ use ruma::{RoomId, events::{room::{member::MembershipState, create::RoomCreateEv
use serde::Deserialize;
use tracing::warn;
use crate::{service::*, SERVICE, PduEvent, Error, utils::calculate_hash};
use crate::{Result, services, PduEvent, Error, utils::calculate_hash};
use super::state_compressor::CompressedStateEvent;
pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
impl<D: Data> Service<D> {
/// Set the room to the given statehash and update caches.
pub fn force_state(
&self,
@ -23,11 +25,11 @@ impl Service<_> {
) -> Result<()> {
for event_id in statediffnew.into_iter().filter_map(|new| {
SERVICE.rooms.state_compressor.parse_compressed_state_event(new)
services().rooms.state_compressor.parse_compressed_state_event(new)
.ok()
.map(|(_, id)| id)
}) {
let pdu = match SERVICE.rooms.timeline.get_pdu_json(&event_id)? {
let pdu = match services().rooms.timeline.get_pdu_json(&event_id)? {
Some(pdu) => pdu,
None => continue,
};
@ -63,10 +65,10 @@ impl Service<_> {
Err(_) => continue,
};
SERVICE.room.state_cache.update_membership(room_id, &user_id, membership, &pdu.sender, None, false)?;
services().room.state_cache.update_membership(room_id, &user_id, membership, &pdu.sender, None, false)?;
}
SERVICE.room.state_cache.update_joined_count(room_id)?;
services().room.state_cache.update_joined_count(room_id)?;
self.db.set_room_state(room_id, shortstatehash);
@ -84,7 +86,7 @@ impl Service<_> {
room_id: &RoomId,
state_ids_compressed: HashSet<CompressedStateEvent>,
) -> Result<()> {
let shorteventid = SERVICE.short.get_or_create_shorteventid(event_id)?;
let shorteventid = services().short.get_or_create_shorteventid(event_id)?;
let previous_shortstatehash = self.db.get_room_shortstatehash(room_id)?;
@ -96,11 +98,11 @@ impl Service<_> {
);
let (shortstatehash, already_existed) =
SERVICE.short.get_or_create_shortstatehash(&state_hash)?;
services().short.get_or_create_shortstatehash(&state_hash)?;
if !already_existed {
let states_parents = previous_shortstatehash
.map_or_else(|| Ok(Vec::new()), |p| SERVICE.room.state_compressor.load_shortstatehash_info(p))?;
.map_or_else(|| Ok(Vec::new()), |p| services().room.state_compressor.load_shortstatehash_info(p))?;
let (statediffnew, statediffremoved) =
if let Some(parent_stateinfo) = states_parents.last() {
@ -119,7 +121,7 @@ impl Service<_> {
} else {
(state_ids_compressed, HashSet::new())
};
SERVICE.room.state_compressor.save_state_from_diff(
services().room.state_compressor.save_state_from_diff(
shortstatehash,
statediffnew,
statediffremoved,
@ -176,7 +178,7 @@ impl Service<_> {
}
// TODO: statehash with deterministic inputs
let shortstatehash = SERVICE.globals.next_count()?;
let shortstatehash = services().globals.next_count()?;
let mut statediffnew = HashSet::new();
statediffnew.insert(new);
@ -273,4 +275,8 @@ impl Service<_> {
.ok_or_else(|| Error::BadDatabase("Invalid room version"))?;
Ok(room_version)
}
pub fn get_room_shortstatehash(&self, room_id: &RoomId) -> Result<Option<u64>> {
self.db.get_room_shortstatehash(room_id)
}
}

6
src/service/rooms/state_accessor/data.rs
View file

@ -1,9 +1,11 @@
use std::{sync::Arc, collections::HashMap};
use std::{sync::Arc, collections::{HashMap, BTreeMap}};
use async_trait::async_trait;
use ruma::{EventId, events::StateEventType, RoomId};
use crate::PduEvent;
use crate::{Result, PduEvent};
#[async_trait]
pub trait Data {
/// Builds a StateMap by iterating over all keys that start
/// with state_hash, this gives the full state for the given state_hash.

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

@ -4,13 +4,13 @@ use std::{sync::Arc, collections::{HashMap, BTreeMap}};
pub use data::Data;
use ruma::{events::StateEventType, RoomId, EventId};
use crate::{service::*, PduEvent};
use crate::{Result, PduEvent};
pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
impl<D: Data> Service<D> {
/// Builds a StateMap by iterating over all keys that start
/// with state_hash, this gives the full state for the given state_hash.
#[tracing::instrument(skip(self))]

8
src/service/rooms/state_cache/data.rs
View file

@ -1,5 +1,9 @@
use ruma::{UserId, RoomId};
use ruma::{UserId, RoomId, serde::Raw, events::AnyStrippedStateEvent};
use crate::Result;
pub trait Data {
fn mark_as_once_joined(user_id: &UserId, room_id: &RoomId) -> Result<()>;
fn mark_as_once_joined(&self, user_id: &UserId, room_id: &RoomId) -> Result<()>;
fn mark_as_joined(&self, user_id: &UserId, room_id: &RoomId) -> Result<()>;
fn mark_as_invited(&self, user_id: &UserId, room_id: &RoomId, last_state: Option<Vec<Raw<AnyStrippedStateEvent>>>) -> Result<()>;
fn mark_as_left(&self, user_id: &UserId, room_id: &RoomId) -> Result<()>;
}

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

@ -5,13 +5,13 @@ pub use data::Data;
use regex::Regex;
use ruma::{RoomId, UserId, events::{room::{member::MembershipState, create::RoomCreateEventContent}, AnyStrippedStateEvent, StateEventType, tag::TagEvent, RoomAccountDataEventType, GlobalAccountDataEventType, direct::DirectEvent, ignored_user_list::IgnoredUserListEvent, AnySyncStateEvent}, serde::Raw, ServerName};
use crate::{service::*, SERVICE, utils, Error};
use crate::{Result, services, utils, Error};
pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
impl<D: Data> Service<D> {
/// Update current membership data.
#[tracing::instrument(skip(self, last_state))]
pub fn update_membership(
@ -24,8 +24,8 @@ impl Service<_> {
update_joined_count: bool,
) -> Result<()> {
// Keep track what remote users exist by adding them as "deactivated" users
if user_id.server_name() != SERVICE.globals.server_name() {
SERVICE.users.create(user_id, None)?;
if user_id.server_name() != services().globals.server_name() {
services().users.create(user_id, None)?;
// TODO: displayname, avatar url
}
@ -37,10 +37,6 @@ impl Service<_> {
serverroom_id.push(0xff);
serverroom_id.extend_from_slice(room_id.as_bytes());
let mut roomuser_id = room_id.as_bytes().to_vec();
roomuser_id.push(0xff);
roomuser_id.extend_from_slice(user_id.as_bytes());
match &membership {
MembershipState::Join => {
// Check if the user never joined this room
@ -80,24 +76,23 @@ impl Service<_> {
// .ok();
// Copy old tags to new room
if let Some(tag_event) = db.account_data.get::<TagEvent>(
if let Some(tag_event) = services().account_data.get::<TagEvent>(
Some(&predecessor.room_id),
user_id,
RoomAccountDataEventType::Tag,
)? {
SERVICE.account_data
services().account_data
.update(
Some(room_id),
user_id,
RoomAccountDataEventType::Tag,
&tag_event,
&db.globals,
)
.ok();
};
// Copy direct chat flag
if let Some(mut direct_event) = SERVICE.account_data.get::<DirectEvent>(
if let Some(mut direct_event) = services().account_data.get::<DirectEvent>(
None,
user_id,
GlobalAccountDataEventType::Direct.to_string().into(),
@ -112,7 +107,7 @@ impl Service<_> {
}
if room_ids_updated {
SERVICE.account_data.update(
services().account_data.update(
None,
user_id,
GlobalAccountDataEventType::Direct.to_string().into(),
@ -123,16 +118,11 @@ impl Service<_> {
}
}
self.userroomid_joined.insert(&userroom_id, &[])?;
self.roomuserid_joined.insert(&roomuser_id, &[])?;
self.userroomid_invitestate.remove(&userroom_id)?;
self.roomuserid_invitecount.remove(&roomuser_id)?;
self.userroomid_leftstate.remove(&userroom_id)?;
self.roomuserid_leftcount.remove(&roomuser_id)?;
self.db.mark_as_joined(user_id, room_id)?;
}
MembershipState::Invite => {
// We want to know if the sender is ignored by the receiver
let is_ignored = SERVICE
let is_ignored = services()
.account_data
.get::<IgnoredUserListEvent>(
None, // Ignored users are in global account data
@ -153,41 +143,22 @@ impl Service<_> {
return Ok(());
}
self.userroomid_invitestate.insert(
&userroom_id,
&serde_json::to_vec(&last_state.unwrap_or_default())
.expect("state to bytes always works"),
)?;
self.roomuserid_invitecount
.insert(&roomuser_id, &db.globals.next_count()?.to_be_bytes())?;
self.userroomid_joined.remove(&userroom_id)?;
self.roomuserid_joined.remove(&roomuser_id)?;
self.userroomid_leftstate.remove(&userroom_id)?;
self.roomuserid_leftcount.remove(&roomuser_id)?;
self.db.mark_as_invited(user_id, room_id, last_state)?;
}
MembershipState::Leave | MembershipState::Ban => {
self.userroomid_leftstate.insert(
&userroom_id,
&serde_json::to_vec(&Vec::<Raw<AnySyncStateEvent>>::new()).unwrap(),
)?; // TODO
self.roomuserid_leftcount
.insert(&roomuser_id, &db.globals.next_count()?.to_be_bytes())?;
self.userroomid_joined.remove(&userroom_id)?;
self.roomuserid_joined.remove(&roomuser_id)?;
self.userroomid_invitestate.remove(&userroom_id)?;
self.roomuserid_invitecount.remove(&roomuser_id)?;
self.db.mark_as_left(user_id, room_id)?;
}
_ => {}
}
if update_joined_count {
self.update_joined_count(room_id, db)?;
self.update_joined_count(room_id)?;
}
Ok(())
}
#[tracing::instrument(skip(self, room_id, db))]
#[tracing::instrument(skip(self, room_id))]
pub fn update_joined_count(&self, room_id: &RoomId) -> Result<()> {
let mut joinedcount = 0_u64;
let mut invitedcount = 0_u64;
@ -196,8 +167,8 @@ impl Service<_> {
for joined in self.room_members(room_id).filter_map(|r| r.ok()) {
joined_servers.insert(joined.server_name().to_owned());
if joined.server_name() == db.globals.server_name()
&& !db.users.is_deactivated(&joined).unwrap_or(true)
if joined.server_name() == services().globals.server_name()
&& !services().users.is_deactivated(&joined).unwrap_or(true)
{
real_users.insert(joined);
}
@ -285,7 +256,7 @@ impl Service<_> {
.get("sender_localpart")
.and_then(|string| string.as_str())
.and_then(|string| {
UserId::parse_with_server_name(string, SERVICE.globals.server_name()).ok()
UserId::parse_with_server_name(string, services().globals.server_name()).ok()
});
let in_room = bridge_user_id

7
src/service/rooms/state_compressor/data.rs
View file

@ -1,4 +1,5 @@
use crate::service::rooms::CompressedStateEvent;
use super::CompressedStateEvent;
use crate::Result;
pub struct StateDiff {
parent: Option<u64>,
@ -7,6 +8,6 @@ pub struct StateDiff {
}
pub trait Data {
fn get_statediff(shortstatehash: u64) -> Result<StateDiff>;
fn save_statediff(shortstatehash: u64, diff: StateDiff) -> Result<()>;
fn get_statediff(&self, shortstatehash: u64) -> Result<StateDiff>;
fn save_statediff(&self, shortstatehash: u64, diff: StateDiff) -> Result<()>;
}

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

@ -4,7 +4,7 @@ use std::{mem::size_of, sync::Arc, collections::HashSet};
pub use data::Data;
use ruma::{EventId, RoomId};
use crate::{service::*, utils};
use crate::{Result, utils, services};
use self::data::StateDiff;
@ -12,7 +12,9 @@ pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
pub type CompressedStateEvent = [u8; 2 * size_of::<u64>()];
impl<D: Data> Service<D> {
/// Returns a stack with info on shortstatehash, full state, added diff and removed diff for the selected shortstatehash and each parent layer.
#[tracing::instrument(skip(self))]
pub fn load_shortstatehash_info(
@ -62,12 +64,11 @@ impl Service<_> {
&self,
shortstatekey: u64,
event_id: &EventId,
globals: &super::globals::Globals,
) -> Result<CompressedStateEvent> {
let mut v = shortstatekey.to_be_bytes().to_vec();
v.extend_from_slice(
&self
.get_or_create_shorteventid(event_id, globals)?
.get_or_create_shorteventid(event_id)?
.to_be_bytes(),
);
Ok(v.try_into().expect("we checked the size above"))
@ -210,15 +211,16 @@ impl Service<_> {
/// Returns the new shortstatehash
pub fn save_state(
&self,
room_id: &RoomId,
new_state_ids_compressed: HashSet<CompressedStateEvent>,
) -> Result<(u64,
HashSet<CompressedStateEvent>, // added
HashSet<CompressedStateEvent>)> // removed
{
let previous_shortstatehash = self.d.current_shortstatehash(room_id)?;
let previous_shortstatehash = self.db.current_shortstatehash(room_id)?;
let state_hash = self.calculate_hash(
let state_hash = utils::calculate_hash(
&new_state_ids_compressed
.iter()
.map(|bytes| &bytes[..])
@ -226,7 +228,7 @@ impl Service<_> {
);
let (new_shortstatehash, already_existed) =
self.get_or_create_shortstatehash(&state_hash, &db.globals)?;
services().rooms.short.get_or_create_shortstatehash(&state_hash)?;
if Some(new_shortstatehash) == previous_shortstatehash {
return Ok(());

10
src/service/rooms/timeline/data.rs
View file

@ -2,7 +2,7 @@ use std::sync::Arc;
use ruma::{signatures::CanonicalJsonObject, EventId, UserId, RoomId};
use crate::PduEvent;
use crate::{Result, PduEvent};
pub trait Data {
fn last_timeline_count(&self, sender_user: &UserId, room_id: &RoomId) -> Result<u64>;
@ -48,28 +48,26 @@ pub trait Data {
/// Returns an iterator over all events in a room that happened after the event with id `since`
/// in chronological order.
#[tracing::instrument(skip(self))]
fn pdus_since<'a>(
&'a self,
user_id: &UserId,
room_id: &RoomId,
since: u64,
) -> Result<impl Iterator<Item = Result<(Vec<u8>, PduEvent)>> + 'a>;
) -> Result<Box<dyn Iterator<Item = Result<(Vec<u8>, PduEvent)>>>>;
/// Returns an iterator over all events and their tokens in a room that happened before the
/// event with id `until` in reverse-chronological order.
#[tracing::instrument(skip(self))]
fn pdus_until<'a>(
&'a self,
user_id: &UserId,
room_id: &RoomId,
until: u64,
) -> Result<impl Iterator<Item = Result<(Vec<u8>, PduEvent)>> + 'a>;
) -> Result<Box<dyn Iterator<Item = Result<(Vec<u8>, PduEvent)>>>>;
fn pdus_after<'a>(
&'a self,
user_id: &UserId,
room_id: &RoomId,
from: u64,
) -> Result<impl Iterator<Item = Result<(Vec<u8>, PduEvent)>> + 'a>;
) -> Result<Box<dyn Iterator<Item = Result<(Vec<u8>, PduEvent)>>>>;
}

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

@ -1,23 +1,29 @@
mod data;
use std::borrow::Cow;
use std::sync::Arc;
use std::{sync::MutexGuard, iter, collections::HashSet};
use std::fmt::Debug;
pub use data::Data;
use regex::Regex;
use ruma::events::room::power_levels::RoomPowerLevelsEventContent;
use ruma::push::Ruleset;
use ruma::signatures::CanonicalJsonValue;
use ruma::state_res::RoomVersion;
use ruma::{EventId, signatures::CanonicalJsonObject, push::{Action, Tweak}, events::{push_rules::PushRulesEvent, GlobalAccountDataEventType, RoomEventType, room::{member::MembershipState, create::RoomCreateEventContent}, StateEventType}, UserId, RoomAliasId, RoomId, uint, state_res, api::client::error::ErrorKind, serde::to_canonical_value, ServerName};
use serde::Deserialize;
use serde_json::value::to_raw_value;
use tracing::{warn, error};
use crate::SERVICE;
use crate::{service::{*, pdu::{PduBuilder, EventHash}}, Error, PduEvent, utils};
use crate::{services, Result, service::pdu::{PduBuilder, EventHash}, Error, PduEvent, utils};
use super::state_compressor::CompressedStateEvent;
pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
impl<D: Data> Service<D> {
/*
/// Checks if a room exists.
#[tracing::instrument(skip(self))]
@ -44,7 +50,7 @@ impl Service<_> {
#[tracing::instrument(skip(self))]
pub fn last_timeline_count(&self, sender_user: &UserId, room_id: &RoomId) -> Result<u64> {
self.db.last_timeline_count(sender_user: &UserId, room_id: &RoomId)
self.db.last_timeline_count(sender_user, room_id)
}
// TODO Is this the same as the function above?
@ -127,7 +133,7 @@ impl Service<_> {
/// Removes a pdu and creates a new one with the same id.
#[tracing::instrument(skip(self))]
fn replace_pdu(&self, pdu_id: &[u8], pdu: &PduEvent) -> Result<()> {
self.db.pdu_count(pdu_id, pdu: &PduEvent)
self.db.replace_pdu(pdu_id, pdu)
}
/// Creates a new persisted data unit and adds it to a room.
@ -177,7 +183,7 @@ impl Service<_> {
self.replace_pdu_leaves(&pdu.room_id, leaves)?;
let mutex_insert = Arc::clone(
db.globals
services().globals
.roomid_mutex_insert
.write()
.unwrap()
@ -186,14 +192,14 @@ impl Service<_> {
);
let insert_lock = mutex_insert.lock().unwrap();
let count1 = db.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
// fails
self.edus
.private_read_set(&pdu.room_id, &pdu.sender, count1, &db.globals)?;
.private_read_set(&pdu.room_id, &pdu.sender, count1)?;
self.reset_notification_counts(&pdu.sender, &pdu.room_id)?;
let count2 = db.globals.next_count()?;
let count2 = services().globals.next_count()?;
let mut pdu_id = shortroomid.to_be_bytes().to_vec();
pdu_id.extend_from_slice(&count2.to_be_bytes());
@ -218,7 +224,7 @@ impl Service<_> {
drop(insert_lock);
// See if the event matches any known pushers
let power_levels: RoomPowerLevelsEventContent = db
let power_levels: RoomPowerLevelsEventContent = services()
.rooms
.room_state_get(&pdu.room_id, &StateEventType::RoomPowerLevels, "")?
.map(|ev| {
@ -233,13 +239,13 @@ impl Service<_> {
let mut notifies = Vec::new();
let mut highlights = Vec::new();
for user in self.get_our_real_users(&pdu.room_id, db)?.iter() {
for user in self.get_our_real_users(&pdu.room_id)?.iter() {
// Don't notify the user of their own events
if user == &pdu.sender {
continue;
}
let rules_for_user = db
let rules_for_user = services()
.account_data
.get(
None,
@ -252,7 +258,7 @@ impl Service<_> {
let mut highlight = false;
let mut notify = false;
for action in pusher::get_actions(
for action in services().pusher.get_actions(
user,
&rules_for_user,
&power_levels,
@ -282,8 +288,8 @@ impl Service<_> {
highlights.push(userroom_id);
}
for senderkey in db.pusher.get_pusher_senderkeys(user) {
db.sending.send_push_pdu(&*pdu_id, senderkey)?;
for senderkey in services().pusher.get_pusher_senderkeys(user) {
services().sending.send_push_pdu(&*pdu_id, senderkey)?;
}
}
@ -328,7 +334,6 @@ impl Service<_> {
content.membership,
&pdu.sender,
invite_state,
db,
true,
)?;
}
@ -344,34 +349,34 @@ impl Service<_> {
.map_err(|_| Error::bad_database("Invalid content in pdu."))?;
if let Some(body) = content.body {
DB.rooms.search.index_pdu(room_id, pdu_id, body)?;
services().rooms.search.index_pdu(shortroomid, pdu_id, body)?;
let admin_room = self.id_from_alias(
let admin_room = self.alias.resolve_local_alias(
<&RoomAliasId>::try_from(
format!("#admins:{}", db.globals.server_name()).as_str(),
format!("#admins:{}", services().globals.server_name()).as_str(),
)
.expect("#admins:server_name is a valid room alias"),
)?;
let server_user = format!("@conduit:{}", db.globals.server_name());
let server_user = format!("@conduit:{}", services().globals.server_name());
let to_conduit = body.starts_with(&format!("{}: ", server_user));
// This will evaluate to false if the emergency password is set up so that
// the administrator can execute commands as conduit
let from_conduit =
pdu.sender == server_user && db.globals.emergency_password().is_none();
pdu.sender == server_user && services().globals.emergency_password().is_none();
if to_conduit && !from_conduit && admin_room.as_ref() == Some(&pdu.room_id) {
db.admin.process_message(body.to_string());
services().admin.process_message(body.to_string());
}
}
}
_ => {}
}
for appservice in db.appservice.all()? {
if self.appservice_in_room(room_id, &appservice, db)? {
db.sending.send_pdu_appservice(&appservice.0, &pdu_id)?;
for appservice in services().appservice.all()? {
if self.appservice_in_room(&pdu.room_id, &appservice)? {
services().sending.send_pdu_appservice(&appservice.0, &pdu_id)?;
continue;
}
@ -388,11 +393,11 @@ impl Service<_> {
.get("sender_localpart")
.and_then(|string| string.as_str())
.and_then(|string| {
UserId::parse_with_server_name(string, db.globals.server_name()).ok()
UserId::parse_with_server_name(string, services().globals.server_name()).ok()
})
{
if state_key_uid == &appservice_uid {
db.sending.send_pdu_appservice(&appservice.0, &pdu_id)?;
services().sending.send_pdu_appservice(&appservice.0, &pdu_id)?;
continue;
}
}
@ -431,16 +436,16 @@ impl Service<_> {
.map_or(false, |state_key| users.is_match(state_key))
};
let matching_aliases = |aliases: &Regex| {
self.room_aliases(room_id)
self.room_aliases(&pdu.room_id)
.filter_map(|r| r.ok())
.any(|room_alias| aliases.is_match(room_alias.as_str()))
};
if aliases.iter().any(matching_aliases)
|| rooms.map_or(false, |rooms| rooms.contains(&room_id.as_str().into()))
|| rooms.map_or(false, |rooms| rooms.contains(&pdu.room_id.as_str().into()))
|| users.iter().any(matching_users)
{
db.sending.send_pdu_appservice(&appservice.0, &pdu_id)?;
services().sending.send_pdu_appservice(&appservice.0, &pdu_id)?;
}
}
}
@ -464,14 +469,14 @@ impl Service<_> {
redacts,
} = pdu_builder;
let prev_events: Vec<_> = SERVICE
let prev_events: Vec<_> = services()
.rooms
.get_pdu_leaves(room_id)?
.into_iter()
.take(20)
.collect();
let create_event = SERVICE
let create_event = services()
.rooms
.room_state_get(room_id, &StateEventType::RoomCreate, "")?;
@ -488,7 +493,7 @@ impl Service<_> {
// If there was no create event yet, assume we are creating a room with the default
// version right now
let room_version_id = create_event_content
.map_or(SERVICE.globals.default_room_version(), |create_event| {
.map_or(services().globals.default_room_version(), |create_event| {
create_event.room_version
});
let room_version =
@ -500,7 +505,7 @@ impl Service<_> {
// Our depth is the maximum depth of prev_events + 1
let depth = prev_events
.iter()
.filter_map(|event_id| Some(db.rooms.get_pdu(event_id).ok()??.depth))
.filter_map(|event_id| Some(services().rooms.get_pdu(event_id).ok()??.depth))
.max()
.unwrap_or_else(|| uint!(0))
+ uint!(1);
@ -525,7 +530,7 @@ impl Service<_> {
let pdu = PduEvent {
event_id: ruma::event_id!("$thiswillbefilledinlater").into(),
room_id: room_id.to_owned(),
sender: sender_user.to_owned(),
sender: sender.to_owned(),
origin_server_ts: utils::millis_since_unix_epoch()
.try_into()
.expect("time is valid"),
@ -577,13 +582,13 @@ impl Service<_> {
// Add origin because synapse likes that (and it's required in the spec)
pdu_json.insert(
"origin".to_owned(),
to_canonical_value(db.globals.server_name())
to_canonical_value(services().globals.server_name())
.expect("server name is a valid CanonicalJsonValue"),
);
match ruma::signatures::hash_and_sign_event(
SERVICE.globals.server_name().as_str(),
SERVICE.globals.keypair(),
services().globals.server_name().as_str(),
services().globals.keypair(),
&mut pdu_json,
&room_version_id,
) {
@ -616,22 +621,20 @@ impl Service<_> {
);
// Generate short event id
let _shorteventid = self.get_or_create_shorteventid(&pdu.event_id, &db.globals)?;
let _shorteventid = self.get_or_create_shorteventid(&pdu.event_id)?;
}
/// 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.
#[tracing::instrument(skip(self, _mutex_lock))]
#[tracing::instrument(skip(self, state_lock))]
pub fn build_and_append_pdu(
&self,
pdu_builder: PduBuilder,
sender: &UserId,
room_id: &RoomId,
_mutex_lock: &MutexGuard<'_, ()>, // Take mutex guard to make sure users get the room state mutex
state_lock: &MutexGuard<'_, ()>, // Take mutex guard to make sure users get the room state mutex
) -> Result<Arc<EventId>> {
let (pdu, pdu_json) = self.create_hash_and_sign_event()?;
let (pdu, pdu_json) = self.create_hash_and_sign_event(pdu_builder, sender, room_id, &state_lock);
// We append to state before appending the pdu, so we don't have a moment in time with the
// pdu without it's state. This is okay because append_pdu can't fail.
@ -664,9 +667,9 @@ impl Service<_> {
}
// Remove our server from the server list since it will be added to it by room_servers() and/or the if statement above
servers.remove(SERVICE.globals.server_name());
servers.remove(services().globals.server_name());
SERVICE.sending.send_pdu(servers.into_iter(), &pdu_id)?;
services().sending.send_pdu(servers.into_iter(), &pdu_id)?;
Ok(pdu.event_id)
}
@ -684,20 +687,20 @@ impl Service<_> {
) -> Result<Option<Vec<u8>>> {
// We append to state before appending the pdu, so we don't have a moment in time with the
// pdu without it's state. This is okay because append_pdu can't fail.
SERVICE.rooms.set_event_state(
services().rooms.set_event_state(
&pdu.event_id,
&pdu.room_id,
state_ids_compressed,
)?;
if soft_fail {
SERVICE.rooms
services().rooms
.mark_as_referenced(&pdu.room_id, &pdu.prev_events)?;
SERVICE.rooms.replace_pdu_leaves(&pdu.room_id, new_room_leaves)?;
services().rooms.replace_pdu_leaves(&pdu.room_id, new_room_leaves)?;
return Ok(None);
}
let pdu_id = SERVICE.rooms.append_pdu(pdu, pdu_json, new_room_leaves)?;
let pdu_id = services().rooms.append_pdu(pdu, pdu_json, new_room_leaves)?;
Ok(Some(pdu_id))
}

5
src/service/rooms/user/data.rs
View file

@ -1,3 +1,6 @@
use ruma::{UserId, RoomId};
use crate::Result;
pub trait Data {
fn reset_notification_counts(&self, user_id: &UserId, room_id: &RoomId) -> Result<()>;
@ -17,5 +20,5 @@ pub trait Data {
fn get_shared_rooms<'a>(
&'a self,
users: Vec<Box<UserId>>,
) -> Result<impl Iterator<Item = Result<Box<RoomId>>> + 'a>;
) -> Result<Box<dyn Iterator<Item = Result<Box<RoomId>>>>>;
}

6
src/service/rooms/user/mod.rs
View file

@ -2,13 +2,13 @@ mod data;
pub use data::Data;
use ruma::{RoomId, UserId};
use crate::service::*;
use crate::Result;
pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
impl<D: Data> Service<D> {
pub fn reset_notification_counts(&self, user_id: &UserId, room_id: &RoomId) -> Result<()> {
self.db.reset_notification_counts(user_id, room_id)
}
@ -27,7 +27,7 @@ impl Service<_> {
token: u64,
shortstatehash: u64,
) -> Result<()> {
self.db.associate_token_shortstatehash(user_id, room_id)
self.db.associate_token_shortstatehash(room_id, token, shortstatehash)
}
pub fn get_token_shortstatehash(&self, room_id: &RoomId, token: u64) -> Result<Option<u64>> {

112
src/service/sending.rs → src/service/sending/mod.rs
View file

@ -6,7 +6,7 @@ use std::{
};
use crate::{
appservice_server, database::pusher, server_server, utils, Database, Error, PduEvent, Result,
utils, Error, PduEvent, Result, services, api::{server_server, appservice_server},
};
use federation::transactions::send_transaction_message;
use futures_util::{stream::FuturesUnordered, StreamExt};
@ -34,8 +34,6 @@ use tokio::{
};
use tracing::{error, warn};
use super::abstraction::Tree;
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub enum OutgoingKind {
Appservice(String),
@ -77,11 +75,8 @@ pub enum SendingEventType {
Edu(Vec<u8>),
}
pub struct Sending {
pub struct Service {
/// The state for a given state hash.
pub(super) servername_educount: Arc<dyn Tree>, // EduCount: Count of last EDU sync
pub(super) servernameevent_data: Arc<dyn Tree>, // ServernameEvent = (+ / $)SenderKey / ServerName / UserId + PduId / Id (for edus), Data = EDU content
pub(super) servercurrentevent_data: Arc<dyn Tree>, // ServerCurrentEvents = (+ / $)ServerName / UserId + PduId / Id (for edus), Data = EDU content
pub(super) maximum_requests: Arc<Semaphore>,
pub sender: mpsc::UnboundedSender<(Vec<u8>, Vec<u8>)>,
}
@ -92,10 +87,9 @@ enum TransactionStatus {
Retrying(u32), // number of times failed
}
impl Sending {
impl Service {
pub fn start_handler(
&self,
db: Arc<RwLock<Database>>,
mut receiver: mpsc::UnboundedReceiver<(Vec<u8>, Vec<u8>)>,
) {
tokio::spawn(async move {
@ -106,9 +100,7 @@ impl Sending {
// Retry requests we could not finish yet
let mut initial_transactions = HashMap::<OutgoingKind, Vec<SendingEventType>>::new();
let guard = db.read().await;
for (key, outgoing_kind, event) in guard
for (key, outgoing_kind, event) in services()
.sending
.servercurrentevent_data
.iter()
@ -127,22 +119,19 @@ impl Sending {
"Dropping some current events: {:?} {:?} {:?}",
key, outgoing_kind, event
);
guard.sending.servercurrentevent_data.remove(&key).unwrap();
services().sending.servercurrentevent_data.remove(&key).unwrap();
continue;
}
entry.push(event);
}
drop(guard);
for (outgoing_kind, events) in initial_transactions {
current_transaction_status
.insert(outgoing_kind.get_prefix(), TransactionStatus::Running);
futures.push(Self::handle_events(
outgoing_kind.clone(),
events,
Arc::clone(&db),
));
}
@ -151,17 +140,15 @@ impl Sending {
Some(response) = futures.next() => {
match response {
Ok(outgoing_kind) => {
let guard = db.read().await;
let prefix = outgoing_kind.get_prefix();
for (key, _) in guard.sending.servercurrentevent_data
for (key, _) in services().sending.servercurrentevent_data
.scan_prefix(prefix.clone())
{
guard.sending.servercurrentevent_data.remove(&key).unwrap();
services().sending.servercurrentevent_data.remove(&key).unwrap();
}
// Find events that have been added since starting the last request
let new_events: Vec<_> = guard.sending.servernameevent_data
let new_events: Vec<_> = services().sending.servernameevent_data
.scan_prefix(prefix.clone())
.filter_map(|(k, v)| {
Self::parse_servercurrentevent(&k, v).ok().map(|ev| (ev, k))
@ -175,17 +162,14 @@ impl Sending {
// Insert pdus we found
for (e, key) in &new_events {
let value = if let SendingEventType::Edu(value) = &e.1 { &**value } else { &[] };
guard.sending.servercurrentevent_data.insert(key, value).unwrap();
guard.sending.servernameevent_data.remove(key).unwrap();
services().sending.servercurrentevent_data.insert(key, value).unwrap();
services().sending.servernameevent_data.remove(key).unwrap();
}
drop(guard);
futures.push(
Self::handle_events(
outgoing_kind.clone(),
new_events.into_iter().map(|(event, _)| event.1).collect(),
Arc::clone(&db),
)
);
} else {
@ -206,15 +190,12 @@ impl Sending {
},
Some((key, value)) = receiver.recv() => {
if let Ok((outgoing_kind, event)) = Self::parse_servercurrentevent(&key, value) {
let guard = db.read().await;
if let Ok(Some(events)) = Self::select_events(
&outgoing_kind,
vec![(event, key)],
&mut current_transaction_status,
&guard
) {
futures.push(Self::handle_events(outgoing_kind, events, Arc::clone(&db)));
futures.push(Self::handle_events(outgoing_kind, events));
}
}
}
@ -223,12 +204,11 @@ impl Sending {
});
}
#[tracing::instrument(skip(outgoing_kind, new_events, current_transaction_status, db))]
#[tracing::instrument(skip(outgoing_kind, new_events, current_transaction_status))]
fn select_events(
outgoing_kind: &OutgoingKind,
new_events: Vec<(SendingEventType, Vec<u8>)>, // Events we want to send: event and full key
current_transaction_status: &mut HashMap<Vec<u8>, TransactionStatus>,
db: &Database,
) -> Result<Option<Vec<SendingEventType>>> {
let mut retry = false;
let mut allow = true;
@ -266,7 +246,7 @@ impl Sending {
if retry {
// We retry the previous transaction
for (key, value) in db.sending.servercurrentevent_data.scan_prefix(prefix) {
for (key, value) in services().sending.servercurrentevent_data.scan_prefix(prefix) {
if let Ok((_, e)) = Self::parse_servercurrentevent(&key, value) {
events.push(e);
}
@ -278,22 +258,22 @@ impl Sending {
} else {
&[][..]
};
db.sending
services().sending
.servercurrentevent_data
.insert(&full_key, value)?;
// If it was a PDU we have to unqueue it
// TODO: don't try to unqueue EDUs
db.sending.servernameevent_data.remove(&full_key)?;
services().sending.servernameevent_data.remove(&full_key)?;
events.push(e);
}
if let OutgoingKind::Normal(server_name) = outgoing_kind {
if let Ok((select_edus, last_count)) = Self::select_edus(db, server_name) {
if let Ok((select_edus, last_count)) = Self::select_edus(server_name) {
events.extend(select_edus.into_iter().map(SendingEventType::Edu));
db.sending
services().sending
.servername_educount
.insert(server_name.as_bytes(), &last_count.to_be_bytes())?;
}
@ -303,10 +283,10 @@ impl Sending {
Ok(Some(events))
}
#[tracing::instrument(skip(db, server))]
pub fn select_edus(db: &Database, server: &ServerName) -> Result<(Vec<Vec<u8>>, u64)> {
#[tracing::instrument(skip(server))]
pub fn select_edus(server: &ServerName) -> Result<(Vec<Vec<u8>>, u64)> {
// u64: count of last edu
let since = db
let since = services()
.sending
.servername_educount
.get(server.as_bytes())?
@ -318,25 +298,25 @@ impl Sending {
let mut max_edu_count = since;
let mut device_list_changes = HashSet::new();
'outer: for room_id in db.rooms.server_rooms(server) {
'outer: for room_id in services().rooms.server_rooms(server) {
let room_id = room_id?;
// Look for device list updates in this room
device_list_changes.extend(
db.users
services().users
.keys_changed(&room_id.to_string(), since, None)
.filter_map(|r| r.ok())
.filter(|user_id| user_id.server_name() == db.globals.server_name()),
.filter(|user_id| user_id.server_name() == services().globals.server_name()),
);
// Look for read receipts in this room
for r in db.rooms.edus.readreceipts_since(&room_id, since) {
for r in services().rooms.edus.readreceipts_since(&room_id, since) {
let (user_id, count, read_receipt) = r?;
if count > max_edu_count {
max_edu_count = count;
}
if user_id.server_name() != db.globals.server_name() {
if user_id.server_name() != services().globals.server_name() {
continue;
}
@ -496,14 +476,11 @@ impl Sending {
Ok(())
}
#[tracing::instrument(skip(db, events, kind))]
#[tracing::instrument(skip(events, kind))]
async fn handle_events(
kind: OutgoingKind,
events: Vec<SendingEventType>,
db: Arc<RwLock<Database>>,
) -> Result<OutgoingKind, (OutgoingKind, Error)> {
let db = db.read().await;
match &kind {
OutgoingKind::Appservice(id) => {
let mut pdu_jsons = Vec::new();
@ -511,7 +488,7 @@ impl Sending {
for event in &events {
match event {
SendingEventType::Pdu(pdu_id) => {
pdu_jsons.push(db.rooms
pdu_jsons.push(services().rooms
.get_pdu_from_id(pdu_id)
.map_err(|e| (kind.clone(), e))?
.ok_or_else(|| {
@ -530,11 +507,10 @@ impl Sending {
}
}
let permit = db.sending.maximum_requests.acquire().await;
let permit = services().sending.maximum_requests.acquire().await;
let response = appservice_server::send_request(
&db.globals,
db.appservice
services().appservice
.get_registration(&id)
.map_err(|e| (kind.clone(), e))?
.ok_or_else(|| {
@ -576,7 +552,7 @@ impl Sending {
match event {
SendingEventType::Pdu(pdu_id) => {
pdus.push(
db.rooms
services().rooms
.get_pdu_from_id(pdu_id)
.map_err(|e| (kind.clone(), e))?
.ok_or_else(|| {
@ -624,7 +600,7 @@ impl Sending {
senderkey.push(0xff);
senderkey.extend_from_slice(pushkey);
let pusher = match db
let pusher = match services()
.pusher
.get_pusher(&senderkey)
.map_err(|e| (OutgoingKind::Push(user.clone(), pushkey.clone()), e))?
@ -633,7 +609,7 @@ impl Sending {
None => continue,
};
let rules_for_user = db
let rules_for_user = services()
.account_data
.get(
None,
@ -644,22 +620,21 @@ impl Sending {
.map(|ev: PushRulesEvent| ev.content.global)
.unwrap_or_else(|| push::Ruleset::server_default(&userid));
let unread: UInt = db
let unread: UInt = services()
.rooms
.notification_count(&userid, &pdu.room_id)
.map_err(|e| (kind.clone(), e))?
.try_into()
.expect("notifiation count can't go that high");
let permit = db.sending.maximum_requests.acquire().await;
let permit = services().sending.maximum_requests.acquire().await;
let _response = pusher::send_push_notice(
let _response = services().pusher.send_push_notice(
&userid,
unread,
&pusher,
rules_for_user,
&pdu,
&db,
)
.await
.map(|_response| kind.clone())
@ -678,7 +653,7 @@ impl Sending {
SendingEventType::Pdu(pdu_id) => {
// TODO: check room version and remove event_id if needed
let raw = PduEvent::convert_to_outgoing_federation_event(
db.rooms
services().rooms
.get_pdu_json_from_id(pdu_id)
.map_err(|e| (OutgoingKind::Normal(server.clone()), e))?
.ok_or_else(|| {
@ -700,13 +675,12 @@ impl Sending {
}
}
let permit = db.sending.maximum_requests.acquire().await;
let permit = services().sending.maximum_requests.acquire().await;
let response = server_server::send_request(
&db.globals,
&*server,
send_transaction_message::v1::Request {
origin: db.globals.server_name(),
origin: services().globals.server_name(),
pdus: &pdu_jsons,
edus: &edu_jsons,
origin_server_ts: MilliSecondsSinceUnixEpoch::now(),
@ -809,10 +783,9 @@ impl Sending {
})
}
#[tracing::instrument(skip(self, globals, destination, request))]
#[tracing::instrument(skip(self, destination, request))]
pub async fn send_federation_request<T: OutgoingRequest>(
&self,
globals: &crate::database::globals::Globals,
destination: &ServerName,
request: T,
) -> Result<T::IncomingResponse>
@ -820,16 +793,15 @@ impl Sending {
T: Debug,
{
let permit = self.maximum_requests.acquire().await;
let response = server_server::send_request(globals, destination, request).await;
let response = server_server::send_request(destination, request).await;
drop(permit);
response
}
#[tracing::instrument(skip(self, globals, registration, request))]
#[tracing::instrument(skip(self, registration, request))]
pub async fn send_appservice_request<T: OutgoingRequest>(
&self,
globals: &crate::database::globals::Globals,
registration: serde_yaml::Value,
request: T,
) -> Result<T::IncomingResponse>
@ -837,7 +809,7 @@ impl Sending {
T: Debug,
{
let permit = self.maximum_requests.acquire().await;
let response = appservice_server::send_request(globals, registration, request).await;
let response = appservice_server::send_request(registration, request).await;
drop(permit);
response

3
src/service/transaction_ids/data.rs
View file

@ -1,3 +1,6 @@
use ruma::{DeviceId, UserId, TransactionId};
use crate::Result;
pub trait Data {
fn add_txnid(
&self,

6
src/service/transaction_ids/mod.rs
View file

@ -1,14 +1,14 @@
mod data;
pub use data::Data;
use ruma::{UserId, DeviceId, TransactionId};
use crate::service::*;
use ruma::{UserId, DeviceId, TransactionId};
use crate::Result;
pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
impl<D: Data> Service<D> {
pub fn add_txnid(
&self,
user_id: &UserId,

1
src/service/uiaa/data.rs
View file

@ -1,4 +1,5 @@
use ruma::{api::client::uiaa::UiaaInfo, DeviceId, UserId, signatures::CanonicalJsonValue};
use crate::Result;
pub trait Data {
fn set_uiaa_request(

13
src/service/uiaa/mod.rs
View file

@ -1,15 +1,16 @@
mod data;
pub use data::Data;
use ruma::{api::client::{uiaa::{UiaaInfo, IncomingAuthData, IncomingPassword, AuthType}, error::ErrorKind}, DeviceId, UserId, signatures::CanonicalJsonValue};
use ruma::{api::client::{uiaa::{UiaaInfo, IncomingAuthData, IncomingPassword, AuthType, IncomingUserIdentifier}, error::ErrorKind}, DeviceId, UserId, signatures::CanonicalJsonValue};
use tracing::error;
use crate::{service::*, utils, Error, SERVICE};
use crate::{Result, utils, Error, services, api::client_server::SESSION_ID_LENGTH};
pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
impl<D: Data> Service<D> {
/// Creates a new Uiaa session. Make sure the session token is unique.
pub fn create(
&self,
@ -56,7 +57,7 @@ impl Service<_> {
..
}) => {
let username = match identifier {
UserIdOrLocalpart(username) => username,
IncomingUserIdentifier::UserIdOrLocalpart(username) => username,
_ => {
return Err(Error::BadRequest(
ErrorKind::Unrecognized,
@ -66,13 +67,13 @@ impl Service<_> {
};
let user_id =
UserId::parse_with_server_name(username.clone(), SERVICE.globals.server_name())
UserId::parse_with_server_name(username.clone(), services().globals.server_name())
.map_err(|_| {
Error::BadRequest(ErrorKind::InvalidParam, "User ID is invalid.")
})?;
// Check if password is correct
if let Some(hash) = SERVICE.users.password_hash(&user_id)? {
if let Some(hash) = services().users.password_hash(&user_id)? {
let hash_matches =
argon2::verify_encoded(&hash, password.as_bytes()).unwrap_or(false);

12
src/service/users/data.rs
View file

@ -1,8 +1,8 @@
use std::collections::BTreeMap;
use crate::Result;
use ruma::{UserId, DeviceId, DeviceKeyAlgorithm, DeviceKeyId, serde::Raw, encryption::{OneTimeKey, DeviceKeys, CrossSigningKey}, UInt, events::AnyToDeviceEvent, api::client::{device::Device, filter::IncomingFilterDefinition}, MxcUri};
trait Data {
pub trait Data {
/// Check if a user has an account on this homeserver.
fn exists(&self, user_id: &UserId) -> Result<bool>;
@ -16,7 +16,7 @@ trait Data {
fn find_from_token(&self, token: &str) -> Result<Option<(Box<UserId>, String)>>;
/// Returns an iterator over all users on this homeserver.
fn iter(&self) -> impl Iterator<Item = Result<Box<UserId>>> + '_;
fn iter(&self) -> Box<dyn Iterator<Item = Result<Box<UserId>>>>;
/// Returns a list of local users as list of usernames.
///
@ -69,7 +69,7 @@ trait Data {
fn all_device_ids<'a>(
&'a self,
user_id: &UserId,
) -> impl Iterator<Item = Result<Box<DeviceId>>> + 'a;
) -> Box<dyn Iterator<Item = Result<Box<DeviceId>>>>;
/// Replaces the access token of one device.
fn set_token(&self, user_id: &UserId, device_id: &DeviceId, token: &str) -> Result<()>;
@ -125,7 +125,7 @@ trait Data {
user_or_room_id: &str,
from: u64,
to: Option<u64>,
) -> impl Iterator<Item = Result<Box<UserId>>> + 'a;
) -> Box<dyn Iterator<Item = Result<Box<UserId>>>>;
fn mark_device_key_update(
&self,
@ -193,7 +193,7 @@ trait Data {
fn all_devices_metadata<'a>(
&'a self,
user_id: &UserId,
) -> impl Iterator<Item = Result<Device>> + 'a;
) -> Box<dyn Iterator<Item = Result<Device>>>;
/// Creates a new sync filter. Returns the filter id.
fn create_filter(

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

@ -2,15 +2,15 @@ mod data;
use std::{collections::BTreeMap, mem};
pub use data::Data;
use ruma::{UserId, MxcUri, DeviceId, DeviceKeyId, serde::Raw, encryption::{OneTimeKey, CrossSigningKey, DeviceKeys}, DeviceKeyAlgorithm, UInt, events::AnyToDeviceEvent, api::client::{device::Device, filter::IncomingFilterDefinition}};
use ruma::{UserId, MxcUri, DeviceId, DeviceKeyId, serde::Raw, encryption::{OneTimeKey, CrossSigningKey, DeviceKeys}, DeviceKeyAlgorithm, UInt, events::AnyToDeviceEvent, api::client::{device::Device, filter::IncomingFilterDefinition, error::ErrorKind}, RoomAliasId};
use crate::{service::*, Error};
use crate::{Result, Error, services};
pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
impl<D: Data> Service<D> {
/// Check if a user has an account on this homeserver.
pub fn exists(&self, user_id: &UserId) -> Result<bool> {
self.db.exists(user_id)
@ -22,19 +22,19 @@ impl Service<_> {
}
/// Check if a user is an admin
fn is_admin(
pub fn is_admin(
&self,
user_id: &UserId,
) -> Result<bool> {
let admin_room_alias_id = RoomAliasId::parse(format!("#admins:{}", globals.server_name()))
let admin_room_alias_id = RoomAliasId::parse(format!("#admins:{}", services().globals.server_name()))
.map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "Invalid alias."))?;
let admin_room_id = rooms.id_from_alias(&admin_room_alias_id)?.unwrap();
let admin_room_id = services().rooms.alias.resolve_local_alias(&admin_room_alias_id)?.unwrap();
rooms.is_joined(user_id, &admin_room_id)
services().rooms.state_cache.is_joined(user_id, &admin_room_id)
}
/// Create a new user account on this homeserver.
fn create(&self, user_id: &UserId, password: Option<&str>) -> Result<()> {
pub fn create(&self, user_id: &UserId, password: Option<&str>) -> Result<()> {
self.db.set_password(user_id, password)?;
Ok(())
}