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

abstract account-data deserializations for serde_json::from_elim

Browse source 
Signed-off-by: Jason Volk <jason@zemos.net>
This commit is contained in:
Jason Volk 2024-10-02 07:57:18 +00:00 • committed by strawberry
parent 48a767d52c
commit da34b43302
12 changed files with 133 additions and 210 deletions
Download patch file Download diff file Expand all files Collapse all files

9
src/admin/query/account_data.rs
View file

@ -1,9 +1,6 @@
use clap::Subcommand;
use conduit::Result;
use ruma::{
events::{room::message::RoomMessageEventContent, RoomAccountDataEventType},
RoomId, UserId,
};
use ruma::{events::room::message::RoomMessageEventContent, RoomId, UserId};
use crate::Command;
@ -25,7 +22,7 @@ pub(crate) enum AccountDataCommand {
/// Full user ID
user_id: Box<UserId>,
/// Account data event type
kind: RoomAccountDataEventType,
kind: String,
/// Optional room ID of the account data
room_id: Option<Box<RoomId>>,
},
@ -60,7 +57,7 @@ pub(super) async fn process(subcommand: AccountDataCommand, context: &Command<'_
let timer = tokio::time::Instant::now();
let results = services
.account_data
.get(room_id.as_deref(), &user_id, kind)
.get_raw(room_id.as_deref(), &user_id, &kind)
.await;
let query_time = timer.elapsed();

42
src/admin/user/commands.rs
View file

@ -501,20 +501,16 @@ pub(super) async fn put_room_tag(
) -> Result<RoomMessageEventContent> {
let user_id = parse_active_local_user_id(self.services, &user_id).await?;
let event = self
let mut tags_event = self
.services
.account_data
.get(Some(&room_id), &user_id, RoomAccountDataEventType::Tag)
.await;
let mut tags_event = event.map_or_else(
|_| TagEvent {
.get_room(&room_id, &user_id, RoomAccountDataEventType::Tag)
.await
.unwrap_or(TagEvent {
content: TagEventContent {
tags: BTreeMap::new(),
},
},
|e| serde_json::from_str(e.get()).expect("Bad account data in database for user {user_id}"),
);
});
tags_event
.content
@ -542,20 +538,16 @@ pub(super) async fn delete_room_tag(
) -> Result<RoomMessageEventContent> {
let user_id = parse_active_local_user_id(self.services, &user_id).await?;
let event = self
let mut tags_event = self
.services
.account_data
.get(Some(&room_id), &user_id, RoomAccountDataEventType::Tag)
.await;
let mut tags_event = event.map_or_else(
|_| TagEvent {
.get_room(&room_id, &user_id, RoomAccountDataEventType::Tag)
.await
.unwrap_or(TagEvent {
content: TagEventContent {
tags: BTreeMap::new(),
},
},
|e| serde_json::from_str(e.get()).expect("Bad account data in database for user {user_id}"),
);
});
tags_event.content.tags.remove(&tag.clone().into());
@ -578,20 +570,16 @@ pub(super) async fn delete_room_tag(
pub(super) async fn get_room_tags(&self, user_id: String, room_id: Box<RoomId>) -> Result<RoomMessageEventContent> {
let user_id = parse_active_local_user_id(self.services, &user_id).await?;
let event = self
let tags_event = self
.services
.account_data
.get(Some(&room_id), &user_id, RoomAccountDataEventType::Tag)
.await;
let tags_event = event.map_or_else(
|_| TagEvent {
.get_room(&room_id, &user_id, RoomAccountDataEventType::Tag)
.await
.unwrap_or(TagEvent {
content: TagEventContent {
tags: BTreeMap::new(),
},
},
|e| serde_json::from_str(e.get()).expect("Bad account data in database for user {user_id}"),
);
});
Ok(RoomMessageEventContent::notice_markdown(format!(
"```\n{:#?}\n```",

20
src/api/client/config.rs
View file

@ -58,18 +58,14 @@ pub(crate) async fn get_global_account_data_route(
) -> Result<get_global_account_data::v3::Response> {
let sender_user = body.sender_user.as_ref().expect("user is authenticated");
let event: Box<RawJsonValue> = services
let account_data: ExtractGlobalEventContent = services
.account_data
.get(None, sender_user, body.event_type.to_string().into())
.get_global(sender_user, body.event_type.clone())
.await
.map_err(|_| err!(Request(NotFound("Data not found."))))?;
let account_data = serde_json::from_str::<ExtractGlobalEventContent>(event.get())
.map_err(|_| Error::bad_database("Invalid account data event in db."))?
.content;
Ok(get_global_account_data::v3::Response {
account_data,
account_data: account_data.content,
})
}
@ -81,18 +77,14 @@ pub(crate) async fn get_room_account_data_route(
) -> Result<get_room_account_data::v3::Response> {
let sender_user = body.sender_user.as_ref().expect("user is authenticated");
let event: Box<RawJsonValue> = services
let account_data: ExtractRoomEventContent = services
.account_data
.get(Some(&body.room_id), sender_user, body.event_type.clone())
.get_room(&body.room_id, sender_user, body.event_type.clone())
.await
.map_err(|_| err!(Request(NotFound("Data not found."))))?;
let account_data = serde_json::from_str::<ExtractRoomEventContent>(event.get())
.map_err(|_| Error::bad_database("Invalid account data event in db."))?
.content;
Ok(get_room_account_data::v3::Response {
account_data,
account_data: account_data.content,
})
}

117
src/api/client/push.rs
View file

@ -13,7 +13,7 @@ use ruma::{
GlobalAccountDataEventType,
},
push::{InsertPushRuleError, RemovePushRuleError, Ruleset},
CanonicalJsonObject,
CanonicalJsonObject, CanonicalJsonValue,
};
use service::Services;
@ -27,38 +27,23 @@ pub(crate) async fn get_pushrules_all_route(
) -> Result<get_pushrules_all::v3::Response> {
let sender_user = body.sender_user.as_ref().expect("user is authenticated");
let global_ruleset: Ruleset;
let event = services
let Some(content_value) = services
.account_data
.get(None, sender_user, GlobalAccountDataEventType::PushRules.to_string().into())
.await;
let Ok(event) = event else {
.get_global::<CanonicalJsonObject>(sender_user, GlobalAccountDataEventType::PushRules)
.await
.ok()
.and_then(|event| event.get("content").cloned())
.filter(CanonicalJsonValue::is_object)
else {
// user somehow has non-existent push rule event. recreate it and return server
// default silently
return recreate_push_rules_and_return(&services, sender_user).await;
};
let value = serde_json::from_str::<CanonicalJsonObject>(event.get())
let account_data_content = serde_json::from_value::<PushRulesEventContent>(content_value.into())
.map_err(|e| err!(Database(warn!("Invalid push rules account data event in database: {e}"))))?;
let Some(content_value) = value.get("content") else {
// user somehow has a push rule event with no content key, recreate it and
// return server default silently
return recreate_push_rules_and_return(&services, sender_user).await;
};
if content_value.to_string().is_empty() {
// user somehow has a push rule event with empty content, recreate it and return
// server default silently
return recreate_push_rules_and_return(&services, sender_user).await;
}
let account_data_content = serde_json::from_value::<PushRulesEventContent>(content_value.clone().into())
.map_err(|e| err!(Database(warn!("Invalid push rules account data event in database: {e}"))))?;
global_ruleset = account_data_content.global;
let global_ruleset: Ruleset = account_data_content.global;
Ok(get_pushrules_all::v3::Response {
global: global_ruleset,
@ -73,17 +58,14 @@ pub(crate) async fn get_pushrule_route(
) -> Result<get_pushrule::v3::Response> {
let sender_user = body.sender_user.as_ref().expect("user is authenticated");
let event = services
let event: PushRulesEvent = services
.account_data
.get(None, sender_user, GlobalAccountDataEventType::PushRules.to_string().into())
.get_global(sender_user, GlobalAccountDataEventType::PushRules)
.await
.map_err(|_| Error::BadRequest(ErrorKind::NotFound, "PushRules event not found."))?;
.map_err(|_| err!(Request(NotFound("PushRules event not found."))))?;
let account_data = serde_json::from_str::<PushRulesEvent>(event.get())
.map_err(|_| Error::bad_database("Invalid account data event in db."))?
.content;
let rule = account_data
let rule = event
.content
.global
.get(body.kind.clone(), &body.rule_id)
.map(Into::into);
@ -113,14 +95,11 @@ pub(crate) async fn set_pushrule_route(
));
}
let event = services
let mut account_data: PushRulesEvent = services
.account_data
.get(None, sender_user, GlobalAccountDataEventType::PushRules.to_string().into())
.get_global(sender_user, GlobalAccountDataEventType::PushRules)
.await
.map_err(|_| Error::BadRequest(ErrorKind::NotFound, "PushRules event not found."))?;
let mut account_data = serde_json::from_str::<PushRulesEvent>(event.get())
.map_err(|_| Error::bad_database("Invalid account data event in db."))?;
.map_err(|_| err!(Request(NotFound("PushRules event not found."))))?;
if let Err(error) =
account_data
@ -181,21 +160,18 @@ pub(crate) async fn get_pushrule_actions_route(
));
}
let event = services
let event: PushRulesEvent = services
.account_data
.get(None, sender_user, GlobalAccountDataEventType::PushRules.to_string().into())
.get_global(sender_user, GlobalAccountDataEventType::PushRules)
.await
.map_err(|_| Error::BadRequest(ErrorKind::NotFound, "PushRules event not found."))?;
.map_err(|_| err!(Request(NotFound("PushRules event not found."))))?;
let account_data = serde_json::from_str::<PushRulesEvent>(event.get())
.map_err(|_| Error::bad_database("Invalid account data event in db."))?
.content;
let global = account_data.global;
let actions = global
let actions = event
.content
.global
.get(body.kind.clone(), &body.rule_id)
.map(|rule| rule.actions().to_owned())
.ok_or(Error::BadRequest(ErrorKind::NotFound, "Push rule not found."))?;
.ok_or(err!(Request(NotFound("Push rule not found."))))?;
Ok(get_pushrule_actions::v3::Response {
actions,
@ -217,14 +193,11 @@ pub(crate) async fn set_pushrule_actions_route(
));
}
let event = services
let mut account_data: PushRulesEvent = services
.account_data
.get(None, sender_user, GlobalAccountDataEventType::PushRules.to_string().into())
.get_global(sender_user, GlobalAccountDataEventType::PushRules)
.await
.map_err(|_| Error::BadRequest(ErrorKind::NotFound, "PushRules event not found."))?;
let mut account_data = serde_json::from_str::<PushRulesEvent>(event.get())
.map_err(|_| Error::bad_database("Invalid account data event in db."))?;
.map_err(|_| err!(Request(NotFound("PushRules event not found."))))?;
if account_data
.content
@ -263,20 +236,18 @@ pub(crate) async fn get_pushrule_enabled_route(
));
}
let event = services
let event: PushRulesEvent = services
.account_data
.get(None, sender_user, GlobalAccountDataEventType::PushRules.to_string().into())
.get_global(sender_user, GlobalAccountDataEventType::PushRules)
.await
.map_err(|_| Error::BadRequest(ErrorKind::NotFound, "PushRules event not found."))?;
.map_err(|_| err!(Request(NotFound("PushRules event not found."))))?;
let account_data = serde_json::from_str::<PushRulesEvent>(event.get())
.map_err(|_| Error::bad_database("Invalid account data event in db."))?;
let global = account_data.content.global;
let enabled = global
let enabled = event
.content
.global
.get(body.kind.clone(), &body.rule_id)
.map(ruma::push::AnyPushRuleRef::enabled)
.ok_or(Error::BadRequest(ErrorKind::NotFound, "Push rule not found."))?;
.ok_or(err!(Request(NotFound("Push rule not found."))))?;
Ok(get_pushrule_enabled::v3::Response {
enabled,
@ -298,14 +269,11 @@ pub(crate) async fn set_pushrule_enabled_route(
));
}
let event = services
let mut account_data: PushRulesEvent = services
.account_data
.get(None, sender_user, GlobalAccountDataEventType::PushRules.to_string().into())
.get_global(sender_user, GlobalAccountDataEventType::PushRules)
.await
.map_err(|_| Error::BadRequest(ErrorKind::NotFound, "PushRules event not found."))?;
let mut account_data = serde_json::from_str::<PushRulesEvent>(event.get())
.map_err(|_| Error::bad_database("Invalid account data event in db."))?;
.map_err(|_| err!(Request(NotFound("PushRules event not found."))))?;
if account_data
.content
@ -344,14 +312,11 @@ pub(crate) async fn delete_pushrule_route(
));
}
let event = services
let mut account_data: PushRulesEvent = services
.account_data
.get(None, sender_user, GlobalAccountDataEventType::PushRules.to_string().into())
.get_global(sender_user, GlobalAccountDataEventType::PushRules)
.await
.map_err(|_| Error::BadRequest(ErrorKind::NotFound, "PushRules event not found."))?;
let mut account_data = serde_json::from_str::<PushRulesEvent>(event.get())
.map_err(|_| Error::bad_database("Invalid account data event in db."))?;
.map_err(|_| err!(Request(NotFound("PushRules event not found."))))?;
if let Err(error) = account_data
.content

50
src/api/client/tag.rs
View file

@ -9,7 +9,7 @@ use ruma::{
},
};
use crate::{Error, Result, Ruma};
use crate::{Result, Ruma};
/// # `PUT /_matrix/client/r0/user/{userId}/rooms/{roomId}/tags/{tag}`
///
@ -21,21 +21,15 @@ pub(crate) async fn update_tag_route(
) -> Result<create_tag::v3::Response> {
let sender_user = body.sender_user.as_ref().expect("user is authenticated");
let event = services
let mut tags_event = services
.account_data
.get(Some(&body.room_id), sender_user, RoomAccountDataEventType::Tag)
.await;
let mut tags_event = event.map_or_else(
|_| {
Ok(TagEvent {
.get_room(&body.room_id, sender_user, RoomAccountDataEventType::Tag)
.await
.unwrap_or(TagEvent {
content: TagEventContent {
tags: BTreeMap::new(),
},
})
},
|e| serde_json::from_str(e.get()).map_err(|_| Error::bad_database("Invalid account data event in db.")),
)?;
});
tags_event
.content
@ -65,21 +59,15 @@ pub(crate) async fn delete_tag_route(
) -> Result<delete_tag::v3::Response> {
let sender_user = body.sender_user.as_ref().expect("user is authenticated");
let event = services
let mut tags_event = services
.account_data
.get(Some(&body.room_id), sender_user, RoomAccountDataEventType::Tag)
.await;
let mut tags_event = event.map_or_else(
|_| {
Ok(TagEvent {
.get_room(&body.room_id, sender_user, RoomAccountDataEventType::Tag)
.await
.unwrap_or(TagEvent {
content: TagEventContent {
tags: BTreeMap::new(),
},
})
},
|e| serde_json::from_str(e.get()).map_err(|_| Error::bad_database("Invalid account data event in db.")),
)?;
});
tags_event.content.tags.remove(&body.tag.clone().into());
@ -106,21 +94,15 @@ pub(crate) async fn get_tags_route(
) -> Result<get_tags::v3::Response> {
let sender_user = body.sender_user.as_ref().expect("user is authenticated");
let event = services
let tags_event = services
.account_data
.get(Some(&body.room_id), sender_user, RoomAccountDataEventType::Tag)
.await;
let tags_event = event.map_or_else(
|_| {
Ok(TagEvent {
.get_room(&body.room_id, sender_user, RoomAccountDataEventType::Tag)
.await
.unwrap_or(TagEvent {
content: TagEventContent {
tags: BTreeMap::new(),
},
})
},
|e| serde_json::from_str(e.get()).map_err(|_| Error::bad_database("Invalid account data event in db.")),
)?;
});
Ok(get_tags::v3::Response {
tags: tags_event.content.tags,

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

@ -5,14 +5,17 @@ use conduit::{
utils::{stream::TryIgnore, ReadyExt},
Err, Error, Result,
};
use database::{Deserialized, Map};
use database::{Deserialized, Handle, Map};
use futures::{StreamExt, TryFutureExt};
use ruma::{
events::{AnyGlobalAccountDataEvent, AnyRawAccountDataEvent, AnyRoomAccountDataEvent, RoomAccountDataEventType},
events::{
AnyGlobalAccountDataEvent, AnyRawAccountDataEvent, AnyRoomAccountDataEvent, GlobalAccountDataEventType,
RoomAccountDataEventType,
},
serde::Raw,
RoomId, UserId,
};
use serde_json::value::RawValue;
use serde::Deserialize;
use crate::{globals, Dep};
@ -97,18 +100,36 @@ pub async fn update(
Ok(())
}
/// Searches the account data for a specific kind.
/// Searches the room account data for a specific kind.
#[implement(Service)]
pub async fn get(
&self, room_id: Option<&RoomId>, user_id: &UserId, kind: RoomAccountDataEventType,
) -> Result<Box<RawValue>> {
let key = (room_id, user_id, kind.to_string());
pub async fn get_global<T>(&self, user_id: &UserId, kind: GlobalAccountDataEventType) -> Result<T>
where
T: for<'de> Deserialize<'de>,
{
self.get_raw(None, user_id, &kind.to_string())
.await
.deserialized()
}
/// Searches the global account data for a specific kind.
#[implement(Service)]
pub async fn get_room<T>(&self, room_id: &RoomId, user_id: &UserId, kind: RoomAccountDataEventType) -> Result<T>
where
T: for<'de> Deserialize<'de>,
{
self.get_raw(Some(room_id), user_id, &kind.to_string())
.await
.deserialized()
}
#[implement(Service)]
pub async fn get_raw(&self, room_id: Option<&RoomId>, user_id: &UserId, kind: &str) -> Result<Handle<'_>> {
let key = (room_id, user_id, kind.to_owned());
self.db
.roomusertype_roomuserdataid
.qry(&key)
.and_then(|roomuserdataid| self.db.roomuserdataid_accountdata.get(&roomuserdataid))
.await
.deserialized()
}
/// Returns all changes to the account data that happened after `since`.

3
src/service/admin/grant.rs
View file

@ -143,9 +143,8 @@ async fn set_room_tag(&self, room_id: &RoomId, user_id: &UserId, tag: &str) -> R
let mut event = self
.services
.account_data
.get(Some(room_id), user_id, RoomAccountDataEventType::Tag)
.get_room(room_id, user_id, RoomAccountDataEventType::Tag)
.await
.and_then(|event| serde_json::from_str(event.get()).map_err(Into::into))
.unwrap_or_else(|_| TagEvent {
content: TagEventContent {
tags: BTreeMap::new(),

11
src/service/globals/migrations.rs
View file

@ -215,13 +215,12 @@ async fn db_lt_12(services: &Services) -> Result<()> {
},
};
let raw_rules_list = services
let mut account_data: PushRulesEvent = services
.account_data
.get(None, &user, GlobalAccountDataEventType::PushRules.to_string().into())
.get_global(&user, GlobalAccountDataEventType::PushRules)
.await
.expect("Username is invalid");
let mut account_data = serde_json::from_str::<PushRulesEvent>(raw_rules_list.get()).unwrap();
let rules_list = &mut account_data.content.global;
//content rule
@ -294,14 +293,12 @@ async fn db_lt_13(services: &Services) -> Result<()> {
},
};
let raw_rules_list = services
let mut account_data: PushRulesEvent = services
.account_data
.get(None, &user, GlobalAccountDataEventType::PushRules.to_string().into())
.get_global(&user, GlobalAccountDataEventType::PushRules)
.await
.expect("Username is invalid");
let mut account_data = serde_json::from_str::<PushRulesEvent>(raw_rules_list.get()).unwrap();
let user_default_rules = Ruleset::server_default(&user);
account_data
.content

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

@ -146,12 +146,9 @@ impl Service {
if let Ok(tag_event) = self
.services
.account_data
.get(Some(&predecessor.room_id), user_id, RoomAccountDataEventType::Tag)
.get_room(&predecessor.room_id, user_id, RoomAccountDataEventType::Tag)
.await
.and_then(|event| {
serde_json::from_str(event.get())
.map_err(|e| err!(Database(warn!("Invalid account data event in db: {e:?}"))))
}) {
{
self.services
.account_data
.update(Some(room_id), user_id, RoomAccountDataEventType::Tag, &tag_event)
@ -163,12 +160,9 @@ impl Service {
if let Ok(mut direct_event) = self
.services
.account_data
.get(None, user_id, GlobalAccountDataEventType::Direct.to_string().into())
.get_global::<DirectEvent>(user_id, GlobalAccountDataEventType::Direct)
.await
.and_then(|event| {
serde_json::from_str::<DirectEvent>(event.get())
.map_err(|e| err!(Database(warn!("Invalid account data event in db: {e:?}"))))
}) {
{
let mut room_ids_updated = false;
for room_ids in direct_event.content.0.values_mut() {
if room_ids.iter().any(|r| r == &predecessor.room_id) {

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

@ -407,9 +407,8 @@ impl Service {
let rules_for_user = self
.services
.account_data
.get(None, user, GlobalAccountDataEventType::PushRules.to_string().into())
.get_global(user, GlobalAccountDataEventType::PushRules)
.await
.and_then(|event| serde_json::from_str::<PushRulesEvent>(event.get()).map_err(Into::into))
.map_or_else(|_| Ruleset::server_default(user), |ev: PushRulesEvent| ev.content.global);
let mut highlight = false;

3
src/service/sending/sender.rs
View file

@ -539,9 +539,8 @@ impl Service {
let rules_for_user = self
.services
.account_data
.get(None, userid, GlobalAccountDataEventType::PushRules.to_string().into())
.get_global(userid, GlobalAccountDataEventType::PushRules)
.await
.and_then(|event| serde_json::from_str::<PushRulesEvent>(event.get()).map_err(Into::into))
.map_or_else(
|_| push::Ruleset::server_default(userid),
|ev: PushRulesEvent| ev.content.global,

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

@ -98,19 +98,9 @@ impl Service {
pub async fn user_is_ignored(&self, sender_user: &UserId, recipient_user: &UserId) -> bool {
self.services
.account_data
.get(
None,
recipient_user,
GlobalAccountDataEventType::IgnoredUserList
.to_string()
.into(),
)
.get_global(recipient_user, GlobalAccountDataEventType::IgnoredUserList)
.await
.and_then(|event| {
serde_json::from_str::<IgnoredUserListEvent>(event.get())
.map_err(|e| err!(Database(warn!("Invalid account data event in db: {e:?}"))))
})
.map_or(false, |ignored| {
.map_or(false, |ignored: IgnoredUserListEvent| {
ignored
.content
.ignored_users