Merge branch 'fixtyping' into 'next'
Fix typing indicators and unencrypted messages in encrypted rooms See merge request famedly/conduit!409
This commit is contained in:
commit
2a52f666dc
4 changed files with 72 additions and 61 deletions
|
@ -388,13 +388,35 @@ async fn sync_helper(
|
||||||
))
|
))
|
||||||
};
|
};
|
||||||
|
|
||||||
|
let since_sender_member: Option<RoomMemberEventContent> = since_shortstatehash
|
||||||
|
.and_then(|shortstatehash| {
|
||||||
|
services()
|
||||||
|
.rooms
|
||||||
|
.state_accessor
|
||||||
|
.state_get(
|
||||||
|
shortstatehash,
|
||||||
|
&StateEventType::RoomMember,
|
||||||
|
sender_user.as_str(),
|
||||||
|
)
|
||||||
|
.transpose()
|
||||||
|
})
|
||||||
|
.transpose()?
|
||||||
|
.and_then(|pdu| {
|
||||||
|
serde_json::from_str(pdu.content.get())
|
||||||
|
.map_err(|_| Error::bad_database("Invalid PDU in database."))
|
||||||
|
.ok()
|
||||||
|
});
|
||||||
|
|
||||||
|
let joined_since_last_sync =
|
||||||
|
since_sender_member.map_or(true, |member| member.membership != MembershipState::Join);
|
||||||
|
|
||||||
let (
|
let (
|
||||||
heroes,
|
heroes,
|
||||||
joined_member_count,
|
joined_member_count,
|
||||||
invited_member_count,
|
invited_member_count,
|
||||||
joined_since_last_sync,
|
joined_since_last_sync,
|
||||||
state_events,
|
state_events,
|
||||||
) = if since_shortstatehash.is_none() {
|
) = if since_shortstatehash.is_none() || joined_since_last_sync {
|
||||||
// Probably since = 0, we will do an initial sync
|
// Probably since = 0, we will do an initial sync
|
||||||
|
|
||||||
let (joined_member_count, invited_member_count, heroes) = calculate_counts()?;
|
let (joined_member_count, invited_member_count, heroes) = calculate_counts()?;
|
||||||
|
@ -487,23 +509,6 @@ async fn sync_helper(
|
||||||
// Incremental /sync
|
// Incremental /sync
|
||||||
let since_shortstatehash = since_shortstatehash.unwrap();
|
let since_shortstatehash = since_shortstatehash.unwrap();
|
||||||
|
|
||||||
let since_sender_member: Option<RoomMemberEventContent> = services()
|
|
||||||
.rooms
|
|
||||||
.state_accessor
|
|
||||||
.state_get(
|
|
||||||
since_shortstatehash,
|
|
||||||
&StateEventType::RoomMember,
|
|
||||||
sender_user.as_str(),
|
|
||||||
)?
|
|
||||||
.and_then(|pdu| {
|
|
||||||
serde_json::from_str(pdu.content.get())
|
|
||||||
.map_err(|_| Error::bad_database("Invalid PDU in database."))
|
|
||||||
.ok()
|
|
||||||
});
|
|
||||||
|
|
||||||
let joined_since_last_sync = since_sender_member
|
|
||||||
.map_or(true, |member| member.membership != MembershipState::Join);
|
|
||||||
|
|
||||||
let mut state_events = Vec::new();
|
let mut state_events = Vec::new();
|
||||||
let mut lazy_loaded = HashSet::new();
|
let mut lazy_loaded = HashSet::new();
|
||||||
|
|
||||||
|
|
|
@ -1,4 +1,5 @@
|
||||||
use std::collections::HashSet;
|
use std::collections::HashSet;
|
||||||
|
use std::mem;
|
||||||
|
|
||||||
use ruma::{OwnedUserId, RoomId, UserId};
|
use ruma::{OwnedUserId, RoomId, UserId};
|
||||||
|
|
||||||
|
@ -53,6 +54,47 @@ impl service::rooms::edus::typing::Data for KeyValueDatabase {
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn typings_maintain(&self, room_id: &RoomId) -> Result<()> {
|
||||||
|
let mut prefix = room_id.as_bytes().to_vec();
|
||||||
|
prefix.push(0xff);
|
||||||
|
|
||||||
|
let current_timestamp = utils::millis_since_unix_epoch();
|
||||||
|
|
||||||
|
let mut found_outdated = false;
|
||||||
|
|
||||||
|
// Find all outdated edus before inserting a new one
|
||||||
|
for outdated_edu in self
|
||||||
|
.typingid_userid
|
||||||
|
.scan_prefix(prefix)
|
||||||
|
.map(|(key, _)| {
|
||||||
|
Ok::<_, Error>((
|
||||||
|
key.clone(),
|
||||||
|
utils::u64_from_bytes(
|
||||||
|
&key.splitn(2, |&b| b == 0xff).nth(1).ok_or_else(|| {
|
||||||
|
Error::bad_database("RoomTyping has invalid timestamp or delimiters.")
|
||||||
|
})?[0..mem::size_of::<u64>()],
|
||||||
|
)
|
||||||
|
.map_err(|_| Error::bad_database("RoomTyping has invalid timestamp bytes."))?,
|
||||||
|
))
|
||||||
|
})
|
||||||
|
.filter_map(|r| r.ok())
|
||||||
|
.take_while(|&(_, timestamp)| timestamp < current_timestamp)
|
||||||
|
{
|
||||||
|
// This is an outdated edu (time > timestamp)
|
||||||
|
self.typingid_userid.remove(&outdated_edu.0)?;
|
||||||
|
found_outdated = true;
|
||||||
|
}
|
||||||
|
|
||||||
|
if found_outdated {
|
||||||
|
self.roomid_lasttypingupdate.insert(
|
||||||
|
room_id.as_bytes(),
|
||||||
|
&services().globals.next_count()?.to_be_bytes(),
|
||||||
|
)?;
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
fn last_typing_update(&self, room_id: &RoomId) -> Result<u64> {
|
fn last_typing_update(&self, room_id: &RoomId) -> Result<u64> {
|
||||||
Ok(self
|
Ok(self
|
||||||
.roomid_lasttypingupdate
|
.roomid_lasttypingupdate
|
||||||
|
|
|
@ -10,6 +10,9 @@ pub trait Data: Send + Sync {
|
||||||
/// Removes a user from typing before the timeout is reached.
|
/// Removes a user from typing before the timeout is reached.
|
||||||
fn typing_remove(&self, user_id: &UserId, room_id: &RoomId) -> Result<()>;
|
fn typing_remove(&self, user_id: &UserId, room_id: &RoomId) -> Result<()>;
|
||||||
|
|
||||||
|
/// Makes sure that typing events with old timestamps get removed.
|
||||||
|
fn typings_maintain(&self, room_id: &RoomId) -> Result<()>;
|
||||||
|
|
||||||
/// Returns the count of the last typing update in this room.
|
/// Returns the count of the last typing update in this room.
|
||||||
fn last_typing_update(&self, room_id: &RoomId) -> Result<u64>;
|
fn last_typing_update(&self, room_id: &RoomId) -> Result<u64>;
|
||||||
|
|
||||||
|
|
|
@ -21,54 +21,15 @@ impl Service {
|
||||||
self.db.typing_remove(user_id, room_id)
|
self.db.typing_remove(user_id, room_id)
|
||||||
}
|
}
|
||||||
|
|
||||||
/* TODO: Do this in background thread?
|
|
||||||
/// Makes sure that typing events with old timestamps get removed.
|
/// Makes sure that typing events with old timestamps get removed.
|
||||||
fn typings_maintain(
|
fn typings_maintain(&self, room_id: &RoomId) -> Result<()> {
|
||||||
&self,
|
self.db.typings_maintain(room_id)
|
||||||
room_id: &RoomId,
|
|
||||||
globals: &super::super::globals::Globals,
|
|
||||||
) -> Result<()> {
|
|
||||||
let mut prefix = room_id.as_bytes().to_vec();
|
|
||||||
prefix.push(0xff);
|
|
||||||
|
|
||||||
let current_timestamp = utils::millis_since_unix_epoch();
|
|
||||||
|
|
||||||
let mut found_outdated = false;
|
|
||||||
|
|
||||||
// Find all outdated edus before inserting a new one
|
|
||||||
for outdated_edu in self
|
|
||||||
.typingid_userid
|
|
||||||
.scan_prefix(prefix)
|
|
||||||
.map(|(key, _)| {
|
|
||||||
Ok::<_, Error>((
|
|
||||||
key.clone(),
|
|
||||||
utils::u64_from_bytes(
|
|
||||||
&key.splitn(2, |&b| b == 0xff).nth(1).ok_or_else(|| {
|
|
||||||
Error::bad_database("RoomTyping has invalid timestamp or delimiters.")
|
|
||||||
})?[0..mem::size_of::<u64>()],
|
|
||||||
)
|
|
||||||
.map_err(|_| Error::bad_database("RoomTyping has invalid timestamp bytes."))?,
|
|
||||||
))
|
|
||||||
})
|
|
||||||
.filter_map(|r| r.ok())
|
|
||||||
.take_while(|&(_, timestamp)| timestamp < current_timestamp)
|
|
||||||
{
|
|
||||||
// This is an outdated edu (time > timestamp)
|
|
||||||
self.typingid_userid.remove(&outdated_edu.0)?;
|
|
||||||
found_outdated = true;
|
|
||||||
}
|
|
||||||
|
|
||||||
if found_outdated {
|
|
||||||
self.roomid_lasttypingupdate
|
|
||||||
.insert(room_id.as_bytes(), &globals.next_count()?.to_be_bytes())?;
|
|
||||||
}
|
|
||||||
|
|
||||||
Ok(())
|
|
||||||
}
|
}
|
||||||
*/
|
|
||||||
|
|
||||||
/// Returns the count of the last typing update in this room.
|
/// Returns the count of the last typing update in this room.
|
||||||
pub fn last_typing_update(&self, room_id: &RoomId) -> Result<u64> {
|
pub fn last_typing_update(&self, room_id: &RoomId) -> Result<u64> {
|
||||||
|
self.typings_maintain(room_id)?;
|
||||||
|
|
||||||
self.db.last_typing_update(room_id)
|
self.db.last_typing_update(room_id)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue