diff --git a/synapse/storage/user_directory.py b/synapse/storage/user_directory.py
index fea866c043..4d60a5726f 100644
--- a/synapse/storage/user_directory.py
+++ b/synapse/storage/user_directory.py
@@ -16,22 +16,314 @@
import logging
import re
-from six import iteritems
-
from twisted.internet import defer
from synapse.api.constants import EventTypes, JoinRules
+from synapse.storage.background_updates import BackgroundUpdateStore
from synapse.storage.engines import PostgresEngine, Sqlite3Engine
from synapse.storage.state import StateFilter
+from synapse.storage.state_deltas import StateDeltasStore
from synapse.types import get_domain_from_id, get_localpart_from_id
-from synapse.util.caches.descriptors import cached, cachedInlineCallbacks
-
-from ._base import SQLBaseStore
+from synapse.util.caches.descriptors import cached
logger = logging.getLogger(__name__)
-class UserDirectoryStore(SQLBaseStore):
+TEMP_TABLE = "_temp_populate_user_directory"
+
+
+class UserDirectoryStore(StateDeltasStore, BackgroundUpdateStore):
+
+ # How many records do we calculate before sending it to
+ # add_users_who_share_private_rooms?
+ SHARE_PRIVATE_WORKING_SET = 500
+
+ def __init__(self, db_conn, hs):
+ super(UserDirectoryStore, self).__init__(db_conn, hs)
+
+ self.server_name = hs.hostname
+
+ self.register_background_update_handler(
+ "populate_user_directory_createtables",
+ self._populate_user_directory_createtables,
+ )
+ self.register_background_update_handler(
+ "populate_user_directory_process_rooms",
+ self._populate_user_directory_process_rooms,
+ )
+ self.register_background_update_handler(
+ "populate_user_directory_process_users",
+ self._populate_user_directory_process_users,
+ )
+ self.register_background_update_handler(
+ "populate_user_directory_cleanup", self._populate_user_directory_cleanup
+ )
+
+ @defer.inlineCallbacks
+ def _populate_user_directory_createtables(self, progress, batch_size):
+
+ # Get all the rooms that we want to process.
+ def _make_staging_area(txn):
+ sql = (
+ "CREATE TABLE IF NOT EXISTS "
+ + TEMP_TABLE
+ + "_rooms(room_id TEXT NOT NULL, events BIGINT NOT NULL)"
+ )
+ txn.execute(sql)
+
+ sql = (
+ "CREATE TABLE IF NOT EXISTS "
+ + TEMP_TABLE
+ + "_position(position TEXT NOT NULL)"
+ )
+ txn.execute(sql)
+
+ # Get rooms we want to process from the database
+ sql = """
+ SELECT room_id, count(*) FROM current_state_events
+ GROUP BY room_id
+ """
+ txn.execute(sql)
+ rooms = [{"room_id": x[0], "events": x[1]} for x in txn.fetchall()]
+ self._simple_insert_many_txn(txn, TEMP_TABLE + "_rooms", rooms)
+ del rooms
+
+ # If search all users is on, get all the users we want to add.
+ if self.hs.config.user_directory_search_all_users:
+ sql = (
+ "CREATE TABLE IF NOT EXISTS "
+ + TEMP_TABLE
+ + "_users(user_id TEXT NOT NULL)"
+ )
+ txn.execute(sql)
+
+ txn.execute("SELECT name FROM users")
+ users = [{"user_id": x[0]} for x in txn.fetchall()]
+
+ self._simple_insert_many_txn(txn, TEMP_TABLE + "_users", users)
+
+ new_pos = yield self.get_max_stream_id_in_current_state_deltas()
+ yield self.runInteraction(
+ "populate_user_directory_temp_build", _make_staging_area
+ )
+ yield self._simple_insert(TEMP_TABLE + "_position", {"position": new_pos})
+
+ yield self._end_background_update("populate_user_directory_createtables")
+ defer.returnValue(1)
+
+ @defer.inlineCallbacks
+ def _populate_user_directory_cleanup(self, progress, batch_size):
+ """
+ Update the user directory stream position, then clean up the old tables.
+ """
+ position = yield self._simple_select_one_onecol(
+ TEMP_TABLE + "_position", None, "position"
+ )
+ yield self.update_user_directory_stream_pos(position)
+
+ def _delete_staging_area(txn):
+ txn.execute("DROP TABLE IF EXISTS " + TEMP_TABLE + "_rooms")
+ txn.execute("DROP TABLE IF EXISTS " + TEMP_TABLE + "_users")
+ txn.execute("DROP TABLE IF EXISTS " + TEMP_TABLE + "_position")
+
+ yield self.runInteraction(
+ "populate_user_directory_cleanup", _delete_staging_area
+ )
+
+ yield self._end_background_update("populate_user_directory_cleanup")
+ defer.returnValue(1)
+
+ @defer.inlineCallbacks
+ def _populate_user_directory_process_rooms(self, progress, batch_size):
+ """
+ Args:
+ progress (dict)
+ batch_size (int): Maximum number of state events to process
+ per cycle.
+ """
+ state = self.hs.get_state_handler()
+
+ # If we don't have progress filed, delete everything.
+ if not progress:
+ yield self.delete_all_from_user_dir()
+
+ def _get_next_batch(txn):
+ # Only fetch 250 rooms, so we don't fetch too many at once, even
+ # if those 250 rooms have less than batch_size state events.
+ sql = """
+ SELECT room_id, events FROM %s
+ ORDER BY events DESC
+ LIMIT 250
+ """ % (
+ TEMP_TABLE + "_rooms",
+ )
+ txn.execute(sql)
+ rooms_to_work_on = txn.fetchall()
+
+ if not rooms_to_work_on:
+ return None
+
+ # Get how many are left to process, so we can give status on how
+ # far we are in processing
+ txn.execute("SELECT COUNT(*) FROM " + TEMP_TABLE + "_rooms")
+ progress["remaining"] = txn.fetchone()[0]
+
+ return rooms_to_work_on
+
+ rooms_to_work_on = yield self.runInteraction(
+ "populate_user_directory_temp_read", _get_next_batch
+ )
+
+ # No more rooms -- complete the transaction.
+ if not rooms_to_work_on:
+ yield self._end_background_update("populate_user_directory_process_rooms")
+ defer.returnValue(1)
+
+ logger.info(
+ "Processing the next %d rooms of %d remaining"
+ % (len(rooms_to_work_on), progress["remaining"])
+ )
+
+ processed_event_count = 0
+
+ for room_id, event_count in rooms_to_work_on:
+ is_in_room = yield self.is_host_joined(room_id, self.server_name)
+
+ if is_in_room:
+ is_public = yield self.is_room_world_readable_or_publicly_joinable(
+ room_id
+ )
+
+ users_with_profile = yield state.get_current_user_in_room(room_id)
+ user_ids = set(users_with_profile)
+
+ # Update each user in the user directory.
+ for user_id, profile in users_with_profile.items():
+ yield self.update_profile_in_user_dir(
+ user_id, profile.display_name, profile.avatar_url
+ )
+
+ to_insert = set()
+
+ if is_public:
+ for user_id in user_ids:
+ if self.get_if_app_services_interested_in_user(user_id):
+ continue
+
+ to_insert.add(user_id)
+
+ if to_insert:
+ yield self.add_users_in_public_rooms(room_id, to_insert)
+ to_insert.clear()
+ else:
+ for user_id in user_ids:
+ if not self.hs.is_mine_id(user_id):
+ continue
+
+ if self.get_if_app_services_interested_in_user(user_id):
+ continue
+
+ for other_user_id in user_ids:
+ if user_id == other_user_id:
+ continue
+
+ user_set = (user_id, other_user_id)
+ to_insert.add(user_set)
+
+ # If it gets too big, stop and write to the database
+ # to prevent storing too much in RAM.
+ if len(to_insert) >= self.SHARE_PRIVATE_WORKING_SET:
+ yield self.add_users_who_share_private_room(
+ room_id, to_insert
+ )
+ to_insert.clear()
+
+ if to_insert:
+ yield self.add_users_who_share_private_room(room_id, to_insert)
+ to_insert.clear()
+
+ # We've finished a room. Delete it from the table.
+ yield self._simple_delete_one(TEMP_TABLE + "_rooms", {"room_id": room_id})
+ # Update the remaining counter.
+ progress["remaining"] -= 1
+ yield self.runInteraction(
+ "populate_user_directory",
+ self._background_update_progress_txn,
+ "populate_user_directory_process_rooms",
+ progress,
+ )
+
+ processed_event_count += event_count
+
+ if processed_event_count > batch_size:
+ # Don't process any more rooms, we've hit our batch size.
+ defer.returnValue(processed_event_count)
+
+ defer.returnValue(processed_event_count)
+
+ @defer.inlineCallbacks
+ def _populate_user_directory_process_users(self, progress, batch_size):
+ """
+ If search_all_users is enabled, add all of the users to the user directory.
+ """
+ if not self.hs.config.user_directory_search_all_users:
+ yield self._end_background_update("populate_user_directory_process_users")
+ defer.returnValue(1)
+
+ def _get_next_batch(txn):
+ sql = "SELECT user_id FROM %s LIMIT %s" % (
+ TEMP_TABLE + "_users",
+ str(batch_size),
+ )
+ txn.execute(sql)
+ users_to_work_on = txn.fetchall()
+
+ if not users_to_work_on:
+ return None
+
+ users_to_work_on = [x[0] for x in users_to_work_on]
+
+ # Get how many are left to process, so we can give status on how
+ # far we are in processing
+ sql = "SELECT COUNT(*) FROM " + TEMP_TABLE + "_users"
+ txn.execute(sql)
+ progress["remaining"] = txn.fetchone()[0]
+
+ return users_to_work_on
+
+ users_to_work_on = yield self.runInteraction(
+ "populate_user_directory_temp_read", _get_next_batch
+ )
+
+ # No more users -- complete the transaction.
+ if not users_to_work_on:
+ yield self._end_background_update("populate_user_directory_process_users")
+ defer.returnValue(1)
+
+ logger.info(
+ "Processing the next %d users of %d remaining"
+ % (len(users_to_work_on), progress["remaining"])
+ )
+
+ for user_id in users_to_work_on:
+ profile = yield self.get_profileinfo(get_localpart_from_id(user_id))
+ yield self.update_profile_in_user_dir(
+ user_id, profile.display_name, profile.avatar_url
+ )
+
+ # We've finished processing a user. Delete it from the table.
+ yield self._simple_delete_one(TEMP_TABLE + "_users", {"user_id": user_id})
+ # Update the remaining counter.
+ progress["remaining"] -= 1
+ yield self.runInteraction(
+ "populate_user_directory",
+ self._background_update_progress_txn,
+ "populate_user_directory_process_users",
+ progress,
+ )
+
+ defer.returnValue(len(users_to_work_on))
+
@defer.inlineCallbacks
def is_room_world_readable_or_publicly_joinable(self, room_id):
"""Check if the room is either world_readable or publically joinable
@@ -63,106 +355,16 @@ class UserDirectoryStore(SQLBaseStore):
defer.returnValue(False)
- @defer.inlineCallbacks
- def add_users_to_public_room(self, room_id, user_ids):
- """Add user to the list of users in public rooms
-
- Args:
- room_id (str): A room_id that all users are in that is world_readable
- or publically joinable
- user_ids (list(str)): Users to add
+ def update_profile_in_user_dir(self, user_id, display_name, avatar_url):
"""
- yield self._simple_insert_many(
- table="users_in_public_rooms",
- values=[{"user_id": user_id, "room_id": room_id} for user_id in user_ids],
- desc="add_users_to_public_room",
- )
- for user_id in user_ids:
- self.get_user_in_public_room.invalidate((user_id,))
-
- def add_profiles_to_user_dir(self, room_id, users_with_profile):
- """Add profiles to the user directory
-
- Args:
- room_id (str): A room_id that all users are joined to
- users_with_profile (dict): Users to add to directory in the form of
- mapping of user_id -> ProfileInfo
+ Update or add a user's profile in the user directory.
"""
- if isinstance(self.database_engine, PostgresEngine):
- # We weight the loclpart most highly, then display name and finally
- # server name
- sql = """
- INSERT INTO user_directory_search(user_id, vector)
- VALUES (?,
- setweight(to_tsvector('english', ?), 'A')
- || setweight(to_tsvector('english', ?), 'D')
- || setweight(to_tsvector('english', COALESCE(?, '')), 'B')
- )
- """
- args = (
- (
- user_id,
- get_localpart_from_id(user_id),
- get_domain_from_id(user_id),
- profile.display_name,
- )
- for user_id, profile in iteritems(users_with_profile)
- )
- elif isinstance(self.database_engine, Sqlite3Engine):
- sql = """
- INSERT INTO user_directory_search(user_id, value)
- VALUES (?,?)
- """
- args = (
- (
- user_id,
- "%s %s" % (user_id, p.display_name) if p.display_name else user_id,
- )
- for user_id, p in iteritems(users_with_profile)
- )
- else:
- # This should be unreachable.
- raise Exception("Unrecognized database engine")
-
- def _add_profiles_to_user_dir_txn(txn):
- txn.executemany(sql, args)
- self._simple_insert_many_txn(
- txn,
- table="user_directory",
- values=[
- {
- "user_id": user_id,
- "room_id": room_id,
- "display_name": profile.display_name,
- "avatar_url": profile.avatar_url,
- }
- for user_id, profile in iteritems(users_with_profile)
- ],
- )
- for user_id in users_with_profile:
- txn.call_after(self.get_user_in_directory.invalidate, (user_id,))
- return self.runInteraction(
- "add_profiles_to_user_dir", _add_profiles_to_user_dir_txn
- )
-
- @defer.inlineCallbacks
- def update_user_in_user_dir(self, user_id, room_id):
- yield self._simple_update_one(
- table="user_directory",
- keyvalues={"user_id": user_id},
- updatevalues={"room_id": room_id},
- desc="update_user_in_user_dir",
- )
- self.get_user_in_directory.invalidate((user_id,))
-
- def update_profile_in_user_dir(self, user_id, display_name, avatar_url, room_id):
def _update_profile_in_user_dir_txn(txn):
new_entry = self._simple_upsert_txn(
txn,
table="user_directory",
keyvalues={"user_id": user_id},
- insertion_values={"room_id": room_id},
values={"display_name": display_name, "avatar_url": avatar_url},
lock=False, # We're only inserter
)
@@ -250,16 +452,6 @@ class UserDirectoryStore(SQLBaseStore):
"update_profile_in_user_dir", _update_profile_in_user_dir_txn
)
- @defer.inlineCallbacks
- def update_user_in_public_user_list(self, user_id, room_id):
- yield self._simple_update_one(
- table="users_in_public_rooms",
- keyvalues={"user_id": user_id},
- updatevalues={"room_id": room_id},
- desc="update_user_in_public_user_list",
- )
- self.get_user_in_public_room.invalidate((user_id,))
-
def remove_from_user_dir(self, user_id):
def _remove_from_user_dir_txn(txn):
self._simple_delete_txn(
@@ -271,234 +463,154 @@ class UserDirectoryStore(SQLBaseStore):
self._simple_delete_txn(
txn, table="users_in_public_rooms", keyvalues={"user_id": user_id}
)
+ self._simple_delete_txn(
+ txn,
+ table="users_who_share_private_rooms",
+ keyvalues={"user_id": user_id},
+ )
+ self._simple_delete_txn(
+ txn,
+ table="users_who_share_private_rooms",
+ keyvalues={"other_user_id": user_id},
+ )
txn.call_after(self.get_user_in_directory.invalidate, (user_id,))
- txn.call_after(self.get_user_in_public_room.invalidate, (user_id,))
return self.runInteraction("remove_from_user_dir", _remove_from_user_dir_txn)
@defer.inlineCallbacks
- def remove_from_user_in_public_room(self, user_id):
- yield self._simple_delete(
- table="users_in_public_rooms",
- keyvalues={"user_id": user_id},
- desc="remove_from_user_in_public_room",
- )
- self.get_user_in_public_room.invalidate((user_id,))
-
- def get_users_in_public_due_to_room(self, room_id):
- """Get all user_ids that are in the room directory because they're
- in the given room_id
- """
- return self._simple_select_onecol(
- table="users_in_public_rooms",
- keyvalues={"room_id": room_id},
- retcol="user_id",
- desc="get_users_in_public_due_to_room",
- )
-
- @defer.inlineCallbacks
def get_users_in_dir_due_to_room(self, room_id):
"""Get all user_ids that are in the room directory because they're
in the given room_id
"""
- user_ids_dir = yield self._simple_select_onecol(
- table="user_directory",
- keyvalues={"room_id": room_id},
- retcol="user_id",
- desc="get_users_in_dir_due_to_room",
- )
-
- user_ids_pub = yield self._simple_select_onecol(
+ user_ids_share_pub = yield self._simple_select_onecol(
table="users_in_public_rooms",
keyvalues={"room_id": room_id},
retcol="user_id",
desc="get_users_in_dir_due_to_room",
)
- user_ids_share = yield self._simple_select_onecol(
- table="users_who_share_rooms",
+ user_ids_share_priv = yield self._simple_select_onecol(
+ table="users_who_share_private_rooms",
keyvalues={"room_id": room_id},
- retcol="user_id",
+ retcol="other_user_id",
desc="get_users_in_dir_due_to_room",
)
- user_ids = set(user_ids_dir)
- user_ids.update(user_ids_pub)
- user_ids.update(user_ids_share)
+ user_ids = set(user_ids_share_pub)
+ user_ids.update(user_ids_share_priv)
defer.returnValue(user_ids)
- @defer.inlineCallbacks
- def get_all_rooms(self):
- """Get all room_ids we've ever known about, in ascending order of "size"
- """
- sql = """
- SELECT room_id FROM current_state_events
- GROUP BY room_id
- ORDER BY count(*) ASC
- """
- rows = yield self._execute("get_all_rooms", None, sql)
- defer.returnValue([room_id for room_id, in rows])
-
- @defer.inlineCallbacks
- def get_all_local_users(self):
- """Get all local users
- """
- sql = """
- SELECT name FROM users
- """
- rows = yield self._execute("get_all_local_users", None, sql)
- defer.returnValue([name for name, in rows])
-
- def add_users_who_share_room(self, room_id, share_private, user_id_tuples):
- """Insert entries into the users_who_share_rooms table. The first
+ def add_users_who_share_private_room(self, room_id, user_id_tuples):
+ """Insert entries into the users_who_share_private_rooms table. The first
user should be a local user.
Args:
room_id (str)
- share_private (bool): Is the room private
user_id_tuples([(str, str)]): iterable of 2-tuple of user IDs.
"""
def _add_users_who_share_room_txn(txn):
- self._simple_insert_many_txn(
+ self._simple_upsert_many_txn(
txn,
- table="users_who_share_rooms",
- values=[
- {
- "user_id": user_id,
- "other_user_id": other_user_id,
- "room_id": room_id,
- "share_private": share_private,
- }
+ table="users_who_share_private_rooms",
+ key_names=["user_id", "other_user_id", "room_id"],
+ key_values=[
+ (user_id, other_user_id, room_id)
for user_id, other_user_id in user_id_tuples
],
+ value_names=(),
+ value_values=None,
)
- for user_id, other_user_id in user_id_tuples:
- txn.call_after(
- self.get_users_who_share_room_from_dir.invalidate, (user_id,)
- )
- txn.call_after(
- self.get_if_users_share_a_room.invalidate, (user_id, other_user_id)
- )
return self.runInteraction(
"add_users_who_share_room", _add_users_who_share_room_txn
)
- def update_users_who_share_room(self, room_id, share_private, user_id_sets):
- """Updates entries in the users_who_share_rooms table. The first
+ def add_users_in_public_rooms(self, room_id, user_ids):
+ """Insert entries into the users_who_share_private_rooms table. The first
user should be a local user.
Args:
room_id (str)
- share_private (bool): Is the room private
- user_id_tuples([(str, str)]): iterable of 2-tuple of user IDs.
+ user_ids (list[str])
"""
- def _update_users_who_share_room_txn(txn):
- sql = """
- UPDATE users_who_share_rooms
- SET room_id = ?, share_private = ?
- WHERE user_id = ? AND other_user_id = ?
- """
- txn.executemany(
- sql, ((room_id, share_private, uid, oid) for uid, oid in user_id_sets)
+ def _add_users_in_public_rooms_txn(txn):
+
+ self._simple_upsert_many_txn(
+ txn,
+ table="users_in_public_rooms",
+ key_names=["user_id", "room_id"],
+ key_values=[(user_id, room_id) for user_id in user_ids],
+ value_names=(),
+ value_values=None,
)
- for user_id, other_user_id in user_id_sets:
- txn.call_after(
- self.get_users_who_share_room_from_dir.invalidate, (user_id,)
- )
- txn.call_after(
- self.get_if_users_share_a_room.invalidate, (user_id, other_user_id)
- )
return self.runInteraction(
- "update_users_who_share_room", _update_users_who_share_room_txn
+ "add_users_in_public_rooms", _add_users_in_public_rooms_txn
)
- def remove_user_who_share_room(self, user_id, other_user_id):
- """Deletes entries in the users_who_share_rooms table. The first
+ def remove_user_who_share_room(self, user_id, room_id):
+ """
+ Deletes entries in the users_who_share_*_rooms table. The first
user should be a local user.
Args:
+ user_id (str)
room_id (str)
- share_private (bool): Is the room private
- user_id_tuples([(str, str)]): iterable of 2-tuple of user IDs.
"""
def _remove_user_who_share_room_txn(txn):
self._simple_delete_txn(
txn,
- table="users_who_share_rooms",
- keyvalues={"user_id": user_id, "other_user_id": other_user_id},
+ table="users_who_share_private_rooms",
+ keyvalues={"user_id": user_id, "room_id": room_id},
)
- txn.call_after(
- self.get_users_who_share_room_from_dir.invalidate, (user_id,)
+ self._simple_delete_txn(
+ txn,
+ table="users_who_share_private_rooms",
+ keyvalues={"other_user_id": user_id, "room_id": room_id},
)
- txn.call_after(
- self.get_if_users_share_a_room.invalidate, (user_id, other_user_id)
+ self._simple_delete_txn(
+ txn,
+ table="users_in_public_rooms",
+ keyvalues={"user_id": user_id, "room_id": room_id},
)
return self.runInteraction(
"remove_user_who_share_room", _remove_user_who_share_room_txn
)
- @cached(max_entries=500000)
- def get_if_users_share_a_room(self, user_id, other_user_id):
- """Gets if users share a room.
-
- Args:
- user_id (str): Must be a local user_id
- other_user_id (str)
-
- Returns:
- bool|None: None if they don't share a room, otherwise whether they
- share a private room or not.
+ @defer.inlineCallbacks
+ def get_user_dir_rooms_user_is_in(self, user_id):
"""
- return self._simple_select_one_onecol(
- table="users_who_share_rooms",
- keyvalues={"user_id": user_id, "other_user_id": other_user_id},
- retcol="share_private",
- allow_none=True,
- desc="get_if_users_share_a_room",
- )
-
- @cachedInlineCallbacks(max_entries=500000, iterable=True)
- def get_users_who_share_room_from_dir(self, user_id):
- """Returns the set of users who share a room with `user_id`
+ Returns the rooms that a user is in.
Args:
user_id(str): Must be a local user
Returns:
- dict: user_id -> share_private mapping
+ list: user_id
"""
- rows = yield self._simple_select_list(
- table="users_who_share_rooms",
+ rows = yield self._simple_select_onecol(
+ table="users_who_share_private_rooms",
keyvalues={"user_id": user_id},
- retcols=("other_user_id", "share_private"),
- desc="get_users_who_share_room_with_user",
+ retcol="room_id",
+ desc="get_rooms_user_is_in",
)
- defer.returnValue({row["other_user_id"]: row["share_private"] for row in rows})
-
- def get_users_in_share_dir_with_room_id(self, user_id, room_id):
- """Get all user tuples that are in the users_who_share_rooms due to the
- given room_id.
-
- Returns:
- [(user_id, other_user_id)]: where one of the two will match the given
- user_id.
- """
- sql = """
- SELECT user_id, other_user_id FROM users_who_share_rooms
- WHERE room_id = ? AND (user_id = ? OR other_user_id = ?)
- """
- return self._execute(
- "get_users_in_share_dir_with_room_id", None, sql, room_id, user_id, user_id
+ pub_rows = yield self._simple_select_onecol(
+ table="users_in_public_rooms",
+ keyvalues={"user_id": user_id},
+ retcol="room_id",
+ desc="get_rooms_user_is_in",
)
+ users = set(pub_rows)
+ users.update(rows)
+ defer.returnValue(list(users))
+
@defer.inlineCallbacks
def get_rooms_in_common_for_users(self, user_id, other_user_id):
"""Given two user_ids find out the list of rooms they share.
@@ -533,11 +645,8 @@ class UserDirectoryStore(SQLBaseStore):
txn.execute("DELETE FROM user_directory")
txn.execute("DELETE FROM user_directory_search")
txn.execute("DELETE FROM users_in_public_rooms")
- txn.execute("DELETE FROM users_who_share_rooms")
+ txn.execute("DELETE FROM users_who_share_private_rooms")
txn.call_after(self.get_user_in_directory.invalidate_all)
- txn.call_after(self.get_user_in_public_room.invalidate_all)
- txn.call_after(self.get_users_who_share_room_from_dir.invalidate_all)
- txn.call_after(self.get_if_users_share_a_room.invalidate_all)
return self.runInteraction(
"delete_all_from_user_dir", _delete_all_from_user_dir_txn
@@ -548,21 +657,11 @@ class UserDirectoryStore(SQLBaseStore):
return self._simple_select_one(
table="user_directory",
keyvalues={"user_id": user_id},
- retcols=("room_id", "display_name", "avatar_url"),
+ retcols=("display_name", "avatar_url"),
allow_none=True,
desc="get_user_in_directory",
)
- @cached()
- def get_user_in_public_room(self, user_id):
- return self._simple_select_one(
- table="users_in_public_rooms",
- keyvalues={"user_id": user_id},
- retcols=("room_id",),
- allow_none=True,
- desc="get_user_in_public_room",
- )
-
def get_user_directory_stream_pos(self):
return self._simple_select_one_onecol(
table="user_directory_stream_pos",
@@ -579,59 +678,6 @@ class UserDirectoryStore(SQLBaseStore):
desc="update_user_directory_stream_pos",
)
- def get_current_state_deltas(self, prev_stream_id):
- prev_stream_id = int(prev_stream_id)
- if not self._curr_state_delta_stream_cache.has_any_entity_changed(
- prev_stream_id
- ):
- return []
-
- def get_current_state_deltas_txn(txn):
- # First we calculate the max stream id that will give us less than
- # N results.
- # We arbitarily limit to 100 stream_id entries to ensure we don't
- # select toooo many.
- sql = """
- SELECT stream_id, count(*)
- FROM current_state_delta_stream
- WHERE stream_id > ?
- GROUP BY stream_id
- ORDER BY stream_id ASC
- LIMIT 100
- """
- txn.execute(sql, (prev_stream_id,))
-
- total = 0
- max_stream_id = prev_stream_id
- for max_stream_id, count in txn:
- total += count
- if total > 100:
- # We arbitarily limit to 100 entries to ensure we don't
- # select toooo many.
- break
-
- # Now actually get the deltas
- sql = """
- SELECT stream_id, room_id, type, state_key, event_id, prev_event_id
- FROM current_state_delta_stream
- WHERE ? < stream_id AND stream_id <= ?
- ORDER BY stream_id ASC
- """
- txn.execute(sql, (prev_stream_id, max_stream_id))
- return self.cursor_to_dict(txn)
-
- return self.runInteraction(
- "get_current_state_deltas", get_current_state_deltas_txn
- )
-
- def get_max_stream_id_in_current_state_deltas(self):
- return self._simple_select_one_onecol(
- table="current_state_delta_stream",
- keyvalues={},
- retcol="COALESCE(MAX(stream_id), -1)",
- desc="get_max_stream_id_in_current_state_deltas",
- )
-
@defer.inlineCallbacks
def search_user_dir(self, user_id, search_term, limit):
"""Searches for users in directory
@@ -652,22 +698,19 @@ class UserDirectoryStore(SQLBaseStore):
"""
if self.hs.config.user_directory_search_all_users:
- # make s.user_id null to keep the ordering algorithm happy
- join_clause = """
- CROSS JOIN (SELECT NULL as user_id) AS s
- """
- join_args = ()
- where_clause = "1=1"
+ join_args = (user_id,)
+ where_clause = "user_id != ?"
else:
- join_clause = """
- LEFT JOIN users_in_public_rooms AS p USING (user_id)
- LEFT JOIN (
- SELECT other_user_id AS user_id FROM users_who_share_rooms
- WHERE user_id = ? AND share_private
- ) AS s USING (user_id)
- """
join_args = (user_id,)
- where_clause = "(s.user_id IS NOT NULL OR p.user_id IS NOT NULL)"
+ where_clause = """
+ (
+ EXISTS (select 1 from users_in_public_rooms WHERE user_id = t.user_id)
+ OR EXISTS (
+ SELECT 1 FROM users_who_share_private_rooms
+ WHERE user_id = ? AND other_user_id = t.user_id
+ )
+ )
+ """
if isinstance(self.database_engine, PostgresEngine):
full_query, exact_query, prefix_query = _parse_query_postgres(search_term)
@@ -679,14 +722,13 @@ class UserDirectoryStore(SQLBaseStore):
# search: (domain, _, display name, localpart)
sql = """
SELECT d.user_id AS user_id, display_name, avatar_url
- FROM user_directory_search
+ FROM user_directory_search as t
INNER JOIN user_directory AS d USING (user_id)
- %s
WHERE
%s
AND vector @@ to_tsquery('english', ?)
ORDER BY
- (CASE WHEN s.user_id IS NOT NULL THEN 4.0 ELSE 1.0 END)
+ (CASE WHEN d.user_id IS NOT NULL THEN 4.0 ELSE 1.0 END)
* (CASE WHEN display_name IS NOT NULL THEN 1.2 ELSE 1.0 END)
* (CASE WHEN avatar_url IS NOT NULL THEN 1.2 ELSE 1.0 END)
* (
@@ -708,7 +750,6 @@ class UserDirectoryStore(SQLBaseStore):
avatar_url IS NULL
LIMIT ?
""" % (
- join_clause,
where_clause,
)
args = join_args + (full_query, exact_query, prefix_query, limit + 1)
@@ -717,9 +758,8 @@ class UserDirectoryStore(SQLBaseStore):
sql = """
SELECT d.user_id AS user_id, display_name, avatar_url
- FROM user_directory_search
+ FROM user_directory_search as t
INNER JOIN user_directory AS d USING (user_id)
- %s
WHERE
%s
AND value MATCH ?
@@ -729,7 +769,6 @@ class UserDirectoryStore(SQLBaseStore):
avatar_url IS NULL
LIMIT ?
""" % (
- join_clause,
where_clause,
)
args = join_args + (search_query, limit + 1)
|
