feat: Implement backfill joined/invited checks for private history

Co-authored-by: Nyaaori <+@nyaaori.cat>

src/api/server_server.rs

@@ -1048,12 +1048,12 @@ fn get_missing_events(
     sender_servername: &ServerName,
     room_id: &RoomId,
     earliest_events: &[OwnedEventId],
-    latest_events: &Vec<OwnedEventId>,
+    latest_events: &[OwnedEventId],
     limit: UInt,
 ) -> Result<Vec<Box<RawJsonValue>>> {
     let limit = u64::from(limit) as usize;
 
-    let mut queued_events = latest_events.clone();
+    let mut queued_events = latest_events.to_owned();
     let mut events = Vec::new();
 
     let mut stop_at_events = HashSet::with_capacity(limit);
@@ -1086,10 +1086,11 @@ fn get_missing_events(
                 ));
             }
 
-            let event_is_visible = services()
-                .rooms
-                .state_accessor
-                .server_can_see_event(sender_servername, &queued_events[i])?;
+            let event_is_visible = services().rooms.state_accessor.server_can_see_event(
+                sender_servername,
+                room_id,
+                &queued_events[i],
+            )?;
 
             if !event_is_visible {
                 i += 1;

src/database/key_value/rooms/state_accessor.rs

@@ -3,11 +3,8 @@ use std::{collections::HashMap, sync::Arc};
 use crate::{database::KeyValueDatabase, service, services, utils, Error, PduEvent, Result};
 use async_trait::async_trait;
 use ruma::{
-    events::{
-        room::history_visibility::{HistoryVisibility, RoomHistoryVisibilityEventContent},
-        StateEventType,
-    },
-    EventId, RoomId, ServerName,
+    events::{room::member::MembershipState, StateEventType},
+    EventId, RoomId, UserId,
 };
 
 #[async_trait]
@@ -126,6 +123,21 @@ impl service::rooms::state_accessor::Data for KeyValueDatabase {
             })
     }
 
+    fn state_get_content(
+        &self,
+        shortstatehash: u64,
+        event_type: &StateEventType,
+        state_key: &str,
+    ) -> Result<Option<serde_json::Value>> {
+        let content = self
+            .state_get(shortstatehash, event_type, state_key)?
+            .map(|event| serde_json::from_str(event.content.get()))
+            .transpose()
+            .map_err(|_| Error::bad_database("Invalid event in database"))?;
+
+        Ok(content)
+    }
+
     /// Returns the state hash for this pdu.
     fn pdu_shortstatehash(&self, event_id: &EventId) -> Result<Option<u64>> {
         self.eventid_shorteventid
@@ -144,33 +156,40 @@ impl service::rooms::state_accessor::Data for KeyValueDatabase {
             })
     }
 
