MatrixSynapse/synapse/handlers/user_directory.py

340 lines
12 KiB
Python
Raw Normal View History

2017-05-31 12:51:01 +02:00
# -*- coding: utf-8 -*-
# Copyright 2017 Vector Creations Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import logging
from twisted.internet import defer
from synapse.api.constants import EventTypes, JoinRules, Membership
from synapse.storage.roommember import ProfileInfo
from synapse.util.metrics import Measure
logger = logging.getLogger(__name__)
class UserDirectoyHandler(object):
2017-05-31 16:00:29 +02:00
"""Handles querying of and keeping updated the user_directory.
N.B.: ASSUMES IT IS THE ONLY THING THAT MODIFIES THE USER DIRECTORY
"""
2017-05-31 12:51:01 +02:00
def __init__(self, hs):
self.store = hs.get_datastore()
self.state = hs.get_state_handler()
self.server_name = hs.hostname
self.clock = hs.get_clock()
2017-05-31 16:00:29 +02:00
# When start up for the first time we need to populate the user_directory.
# This is a set of user_id's we've inserted already
2017-05-31 12:51:01 +02:00
self.initially_handled_users = set()
2017-05-31 16:00:29 +02:00
# The current position in the current_state_delta stream
2017-05-31 12:51:01 +02:00
self.pos = None
2017-05-31 16:00:29 +02:00
# Guard to ensure we only process deltas one at a time
2017-05-31 12:51:01 +02:00
self._is_processing = False
2017-05-31 16:00:29 +02:00
# We kick this off so that we don't have to wait for a change before
# we start populating the user directory
2017-05-31 12:59:36 +02:00
self.clock.call_later(0, self.notify_new_event)
2017-05-31 15:00:01 +02:00
def search_users(self, search_term, limit):
2017-05-31 16:00:29 +02:00
"""Searches for users in directory
Returns:
dict of the form::
{
"limited": <bool>, # whether there were more results or not
"results": [ # Ordered by best match first
{
"user_id": <user_id>,
"display_name": <display_name>,
"avatar_url": <avatar_url>
}
]
}
"""
2017-05-31 15:00:01 +02:00
return self.store.search_user_dir(search_term, limit)
2017-05-31 12:51:01 +02:00
@defer.inlineCallbacks
def notify_new_event(self):
2017-05-31 16:00:29 +02:00
"""Called when there may be more deltas to process
"""
2017-05-31 12:51:01 +02:00
if self._is_processing:
return
self._is_processing = True
try:
yield self._unsafe_process()
finally:
self._is_processing = False
@defer.inlineCallbacks
def _unsafe_process(self):
2017-05-31 16:00:29 +02:00
# If self.pos is None then means we haven't fetched it from DB
2017-05-31 12:51:01 +02:00
if self.pos is None:
self.pos = yield self.store.get_user_directory_stream_pos()
2017-05-31 16:00:29 +02:00
# If still None then we need to do the initial fill of directory
2017-05-31 12:51:01 +02:00
if self.pos is None:
yield self._do_initial_spam()
self.pos = yield self.store.get_user_directory_stream_pos()
2017-05-31 16:00:29 +02:00
# Loop round handling deltas until we're up to date
2017-05-31 12:51:01 +02:00
while True:
with Measure(self.clock, "user_dir_delta"):
deltas = yield self.store.get_current_state_deltas(self.pos)
if not deltas:
return
yield self._handle_deltas(deltas)
2017-05-31 12:55:13 +02:00
self.pos = deltas[-1]["stream_id"]
yield self.store.update_user_directory_stream_pos(self.pos)
2017-05-31 12:51:01 +02:00
@defer.inlineCallbacks
def _do_initial_spam(self):
2017-05-31 16:00:29 +02:00
"""Populates the user_directory from the current state of the DB, used
when synapse first starts with user_directory support
"""
2017-05-31 12:56:27 +02:00
# TODO: pull from current delta stream_id
new_pos = self.store.get_room_max_stream_ordering()
2017-05-31 16:00:29 +02:00
# Delete any existing entries just in case there are any
2017-05-31 12:51:01 +02:00
yield self.store.delete_all_from_user_dir()
2017-05-31 16:00:29 +02:00
# We process by going through each existing room at a time.
2017-05-31 12:51:01 +02:00
room_ids = yield self.store.get_all_rooms()
for room_id in room_ids:
2017-05-31 16:00:29 +02:00
yield self._handle_intial_room(room_id)
2017-05-31 12:51:01 +02:00
self.initially_handled_users = None
2017-05-31 12:56:27 +02:00
yield self.store.update_user_directory_stream_pos(new_pos)
2017-05-31 12:51:01 +02:00
@defer.inlineCallbacks
2017-05-31 16:00:29 +02:00
def _handle_intial_room(self, room_id):
"""Called when we initially fill out user_directory one room at a time
"""
is_in_room = yield self.store.get_is_host_in_room(room_id, self.server_name)
if not is_in_room:
return
2017-05-31 12:51:01 +02:00
2017-05-31 16:00:29 +02:00
is_public = yield self.store.is_room_world_readable_or_publicly_joinable(room_id)
if not is_public:
2017-05-31 12:51:01 +02:00
return
2017-05-31 16:00:29 +02:00
users_with_profile = yield self.state.get_current_user_in_room(room_id)
unhandled_users = set(users_with_profile) - self.initially_handled_users
2017-05-31 12:51:01 +02:00
2017-05-31 16:00:29 +02:00
yield self.store.add_profiles_to_user_dir(
room_id, {
user_id: users_with_profile[user_id] for user_id in unhandled_users
}
)
2017-05-31 12:51:01 +02:00
2017-05-31 16:00:29 +02:00
self.initially_handled_users |= unhandled_users
2017-05-31 12:51:01 +02:00
@defer.inlineCallbacks
def _handle_deltas(self, deltas):
2017-05-31 16:00:29 +02:00
"""Called with the state deltas to process
"""
2017-05-31 12:51:01 +02:00
for delta in deltas:
typ = delta["type"]
state_key = delta["state_key"]
room_id = delta["room_id"]
event_id = delta["event_id"]
prev_event_id = delta["prev_event_id"]
2017-05-31 16:00:29 +02:00
# For join rule and visibility changes we need to check if the room
# may have become public or not and add/remove the users in said room
2017-05-31 12:51:01 +02:00
if typ == EventTypes.RoomHistoryVisibility:
change = yield self._get_key_change(
prev_event_id, event_id,
key_name="history_visibility",
public_value="world_readable",
)
2017-05-31 16:00:29 +02:00
# If change is None, no change. True => become world readable,
# False => was world readable
2017-05-31 12:51:01 +02:00
if change is None:
continue
2017-05-31 16:00:29 +02:00
# There's been a change to or from being world readable.
is_public = yield self.store.is_room_world_readable_or_publicly_joinable(
room_id
)
2017-05-31 16:00:29 +02:00
if change and not is_public:
# If we became world readable but room isn't currently public then
# we ignore the change
continue
2017-05-31 16:00:29 +02:00
elif not change and is_public:
# If we stopped being world readable but are still public,
# ignore the change
continue
2017-05-31 12:51:01 +02:00
users_with_profile = yield self.state.get_current_user_in_room(room_id)
for user_id, profile in users_with_profile.iteritems():
if change:
yield self._handle_new_user(room_id, user_id, profile)
else:
yield self._handle_remove_user(room_id, user_id)
elif typ == EventTypes.JoinRules:
change = yield self._get_key_change(
prev_event_id, event_id,
key_name="join_rules",
public_value=JoinRules.PUBLIC,
)
if change is None:
continue
is_public = yield self.store.is_room_world_readable_or_publicly_joinable(
room_id
)
if change and is_public:
continue
elif not change and not is_public:
continue
2017-05-31 12:51:01 +02:00
users_with_profile = yield self.state.get_current_user_in_room(room_id)
for user_id, profile in users_with_profile.iteritems():
if change:
yield self._handle_new_user(room_id, user_id, profile)
else:
yield self._handle_remove_user(room_id, user_id)
elif typ == EventTypes.Member:
change = yield self._get_key_change(
prev_event_id, event_id,
key_name="membership",
public_value=Membership.JOIN,
)
if change is None:
continue
if not change:
# Need to check if the server left the room entirely, if so
# we might need to remove all the users in that room
is_in_room = yield self.store.get_is_host_in_room(
room_id, self.server_name,
)
if not is_in_room:
# Fetch all the users that we marked as being in user
# directory due to being in the room and then check if
# need to remove those users or not
user_ids = yield self.store.get_users_in_dir_due_to_room(room_id)
for user_id in user_ids:
yield self._handle_remove_user(room_id, user_id)
return
if change: # The user joined
2017-05-31 12:51:01 +02:00
event = yield self.store.get_event(event_id)
profile = ProfileInfo(
avatar_url=event.content.get("avatar_url"),
display_name=event.content.get("displayname"),
)
yield self._handle_new_user(room_id, state_key, profile)
else: # The user left
2017-05-31 12:51:01 +02:00
yield self._handle_remove_user(room_id, state_key)
2017-05-31 16:00:29 +02:00
@defer.inlineCallbacks
def _handle_new_user(self, room_id, user_id, profile):
"""Called when we might need to add user to directory
Args:
room_id (str): room_id that user joined or started being public that
user_id (str)
"""
row = yield self.store.get_user_in_directory(user_id)
if row:
return
yield self.store.add_profiles_to_user_dir(room_id, {user_id: profile})
def _handle_remove_user(self, room_id, user_id):
"""Called when we might need to remove user to directory
Args:
room_id (str): room_id that user left or stopped being public that
user_id (str)
"""
row = yield self.store.get_user_in_directory(user_id)
if not row or row["room_id"] != room_id:
# Either the user wasn't in directory or we're still in a room that
# is public (i.e. the room_id in the database)
return
# TODO: Make this faster?
rooms = yield self.store.get_rooms_for_user(user_id)
for j_room_id in rooms:
is_public = yield self.store.is_room_world_readable_or_publicly_joinable(
j_room_id
)
if is_public:
yield self.store.update_user_in_user_dir(user_id, j_room_id)
return
yield self.store.remove_from_user_dir(user_id)
2017-05-31 12:51:01 +02:00
@defer.inlineCallbacks
def _get_key_change(self, prev_event_id, event_id, key_name, public_value):
2017-05-31 16:00:29 +02:00
"""Given two events check if the `key_name` field in content changed
from not matching `public_value` to doing so.
For example, check if `history_visibility` (`key_name`) changed from
`shared` to `world_readable` (`public_value`).
Returns:
None if the field in the events either both match `public_value` o
neither do, i.e. there has been no change.
True if it didnt match `public_value` but now does
Falsse if it did match `public_value` but now doesn't
"""
2017-05-31 12:51:01 +02:00
prev_event = None
event = None
if prev_event_id:
prev_event = yield self.store.get_event(prev_event_id, allow_none=True)
if event_id:
event = yield self.store.get_event(event_id, allow_none=True)
if not event and not prev_event:
defer.returnValue(None)
prev_hist_vis = None
hist_vis = None
if prev_event:
prev_hist_vis = prev_event.content.get(key_name, None)
if event:
hist_vis = event.content.get(key_name, None)
if hist_vis == public_value and prev_hist_vis != public_value:
defer.returnValue(True)
elif hist_vis != public_value and prev_hist_vis == public_value:
defer.returnValue(False)
else:
defer.returnValue(None)