mirror of https://gitlab.com/famedly/conduit
Merge branch 'develop' into 'master'
State resolution overhaul See merge request famedly/conduit!53merge-requests/56/head
Timo Kösters
3 years ago
commit
d6b59cd20c
File diff suppressed because it is too large
Load Diff
@ -1 +1 @@
|
||||
1.47.0
|
||||
1.50.0
|
||||
|
||||
@ -1 +1,2 @@
|
||||
merge_imports = true
|
||||
unstable_features = true
|
||||
imports_granularity="Crate"
|
||||
|
||||
@ -0,0 +1,327 @@
|
||||
use crate::{Database, Error, PduEvent, Result};
|
||||
use log::{error, info, warn};
|
||||
use ruma::{
|
||||
api::{
|
||||
client::r0::push::{Pusher, PusherKind},
|
||||
push_gateway::send_event_notification::{
|
||||
self,
|
||||
v1::{Device, Notification, NotificationCounts, NotificationPriority},
|
||||
},
|
||||
IncomingResponse, OutgoingRequest,
|
||||
},
|
||||
events::{room::power_levels::PowerLevelsEventContent, EventType},
|
||||
push::{Action, PushConditionRoomCtx, PushFormat, Ruleset, Tweak},
|
||||
uint, UInt, UserId,
|
||||
};
|
||||
use sled::IVec;
|
||||
|
||||
use std::{convert::TryFrom, fmt::Debug};
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct PushData {
|
||||
/// UserId + pushkey -> Pusher
|
||||
pub(super) senderkey_pusher: sled::Tree,
|
||||
}
|
||||
|
||||
impl PushData {
|
||||
pub fn new(db: &sled::Db) -> Result<Self> {
|
||||
Ok(Self {
|
||||
senderkey_pusher: db.open_tree("senderkey_pusher")?,
|
||||
})
|
||||
}
|
||||
|
||||
pub fn set_pusher(&self, sender: &UserId, pusher: Pusher) -> Result<()> {
|
||||
let mut key = sender.as_bytes().to_vec();
|
||||
key.push(0xff);
|
||||
key.extend_from_slice(pusher.pushkey.as_bytes());
|
||||
|
||||
// There are 2 kinds of pushers but the spec says: null deletes the pusher.
|
||||
if pusher.kind.is_none() {
|
||||
return self
|
||||
.senderkey_pusher
|
||||
.remove(key)
|
||||
.map(|_| ())
|
||||
.map_err(Into::into);
|
||||
}
|
||||
|
||||
self.senderkey_pusher.insert(
|
||||
key,
|
||||
&*serde_json::to_string(&pusher).expect("Pusher is valid JSON string"),
|
||||
)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn get_pusher(&self, senderkey: &[u8]) -> Result<Option<Pusher>> {
|
||||
self.senderkey_pusher
|
||||
.get(senderkey)?
|
||||
.map(|push| {
|
||||
Ok(serde_json::from_slice(&*push)
|
||||
.map_err(|_| Error::bad_database("Invalid Pusher in db."))?)
|
||||
})
|
||||
.transpose()
|
||||
}
|
||||
|
||||
pub fn get_pushers(&self, sender: &UserId) -> Result<Vec<Pusher>> {
|
||||
let mut prefix = sender.as_bytes().to_vec();
|
||||
prefix.push(0xff);
|
||||
|
||||
self.senderkey_pusher
|
||||
.scan_prefix(prefix)
|
||||
.values()
|
||||
.map(|push| {
|
||||
let push = push.map_err(|_| Error::bad_database("Invalid push bytes in db."))?;
|
||||
Ok(serde_json::from_slice(&*push)
|
||||
.map_err(|_| Error::bad_database("Invalid Pusher in db."))?)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
pub fn get_pusher_senderkeys(&self, sender: &UserId) -> impl Iterator<Item = Result<IVec>> {
|
||||
let mut prefix = sender.as_bytes().to_vec();
|
||||
prefix.push(0xff);
|
||||
|
||||
self.senderkey_pusher
|
||||
.scan_prefix(prefix)
|
||||
.keys()
|
||||
.map(|r| Ok(r?))
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn send_request<T: OutgoingRequest>(
|
||||
globals: &crate::database::globals::Globals,
|
||||
destination: &str,
|
||||
request: T,
|
||||
) -> Result<T::IncomingResponse>
|
||||
where
|
||||
T: Debug,
|
||||
{
|
||||
let destination = destination.replace("/_matrix/push/v1/notify", "");
|
||||
|
||||
let http_request = request
|
||||
.try_into_http_request(&destination, Some(""))
|
||||
.map_err(|e| {
|
||||
warn!("Failed to find destination {}: {}", destination, e);
|
||||
Error::BadServerResponse("Invalid destination")
|
||||
})?;
|
||||
|
||||
let reqwest_request = reqwest::Request::try_from(http_request)
|
||||
.expect("all http requests are valid reqwest requests");
|
||||
|
||||
// TODO: we could keep this very short and let expo backoff do it's thing...
|
||||
//*reqwest_request.timeout_mut() = Some(Duration::from_secs(5));
|
||||
|
||||
let url = reqwest_request.url().clone();
|
||||
let reqwest_response = globals.reqwest_client().execute(reqwest_request).await;
|
||||
|
||||
// Because reqwest::Response -> http::Response is complicated:
|
||||
match reqwest_response {
|
||||
Ok(mut reqwest_response) => {
|
||||
let status = reqwest_response.status();
|
||||
let mut http_response = http::Response::builder().status(status);
|
||||
let headers = http_response.headers_mut().unwrap();
|
||||
|
||||
for (k, v) in reqwest_response.headers_mut().drain() {
|
||||
if let Some(key) = k {
|
||||
headers.insert(key, v);
|
||||
}
|
||||
}
|
||||
|
||||
let status = reqwest_response.status();
|
||||
|
||||
let body = reqwest_response.bytes().await.unwrap_or_else(|e| {
|
||||
warn!("server error {}", e);
|
||||
Vec::new().into()
|
||||
}); // TODO: handle timeout
|
||||
|
||||
if status != 200 {
|
||||
info!(
|
||||
"Push gateway returned bad response {} {}\n{}\n{:?}",
|
||||
destination,
|
||||
status,
|
||||
url,
|
||||
crate::utils::string_from_bytes(&body)
|
||||
);
|
||||
}
|
||||
|
||||
let response = T::IncomingResponse::try_from_http_response(
|
||||
http_response
|
||||
.body(body)
|
||||
.expect("reqwest body is valid http body"),
|
||||
);
|
||||
response.map_err(|_| {
|
||||
info!(
|
||||
"Push gateway returned invalid response bytes {}\n{}",
|
||||
destination, url
|
||||
);
|
||||
Error::BadServerResponse("Push gateway returned bad response.")
|
||||
})
|
||||
}
|
||||
Err(e) => Err(e.into()),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn send_push_notice(
|
||||
user: &UserId,
|
||||
unread: UInt,
|
||||
pusher: &Pusher,
|
||||
ruleset: Ruleset,
|
||||
pdu: &PduEvent,
|
||||
db: &Database,
|
||||
) -> Result<()> {
|
||||
let mut notify = None;
|
||||
let mut tweaks = Vec::new();
|
||||
|
||||
for action in get_actions(user, &ruleset, pdu, db)? {
|
||||
let n = match action {
|
||||
Action::DontNotify => false,
|
||||
// TODO: Implement proper support for coalesce
|
||||
Action::Notify | Action::Coalesce => true,
|
||||
Action::SetTweak(tweak) => {
|
||||
tweaks.push(tweak.clone());
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
if notify.is_some() {
|
||||
return Err(Error::bad_database(
|
||||
r#"Malformed pushrule contains more than one of these actions: ["dont_notify", "notify", "coalesce"]"#,
|
||||
));
|
||||
}
|
||||
|
||||
notify = Some(n);
|
||||
}
|
||||
|
||||
if notify == Some(true) {
|
||||
send_notice(unread, pusher, tweaks, pdu, db).await?;
|
||||
}
|
||||
// Else the event triggered no actions
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn get_actions<'a>(
|
||||
user: &UserId,
|
||||
ruleset: &'a Ruleset,
|
||||
pdu: &PduEvent,
|
||||
db: &Database,
|
||||
) -> Result<impl 'a + Iterator<Item = Action>> {
|
||||
let power_levels: PowerLevelsEventContent = db
|
||||
.rooms
|
||||
.room_state_get(&pdu.room_id, &EventType::RoomPowerLevels, "")?
|
||||
.map(|ev| {
|
||||
serde_json::from_value(ev.content)
|
||||
.map_err(|_| Error::bad_database("invalid m.room.power_levels event"))
|
||||
})
|
||||
.transpose()?
|
||||
.unwrap_or_default();
|
||||
|
||||
let ctx = PushConditionRoomCtx {
|
||||
room_id: pdu.room_id.clone(),
|
||||
member_count: 10_u32.into(), // TODO: get member count efficiently
|
||||
user_display_name: db
|
||||
.users
|
||||
.displayname(&user)?
|
||||
.unwrap_or_else(|| user.localpart().to_owned()),
|
||||
users_power_levels: power_levels.users,
|
||||
default_power_level: power_levels.users_default,
|
||||
notification_power_levels: power_levels.notifications,
|
||||
};
|
||||
|
||||
Ok(ruleset
|
||||
.get_actions(&pdu.to_sync_room_event(), &ctx)
|
||||
.map(Clone::clone))
|
||||
}
|
||||
|
||||
async fn send_notice(
|
||||
unread: UInt,
|
||||
pusher: &Pusher,
|
||||
tweaks: Vec<Tweak>,
|
||||
event: &PduEvent,
|
||||
db: &Database,
|
||||
) -> Result<()> {
|
||||
// TODO: email
|
||||
if pusher.kind == Some(PusherKind::Email) {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
// TODO:
|
||||
// Two problems with this
|
||||
// 1. if "event_id_only" is the only format kind it seems we should never add more info
|
||||
// 2. can pusher/devices have conflicting formats
|
||||
let event_id_only = pusher.data.format == Some(PushFormat::EventIdOnly);
|
||||
let url = if let Some(url) = pusher.data.url.as_ref() {
|
||||
url
|
||||
} else {
|
||||
error!("Http Pusher must have URL specified.");
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
let mut device = Device::new(pusher.app_id.clone(), pusher.pushkey.clone());
|
||||
let mut data_minus_url = pusher.data.clone();
|
||||
// The url must be stripped off according to spec
|
||||
data_minus_url.url = None;
|
||||
device.data = Some(data_minus_url);
|
||||
|
||||
// Tweaks are only added if the format is NOT event_id_only
|
||||
if !event_id_only {
|
||||
device.tweaks = tweaks.clone();
|
||||
}
|
||||
|
||||
let d = &[device];
|
||||
let mut notifi = Notification::new(d);
|
||||
|
||||
notifi.prio = NotificationPriority::Low;
|
||||
notifi.event_id = Some(&event.event_id);
|
||||
notifi.room_id = Some(&event.room_id);
|
||||
// TODO: missed calls
|
||||
notifi.counts = NotificationCounts::new(unread, uint!(0));
|
||||
|
||||
if event.kind == EventType::RoomEncrypted
|
||||
|| tweaks
|
||||
.iter()
|
||||
.any(|t| matches!(t, Tweak::Highlight(true) | Tweak::Sound(_)))
|
||||
{
|
||||
notifi.prio = NotificationPriority::High
|
||||
}
|
||||
|
||||
if event_id_only {
|
||||
send_request(
|
||||
&db.globals,
|
||||
&url,
|
||||
send_event_notification::v1::Request::new(notifi),
|
||||
)
|
||||
.await?;
|
||||
} else {
|
||||
notifi.sender = Some(&event.sender);
|
||||
notifi.event_type = Some(&event.kind);
|
||||
notifi.content = serde_json::value::to_raw_value(&event.content).ok();
|
||||
|
||||
if event.kind == EventType::RoomMember {
|
||||
notifi.user_is_target = event.state_key.as_deref() == Some(event.sender.as_str());
|
||||
}
|
||||
|
||||
let user_name = db.users.displayname(&event.sender)?;
|
||||
notifi.sender_display_name = user_name.as_deref();
|
||||
let room_name = db
|
||||
.rooms
|
||||
.room_state_get(&event.room_id, &EventType::RoomName, "")?
|
||||
.map(|pdu| match pdu.content.get("name") {
|
||||
Some(serde_json::Value::String(s)) => Some(s.to_string()),
|
||||
_ => None,
|
||||
})
|
||||
.flatten();
|
||||
notifi.room_name = room_name.as_deref();
|
||||
|
||||
send_request(
|
||||
&db.globals,
|
||||
&url,
|
||||
send_event_notification::v1::Request::new(notifi),
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
|
||||
// TODO: email
|
||||
|
||||
Ok(())
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@ -1,256 +0,0 @@
|
||||
use ruma::{
|
||||
push::{
|
||||
Action, ConditionalPushRule, ConditionalPushRuleInit, ContentPushRule, OverridePushRule,
|
||||
PatternedPushRule, PatternedPushRuleInit, PushCondition, RoomMemberCountIs, Ruleset, Tweak,
|
||||
UnderridePushRule,
|
||||
},
|
||||
UserId,
|
||||
};
|
||||
|
||||
pub fn default_pushrules(user_id: &UserId) -> Ruleset {
|
||||
let mut rules = Ruleset::default();
|
||||
|
||||
rules.add(ContentPushRule(contains_user_name_rule(&user_id)));
|
||||
|
||||
for rule in vec![
|
||||
master_rule(),
|
||||
suppress_notices_rule(),
|
||||
invite_for_me_rule(),
|
||||
member_event_rule(),
|
||||
contains_display_name_rule(),
|
||||
tombstone_rule(),
|
||||
roomnotif_rule(),
|
||||
] {
|
||||
rules.add(OverridePushRule(rule));
|
||||
}
|
||||
|
||||
for rule in vec![
|
||||
call_rule(),
|
||||
encrypted_room_one_to_one_rule(),
|
||||
room_one_to_one_rule(),
|
||||
message_rule(),
|
||||
encrypted_rule(),
|
||||
] {
|
||||
rules.add(UnderridePushRule(rule));
|
||||
}
|
||||
|
||||
rules
|
||||
}
|
||||
|
||||
pub fn master_rule() -> ConditionalPushRule {
|
||||
ConditionalPushRuleInit {
|
||||
actions: vec![Action::DontNotify],
|
||||
default: true,
|
||||
enabled: false,
|
||||
rule_id: ".m.rule.master".to_owned(),
|
||||
conditions: vec![],
|
||||
}
|
||||
.into()
|
||||
}
|
||||
|
||||
pub fn suppress_notices_rule() -> ConditionalPushRule {
|
||||
ConditionalPushRuleInit {
|
||||
actions: vec![Action::DontNotify],
|
||||
default: true,
|
||||
enabled: true,
|
||||
rule_id: ".m.rule.suppress_notices".to_owned(),
|
||||
conditions: vec![PushCondition::EventMatch {
|
||||
key: "content.msgtype".to_owned(),
|
||||
pattern: "m.notice".to_owned(),
|
||||
}],
|
||||
}
|
||||
.into()
|
||||
}
|
||||
|
||||
pub fn invite_for_me_rule() -> ConditionalPushRule {
|
||||
ConditionalPushRuleInit {
|
||||
actions: vec![
|
||||
Action::Notify,
|
||||
Action::SetTweak(Tweak::Sound("default".to_owned())),
|
||||
Action::SetTweak(Tweak::Highlight(false)),
|
||||
],
|
||||
default: true,
|
||||
enabled: true,
|
||||
rule_id: ".m.rule.invite_for_me".to_owned(),
|
||||
conditions: vec![PushCondition::EventMatch {
|
||||
key: "content.membership".to_owned(),
|
||||
pattern: "m.invite".to_owned(),
|
||||
}],
|
||||
}
|
||||
.into()
|
||||
}
|
||||
|
||||
pub fn member_event_rule() -> ConditionalPushRule {
|
||||
ConditionalPushRuleInit {
|
||||
actions: vec![Action::DontNotify],
|
||||
default: true,
|
||||
enabled: true,
|
||||
rule_id: ".m.rule.member_event".to_owned(),
|
||||
conditions: vec![PushCondition::EventMatch {
|
||||
key: "content.membership".to_owned(),
|
||||
pattern: "type".to_owned(),
|
||||
}],
|
||||
}
|
||||
.into()
|
||||
}
|
||||
|
||||
pub fn contains_display_name_rule() -> ConditionalPushRule {
|
||||
ConditionalPushRuleInit {
|
||||
actions: vec![
|
||||
Action::Notify,
|
||||
Action::SetTweak(Tweak::Sound("default".to_owned())),
|
||||
Action::SetTweak(Tweak::Highlight(true)),
|
||||
],
|
||||
default: true,
|
||||
enabled: true,
|
||||
rule_id: ".m.rule.contains_display_name".to_owned(),
|
||||
conditions: vec![PushCondition::ContainsDisplayName],
|
||||
}
|
||||
.into()
|
||||
}
|
||||
|
||||
pub fn tombstone_rule() -> ConditionalPushRule {
|
||||
ConditionalPushRuleInit {
|
||||
actions: vec![Action::Notify, Action::SetTweak(Tweak::Highlight(true))],
|
||||
default: true,
|
||||
enabled: true,
|
||||
rule_id: ".m.rule.tombstone".to_owned(),
|
||||
conditions: vec![
|
||||
PushCondition::EventMatch {
|
||||
key: "type".to_owned(),
|
||||
pattern: "m.room.tombstone".to_owned(),
|
||||
},
|
||||
PushCondition::EventMatch {
|
||||
key: "state_key".to_owned(),
|
||||
pattern: "".to_owned(),
|
||||
},
|
||||
],
|
||||
}
|
||||
.into()
|
||||
}
|
||||
|
||||
pub fn roomnotif_rule() -> ConditionalPushRule {
|
||||
ConditionalPushRuleInit {
|
||||
actions: vec![Action::Notify, Action::SetTweak(Tweak::Highlight(true))],
|
||||
default: true,
|
||||
enabled: true,
|
||||
rule_id: ".m.rule.roomnotif".to_owned(),
|
||||
conditions: vec![
|
||||
PushCondition::EventMatch {
|
||||
key: "content.body".to_owned(),
|
||||
pattern: "@room".to_owned(),
|
||||
},
|
||||
PushCondition::SenderNotificationPermission {
|
||||
key: "room".to_owned(),
|
||||
},
|
||||
],
|
||||
}
|
||||
.into()
|
||||
}
|
||||
|
||||
pub fn contains_user_name_rule(user_id: &UserId) -> PatternedPushRule {
|
||||
PatternedPushRuleInit {
|
||||
actions: vec![
|
||||
Action::Notify,
|
||||
Action::SetTweak(Tweak::Sound("default".to_owned())),
|
||||
Action::SetTweak(Tweak::Highlight(true)),
|
||||
],
|
||||
default: true,
|
||||
enabled: true,
|
||||
rule_id: ".m.rule.contains_user_name".to_owned(),
|
||||
pattern: user_id.localpart().to_owned(),
|
||||
}
|
||||
.into()
|
||||
}
|
||||
|
||||
pub fn call_rule() -> ConditionalPushRule {
|
||||
ConditionalPushRuleInit {
|
||||
actions: vec![
|
||||
Action::Notify,
|
||||
Action::SetTweak(Tweak::Sound("ring".to_owned())),
|
||||
Action::SetTweak(Tweak::Highlight(false)),
|
||||
],
|
||||
default: true,
|
||||
enabled: true,
|
||||
rule_id: ".m.rule.call".to_owned(),
|
||||
conditions: vec![PushCondition::EventMatch {
|
||||
key: "type".to_owned(),
|
||||
pattern: "m.call.invite".to_owned(),
|
||||
}],
|
||||
}
|
||||
.into()
|
||||
}
|
||||
|
||||
pub fn encrypted_room_one_to_one_rule() -> ConditionalPushRule {
|
||||
ConditionalPushRuleInit {
|
||||
actions: vec![
|
||||
Action::Notify,
|
||||
Action::SetTweak(Tweak::Sound("default".to_owned())),
|
||||
Action::SetTweak(Tweak::Highlight(false)),
|
||||
],
|
||||
default: true,
|
||||
enabled: true,
|
||||
rule_id: ".m.rule.encrypted_room_one_to_one".to_owned(),
|
||||
conditions: vec![
|
||||
PushCondition::RoomMemberCount {
|
||||
is: RoomMemberCountIs::from(2_u32.into()..),
|
||||
},
|
||||
PushCondition::EventMatch {
|
||||
key: "type".to_owned(),
|
||||
pattern: "m.room.encrypted".to_owned(),
|
||||
},
|
||||
],
|
||||
}
|
||||
.into()
|
||||
}
|
||||
|
||||
pub fn room_one_to_one_rule() -> ConditionalPushRule {
|
||||
ConditionalPushRuleInit {
|
||||
actions: vec![
|
||||
Action::Notify,
|
||||
Action::SetTweak(Tweak::Sound("default".to_owned())),
|
||||
Action::SetTweak(Tweak::Highlight(false)),
|
||||
],
|
||||
default: true,
|
||||
enabled: true,
|
||||
rule_id: ".m.rule.room_one_to_one".to_owned(),
|
||||
conditions: vec![
|
||||
PushCondition::RoomMemberCount {
|
||||
is: RoomMemberCountIs::from(2_u32.into()..),
|
||||
},
|
||||
PushCondition::EventMatch {
|
||||
key: "type".to_owned(),
|
||||
pattern: "m.room.message".to_owned(),
|
||||
},
|
||||
],
|
||||
}
|
||||
.into()
|
||||
}
|
||||
|
||||
pub fn message_rule() -> ConditionalPushRule {
|
||||
ConditionalPushRuleInit {
|
||||
actions: vec![Action::Notify, Action::SetTweak(Tweak::Highlight(false))],
|
||||
default: true,
|
||||
enabled: true,
|
||||
rule_id: ".m.rule.message".to_owned(),
|
||||
conditions: vec![PushCondition::EventMatch {
|
||||
key: "type".to_owned(),
|
||||
pattern: "m.room.message".to_owned(),
|
||||
}],
|
||||
}
|
||||
.into()
|
||||
}
|
||||
|
||||
pub fn encrypted_rule() -> ConditionalPushRule {
|
||||
ConditionalPushRuleInit {
|
||||
actions: vec![Action::Notify, Action::SetTweak(Tweak::Highlight(false))],
|
||||
default: true,
|
||||
enabled: true,
|
||||
rule_id: ".m.rule.encrypted".to_owned(),
|
||||
conditions: vec![PushCondition::EventMatch {
|
||||
key: "type".to_owned(),
|
||||
pattern: "m.room.encrypted".to_owned(),
|
||||
}],
|
||||
}
|
||||
.into()
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
Loading…
Reference in New Issue