-    /// Whether a server is allowed to see an event through federation, based on
-    /// the room's history_visibility at that event's state.
-    ///
-    /// Note: Joined/Invited history visibility not yet implemented.
-    #[tracing::instrument(skip(self))]
-    fn server_can_see_event(&self, _server_name: &ServerName, event_id: &EventId) -> Result<bool> {
-        let shortstatehash = match self.pdu_shortstatehash(event_id) {
-            Ok(Some(shortstatehash)) => shortstatehash,
-            _ => return Ok(false),
-        };
+    /// The user was a joined member at this state (potentially in the past)
+    fn user_was_joined(&self, shortstatehash: u64, user_id: &UserId) -> Result<bool> {
+        Ok(self
+            .state_get_content(
+                shortstatehash,
+                &StateEventType::RoomMember,
+                user_id.as_str(),
+            )?
+            .map(|content| match content.get("membership") {
+                Some(membership) => MembershipState::from(membership.as_str().unwrap_or("")),
+                None => MembershipState::Leave,
+            } == MembershipState::Join)
+            .unwrap_or(false))
+    }
 
-        let history_visibility = self
-            .state_get(shortstatehash, &StateEventType::RoomHistoryVisibility, "")?
-            .map(|event| serde_json::from_str(event.content.get()))
-            .transpose()
-            .map_err(|_| Error::bad_database("Invalid room history visibility event in database."))?
-            .map(|content: RoomHistoryVisibilityEventContent| content.history_visibility);
-
-        Ok(match history_visibility {
-            Some(HistoryVisibility::WorldReadable) => true,
-            Some(HistoryVisibility::Shared) => true,
-            // TODO: Check if any of the server's users were invited
-            // at this point in time.
-            Some(HistoryVisibility::Joined) => false,
-            Some(HistoryVisibility::Invited) => false,
-            _ => false,
-        })
+    /// The user was an invited or joined room member at this state (potentially
+    /// in the past)
+    fn user_was_invited(&self, shortstatehash: u64, user_id: &UserId) -> Result<bool> {
+        Ok(self
+            .state_get_content(
+                shortstatehash,
+                &StateEventType::RoomMember,
+                user_id.as_str(),
+            )?
+            .map(|content| {
+                let membership = match content.get("membership") {
+                    Some(membership) => MembershipState::from(membership.as_str().unwrap_or("")),
+                    None => MembershipState::Leave,
+                };
+                let joined = membership == MembershipState::Join;
+                let invited = membership == MembershipState::Invite;
+                invited || joined
+            })
+            .unwrap_or(false))
     }
 
     /// Returns the full room state.

src/service/mod.rs

@@ -77,7 +77,12 @@ impl Services {
                 search: rooms::search::Service { db },
                 short: rooms::short::Service { db },
                 state: rooms::state::Service { db },
-                state_accessor: rooms::state_accessor::Service { db },
+                state_accessor: rooms::state_accessor::Service {
+                    db,
+                    server_visibility_cache: Mutex::new(LruCache::new(
+                        (100.0 * config.conduit_cache_capacity_modifier) as usize,
+                    )),
+                },
                 state_cache: rooms::state_cache::Service { db },
                 state_compressor: rooms::state_compressor::Service {
                     db,

src/service/rooms/state_accessor/data.rs

@@ -1,7 +1,7 @@
 use std::{collections::HashMap, sync::Arc};
 
 use async_trait::async_trait;
-use ruma::{events::StateEventType, EventId, RoomId, ServerName};
+use ruma::{events::StateEventType, EventId, RoomId, UserId};
 
 use crate::{PduEvent, Result};
 
@@ -32,11 +32,22 @@ pub trait Data: Send + Sync {
         state_key: &str,
     ) -> Result<Option<Arc<PduEvent>>>;
 
+    fn state_get_content(
+        &self,
+        shortstatehash: u64,
+        event_type: &StateEventType,
+        state_key: &str,
+    ) -> Result<Option<serde_json::Value>>;
+
     /// Returns the state hash for this pdu.
     fn pdu_shortstatehash(&self, event_id: &EventId) -> Result<Option<u64>>;
 
-    /// Returns true if a server has permission to see an event
-    fn server_can_see_event(&self, sever_name: &ServerName, event_id: &EventId) -> Result<bool>;
+    /// The user was a joined member at this state (potentially in the past)
+    fn user_was_joined(&self, shortstatehash: u64, user_id: &UserId) -> Result<bool>;
+
+    /// The user was an invited or joined room member at this state (potentially
+    /// in the past)
+    fn user_was_invited(&self, shortstatehash: u64, user_id: &UserId) -> Result<bool>;
 
     /// Returns the full room state.
     async fn room_state_full(

src/service/rooms/state_accessor/mod.rs

@@ -1,13 +1,21 @@
 mod data;
-use std::{collections::HashMap, sync::Arc};
+use std::{
+    collections::HashMap,
+    sync::{Arc, Mutex},
+};
 
 pub use data::Data;
-use ruma::{events::StateEventType, EventId, RoomId, ServerName};
+use lru_cache::LruCache;
+use ruma::{
+    events::{room::history_visibility::HistoryVisibility, StateEventType},
+    EventId, OwnedServerName, OwnedUserId, RoomId, ServerName, UserId,
+};
 
-use crate::{PduEvent, Result};
+use crate::{services, PduEvent, Result};
 
 pub struct Service {
     pub db: &'static dyn Data,
+    pub server_visibility_cache: Mutex<LruCache<(OwnedServerName, u64), bool>>,
 }
 
 impl Service {
@@ -46,19 +54,107 @@ impl Service {
         self.db.state_get(shortstatehash, event_type, state_key)
     }
 
+    pub fn state_get_content(
+        &self,
+        shortstatehash: u64,
+        event_type: &StateEventType,
+        state_key: &str,
+    ) -> Result<Option<serde_json::Value>> {
+        self.db
+            .state_get_content(shortstatehash, event_type, state_key)
+    }
+
     /// Returns the state hash for this pdu.
     pub fn pdu_shortstatehash(&self, event_id: &EventId) -> Result<Option<u64>> {
         self.db.pdu_shortstatehash(event_id)
     }
 
-    /// Returns true if a server has permission to see an event
+    /// Whether a server is allowed to see an event through federation, based on
+    /// the room's history_visibility at that event's state.
     #[tracing::instrument(skip(self))]
-    pub fn server_can_see_event<'a>(
-        &'a self,
-        sever_name: &ServerName,
+    pub fn server_can_see_event(
+        &self,
+        server_name: &ServerName,
+        room_id: &RoomId,
         event_id: &EventId,
     ) -> Result<bool> {
-        self.db.server_can_see_event(sever_name, event_id)
+        let shortstatehash = match self.pdu_shortstatehash(event_id) {
+            Ok(Some(shortstatehash)) => shortstatehash,
+            _ => return Ok(false),
+        };
+
+        if let Some(visibility) = self
+            .server_visibility_cache
+            .lock()
+            .unwrap()
+            .get_mut(&(server_name.to_owned(), shortstatehash))
+        {
+            return Ok(*visibility);
+        }
+
+        let current_server_members: Vec<OwnedUserId> = services()
+            .rooms
+            .state_cache
+            .room_members(room_id)
+            .filter(|member| {
+                member
+                    .as_ref()
+                    .map(|member| member.server_name() == server_name)
+                    .unwrap_or(true)
+            })
+            .collect::<Result<_>>()?;
+
+        let history_visibility = self
+            .state_get_content(shortstatehash, &StateEventType::RoomHistoryVisibility, "")?
+            .map(|content| match content.get("history_visibility") {
+                Some(visibility) => HistoryVisibility::from(visibility.as_str().unwrap_or("")),
+                None => HistoryVisibility::Invited,
+            });
+
+        let visibility = match history_visibility {
+            Some(HistoryVisibility::Joined) => {
+                // Look at all members in the room from this server; one of them
+                // triggered a backfill. Was one of them a member in the past,
+                // at this event?
+                let mut visible = false;
+                for member in current_server_members {
+                    if self.user_was_joined(shortstatehash, &member)? {
+                        visible = true;
+                        break;
+                    }
+                }
+                visible
+            }
+            Some(HistoryVisibility::Invited) => {
+                let mut visible = false;
+                for member in current_server_members {
+                    if self.user_was_invited(shortstatehash, &member)? {
+                        visible = true;
+                        break;
+                    }
+                }
+                visible
+            }
+            _ => false,
+        };
+
+        self.server_visibility_cache
+            .lock()
+            .unwrap()
+            .insert((server_name.to_owned(), shortstatehash), visibility);
+
+        Ok(visibility)
+    }
+
+    /// The user was a joined member at this state (potentially in the past)
+    pub fn user_was_joined(&self, shortstatehash: u64, user_id: &UserId) -> Result<bool> {
+        self.db.user_was_joined(shortstatehash, user_id)
+    }
+
+    /// The user was an invited or joined room member at this state (potentially
+    /// in the past)
+    pub fn user_was_invited(&self, shortstatehash: u64, user_id: &UserId) -> Result<bool> {
+        self.db.user_was_invited(shortstatehash, user_id)
     }
 
     /// Returns the full room state.