MatrixSynapse/synapse/handlers/user_directory.py

# -*- 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 six import iteritems, iterkeys

from twisted.internet import defer

import synapse.metrics
from synapse.api.constants import EventTypes, JoinRules, Membership
from synapse.handlers.state_deltas import StateDeltasHandler
from synapse.metrics.background_process_metrics import run_as_background_process
from synapse.storage.roommember import ProfileInfo
from synapse.types import get_localpart_from_id
from synapse.util.metrics import Measure

logger = logging.getLogger(__name__)


class UserDirectoryHandler(StateDeltasHandler):
    """Handles querying of and keeping updated the user_directory.

    N.B.: ASSUMES IT IS THE ONLY THING THAT MODIFIES THE USER DIRECTORY

    The user directory is filled with users who this server can see are joined to a
    world_readable or publically joinable room. We keep a database table up to date
    by streaming changes of the current state and recalculating whether users should
    be in the directory or not when necessary.
    """

    def __init__(self, hs):
        super(UserDirectoryHandler, self).__init__(hs)

        self.store = hs.get_datastore()
        self.state = hs.get_state_handler()
        self.server_name = hs.hostname
        self.clock = hs.get_clock()
        self.notifier = hs.get_notifier()
        self.is_mine_id = hs.is_mine_id
        self.update_user_directory = hs.config.update_user_directory
        self.search_all_users = hs.config.user_directory_search_all_users
        # The current position in the current_state_delta stream
        self.pos = None

        # Guard to ensure we only process deltas one at a time
        self._is_processing = False

        if self.update_user_directory:
            self.notifier.add_replication_callback(self.notify_new_event)

            # We kick this off so that we don't have to wait for a change before
            # we start populating the user directory
            self.clock.call_later(0, self.notify_new_event)

    def search_users(self, user_id, search_term, limit):
        """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>
                        }
                    ]
                }
        """
        return self.store.search_user_dir(user_id, search_term, limit)

    def notify_new_event(self):
        """Called when there may be more deltas to process
        """
        if not self.update_user_directory:
            return

        if self._is_processing:
            return

        @defer.inlineCallbacks
        def process():
            try:
                yield self._unsafe_process()
            finally:
                self._is_processing = False

        self._is_processing = True
        run_as_background_process("user_directory.notify_new_event", process)

    @defer.inlineCallbacks
    def handle_local_profile_change(self, user_id, profile):
        """Called to update index of our local user profiles when they change
        irrespective of any rooms the user may be in.
        """
        # FIXME(#3714): We should probably do this in the same worker as all
        # the other changes.
        is_support = yield self.store.is_support_user(user_id)
        # Support users are for diagnostics and should not appear in the user directory.
        if not is_support:
            yield self.store.update_profile_in_user_dir(
                user_id, profile.display_name, profile.avatar_url
            )

    @defer.inlineCallbacks
    def handle_user_deactivated(self, user_id):
        """Called when a user ID is deactivated
        """
        # FIXME(#3714): We should probably do this in the same worker as all
        # the other changes.
        yield self.store.remove_from_user_dir(user_id)

    @defer.inlineCallbacks
    def _unsafe_process(self):
        # If self.pos is None then means we haven't fetched it from DB
        if self.pos is None:
            self.pos = yield self.store.get_user_directory_stream_pos()

        # If still None then the initial background update hasn't happened yet
        if self.pos is None:
            return None

        # Loop round handling deltas until we're up to date
        while True:
            with Measure(self.clock, "user_dir_delta"):
                room_max_stream_ordering = self.store.get_room_max_stream_ordering()
                if self.pos == room_max_stream_ordering:
                    return

                logger.debug(
                    "Processing user stats %s->%s", self.pos, room_max_stream_ordering
                )
                max_pos, deltas = yield self.store.get_current_state_deltas(
                    self.pos, room_max_stream_ordering
                )

                logger.info("Handling %d state deltas", len(deltas))
                yield self._handle_deltas(deltas)

                self.pos = max_pos

                # Expose current event processing position to prometheus
                synapse.metrics.event_processing_positions.labels("user_dir").set(
                    max_pos
                )

                yield self.store.update_user_directory_stream_pos(max_pos)

    @defer.inlineCallbacks
    def _handle_deltas(self, deltas):
        """Called with the state deltas to process
        """
        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"]

            logger.debug("Handling: %r %r, %s", typ, state_key, event_id)

            # 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
            if typ in (EventTypes.RoomHistoryVisibility, EventTypes.JoinRules):
                yield self._handle_room_publicity_change(
                    room_id, prev_event_id, event_id, typ
                )
            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 False:
                    # 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.is_host_joined(
                        room_id, self.server_name
                    )
                    if not is_in_room:
                        logger.info("Server left room: %r", room_id)
                        # 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
                    else:
                        logger.debug("Server is still in room: %r", room_id)

                is_support = yield self.store.is_support_user(state_key)
                if not is_support:
                    if change is None:
                        # Handle any profile changes
                        yield self._handle_profile_change(
                            state_key, room_id, prev_event_id, event_id
                        )
                        continue

                    if change:  # The user joined
                        event = yield self.store.get_event(event_id, allow_none=True)
                        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
                        yield self._handle_remove_user(room_id, state_key)
            else:
                logger.debug("Ignoring irrelevant type: %r", typ)

    @defer.inlineCallbacks
    def _handle_room_publicity_change(self, room_id, prev_event_id, event_id, typ):
        """Handle a room having potentially changed from/to world_readable/publically
        joinable.

        Args:
            room_id (str)
            prev_event_id (str|None): The previous event before the state change
            event_id (str|None): The new event after the state change
            typ (str): Type of the event
        """
        logger.debug("Handling change for %s: %s", typ, room_id)

        if typ == EventTypes.RoomHistoryVisibility:
            change = yield self._get_key_change(
                prev_event_id,
                event_id,
                key_name="history_visibility",
                public_value="world_readable",
            )
        elif typ == EventTypes.JoinRules:
            change = yield self._get_key_change(
                prev_event_id,
                event_id,
                key_name="join_rule",
                public_value=JoinRules.PUBLIC,
            )
        else:
            raise Exception("Invalid event type")
        # If change is None, no change. True => become world_readable/public,
        # False => was world_readable/public
        if change is None:
            logger.debug("No change")
            return

        # 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
        )

        logger.debug("Change: %r, is_public: %r", change, is_public)

        if change and not is_public:
            # If we became world readable but room isn't currently public then
            # we ignore the change
            return
        elif not change and is_public:
            # If we stopped being world readable but are still public,
            # ignore the change
            return

        users_with_profile = yield self.state.get_current_users_in_room(room_id)

        # Remove every user from the sharing tables for that room.
        for user_id in iterkeys(users_with_profile):
            yield self.store.remove_user_who_share_room(user_id, room_id)

        # Then, re-add them to the tables.
        # NOTE: this is not the most efficient method, as handle_new_user sets
        # up local_user -> other_user and other_user_whos_local -> local_user,
        # which when ran over an entire room, will result in the same values
        # being added multiple times. The batching upserts shouldn't make this
        # too bad, though.
        for user_id, profile in iteritems(users_with_profile):
            yield self._handle_new_user(room_id, user_id, profile)

    @defer.inlineCallbacks
    def _handle_local_user(self, user_id):
        """Adds a new local roomless user into the user_directory_search table.
        Used to populate up the user index when we have an
        user_directory_search_all_users specified.
        """
        logger.debug("Adding new local user to dir, %r", user_id)

        profile = yield self.store.get_profileinfo(get_localpart_from_id(user_id))

        row = yield self.store.get_user_in_directory(user_id)
        if not row:
            yield self.store.update_profile_in_user_dir(
                user_id, profile.display_name, profile.avatar_url
            )

    @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
            user_id (str)
        """
        logger.debug("Adding new user to dir, %r", user_id)

        yield self.store.update_profile_in_user_dir(
            user_id, profile.display_name, profile.avatar_url
        )

        is_public = yield self.store.is_room_world_readable_or_publicly_joinable(
            room_id
        )
        # Now we update users who share rooms with users.
        users_with_profile = yield self.state.get_current_users_in_room(room_id)

        if is_public:
            yield self.store.add_users_in_public_rooms(room_id, (user_id,))
        else:
            to_insert = set()

            # First, if they're our user then we need to update for every user
            if self.is_mine_id(user_id):

                is_appservice = self.store.get_if_app_services_interested_in_user(
                    user_id
                )

                # We don't care about appservice users.
                if not is_appservice:
                    for other_user_id in users_with_profile:
                        if user_id == other_user_id:
                            continue

                        to_insert.add((user_id, other_user_id))

            # Next we need to update for every local user in the room
            for other_user_id in users_with_profile:
                if user_id == other_user_id:
                    continue

                is_appservice = self.store.get_if_app_services_interested_in_user(
                    other_user_id
                )
                if self.is_mine_id(other_user_id) and not is_appservice:
                    to_insert.add((other_user_id, user_id))

            if to_insert:
                yield self.store.add_users_who_share_private_room(room_id, to_insert)

    @defer.inlineCallbacks
    def _handle_remove_user(self, room_id, user_id):
        """Called when we might need to remove user from directory

        Args:
            room_id (str): room_id that user left or stopped being public that
            user_id (str)
        """
        logger.debug("Removing user %r", user_id)

        # Remove user from sharing tables
        yield self.store.remove_user_who_share_room(user_id, room_id)

        # Are they still in any rooms? If not, remove them entirely.
        rooms_user_is_in = yield self.store.get_user_dir_rooms_user_is_in(user_id)

        if len(rooms_user_is_in) == 0:
            yield self.store.remove_from_user_dir(user_id)

    @defer.inlineCallbacks
    def _handle_profile_change(self, user_id, room_id, prev_event_id, event_id):
        """Check member event changes for any profile changes and update the
        database if there are.
        """
        if not prev_event_id or not event_id:
            return

        prev_event = yield self.store.get_event(prev_event_id, allow_none=True)
        event = yield self.store.get_event(event_id, allow_none=True)

        if not prev_event or not event:
            return

        if event.membership != Membership.JOIN:
            return

        prev_name = prev_event.content.get("displayname")
        new_name = event.content.get("displayname")

        prev_avatar = prev_event.content.get("avatar_url")
        new_avatar = event.content.get("avatar_url")

        if prev_name != new_name or prev_avatar != new_avatar:
            yield self.store.update_profile_in_user_dir(user_id, new_name, new_avatar)