Merge pull request #583 from matrix-org/daniel/roomcleanupincremental

Merge all of the room membership codepaths into one

1 files changed, 225 insertions, 255 deletions

diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index 8e3c86d3a7..b00cac4bd4 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -24,7 +24,6 @@ from synapse.api.constants import (
 )
 from synapse.api.errors import AuthError, StoreError, SynapseError, Codes
 from synapse.util import stringutils, unwrapFirstError
-from synapse.util.async import run_on_reactor
 from synapse.util.logcontext import preserve_context_over_fn
 
 from signedjson.sign import verify_signed_json
@@ -42,10 +41,6 @@ logger = logging.getLogger(__name__)
 id_server_scheme = "https://"
 
 
-def collect_presencelike_data(distributor, user, content):
-    return distributor.fire("collect_presencelike_data", user, content)
-
-
 def user_left_room(distributor, user, room_id):
     return preserve_context_over_fn(
         distributor.fire,
@@ -173,9 +168,14 @@ class RoomCreationHandler(BaseHandler):
 
         creation_content = config.get("creation_content", {})
 
-        user = UserID.from_string(user_id)
-        creation_events = self._create_events_for_new_room(
-            user, room_id,
+        msg_handler = self.hs.get_handlers().message_handler
+        room_member_handler = self.hs.get_handlers().room_member_handler
+
+        yield self._send_events_for_new_room(
+            requester,
+            room_id,
+            msg_handler,
+            room_member_handler,
             preset_config=preset_config,
             invite_list=invite_list,
             initial_state=initial_state,
@@ -183,14 +183,9 @@ class RoomCreationHandler(BaseHandler):
             room_alias=room_alias,
         )
 
-        msg_handler = self.hs.get_handlers().message_handler
-
-        for event in creation_events:
-            yield msg_handler.create_and_send_event(event, ratelimit=False)
-
         if "name" in config:
             name = config["name"]
-            yield msg_handler.create_and_send_event({
+            yield msg_handler.create_and_send_nonmember_event({
                 "type": EventTypes.Name,
                 "room_id": room_id,
                 "sender": user_id,
@@ -200,7 +195,7 @@ class RoomCreationHandler(BaseHandler):
 
         if "topic" in config:
             topic = config["topic"]
-            yield msg_handler.create_and_send_event({
+            yield msg_handler.create_and_send_nonmember_event({
                 "type": EventTypes.Topic,
                 "room_id": room_id,
                 "sender": user_id,
@@ -209,13 +204,13 @@ class RoomCreationHandler(BaseHandler):
             }, ratelimit=False)
 
         for invitee in invite_list:
-            yield msg_handler.create_and_send_event({
-                "type": EventTypes.Member,
-                "state_key": invitee,
-                "room_id": room_id,
-                "sender": user_id,
-                "content": {"membership": Membership.INVITE},
-            }, ratelimit=False)
+            room_member_handler.update_membership(
+                requester,
+                UserID.from_string(invitee),
+                room_id,
+                "invite",
+                ratelimit=False,
+            )
 
         for invite_3pid in invite_3pid_list:
             id_server = invite_3pid["id_server"]
@@ -223,11 +218,11 @@ class RoomCreationHandler(BaseHandler):
             medium = invite_3pid["medium"]
             yield self.hs.get_handlers().room_member_handler.do_3pid_invite(
                 room_id,
-                user,
+                requester.user,
                 medium,
                 address,
                 id_server,
-                token_id=None,
+                requester,
                 txn_id=None,
             )
 
@@ -241,19 +236,19 @@ class RoomCreationHandler(BaseHandler):
 
         defer.returnValue(result)
 
-    def _create_events_for_new_room(self, creator, room_id, preset_config,
-                                    invite_list, initial_state, creation_content,
-                                    room_alias):
-        config = RoomCreationHandler.PRESETS_DICT[preset_config]
-
-        creator_id = creator.to_string()
-
-        event_keys = {
-            "room_id": room_id,
-            "sender": creator_id,
-            "state_key": "",
-        }
-
+    @defer.inlineCallbacks
+    def _send_events_for_new_room(
+            self,
+            creator,  # A Requester object.
+            room_id,
+            msg_handler,
+            room_member_handler,
+            preset_config,
+            invite_list,
+            initial_state,
+            creation_content,
+            room_alias
+    ):
         def create(etype, content, **kwargs):
             e = {
                 "type": etype,
@@ -265,26 +260,39 @@ class RoomCreationHandler(BaseHandler):
 
             return e
 
-        creation_content.update({"creator": creator.to_string()})
-        creation_event = create(
+        @defer.inlineCallbacks
+        def send(etype, content, **kwargs):
+            event = create(etype, content, **kwargs)
+            yield msg_handler.create_and_send_nonmember_event(event, ratelimit=False)
+
+        config = RoomCreationHandler.PRESETS_DICT[preset_config]
+
+        creator_id = creator.user.to_string()
+
+        event_keys = {
+            "room_id": room_id,
+            "sender": creator_id,
+            "state_key": "",
+        }
+
+        creation_content.update({"creator": creator_id})
+        yield send(
             etype=EventTypes.Create,
             content=creation_content,
         )
 
-        join_event = create(
-            etype=EventTypes.Member,
-            state_key=creator_id,
-            content={
-                "membership": Membership.JOIN,
-            },
+        yield room_member_handler.update_membership(
+            creator,
+            creator.user,
+            room_id,
+            "join",
+            ratelimit=False,
         )
 
-        returned_events = [creation_event, join_event]
-
         if (EventTypes.PowerLevels, '') not in initial_state:
             power_level_content = {
                 "users": {
-                    creator.to_string(): 100,
+                    creator_id: 100,
                 },
                 "users_default": 0,
                 "events": {
@@ -306,45 +314,35 @@ class RoomCreationHandler(BaseHandler):
                 for invitee in invite_list:
                     power_level_content["users"][invitee] = 100
 
-            power_levels_event = create(
+            yield send(
                 etype=EventTypes.PowerLevels,
                 content=power_level_content,
             )
 
-            returned_events.append(power_levels_event)
-
         if room_alias and (EventTypes.CanonicalAlias, '') not in initial_state:
-            room_alias_event = create(
+            yield send(
                 etype=EventTypes.CanonicalAlias,
                 content={"alias": room_alias.to_string()},
             )
 
-            returned_events.append(room_alias_event)
-
         if (EventTypes.JoinRules, '') not in initial_state:
-            join_rules_event = create(
+            yield send(
                 etype=EventTypes.JoinRules,
                 content={"join_rule": config["join_rules"]},
             )
 
-            returned_events.append(join_rules_event)
-
         if (EventTypes.RoomHistoryVisibility, '') not in initial_state:
-            history_event = create(
+            yield send(
                 etype=EventTypes.RoomHistoryVisibility,
                 content={"history_visibility": config["history_visibility"]}
             )
 
-            returned_events.append(history_event)
-
         for (etype, state_key), content in initial_state.items():
-            returned_events.append(create(
+            yield send(
                 etype=etype,
                 state_key=state_key,
                 content=content,
-            ))
-
-        return returned_events
+            )
 
 
 class RoomMemberHandler(BaseHandler):
@@ -392,7 +390,16 @@ class RoomMemberHandler(BaseHandler):
                     remotedomains.add(member.domain)
 
     @defer.inlineCallbacks
-    def update_membership(self, requester, target, room_id, action, txn_id=None):
+    def update_membership(
+            self,
+            requester,
+            target,
+            room_id,
+            action,
+            txn_id=None,
+            remote_room_hosts=None,
+            ratelimit=True,
+    ):
         effective_membership_state = action
         if action in ["kick", "unban"]:
             effective_membership_state = "leave"
@@ -401,7 +408,7 @@ class RoomMemberHandler(BaseHandler):
 
         msg_handler = self.hs.get_handlers().message_handler
 
-        content = {"membership": unicode(effective_membership_state)}
+        content = {"membership": effective_membership_state}
         if requester.is_guest:
             content["kind"] = "guest"
 
@@ -412,6 +419,9 @@ class RoomMemberHandler(BaseHandler):
                 "room_id": room_id,
                 "sender": requester.user.to_string(),
                 "state_key": target.to_string(),
+
+                # For backwards compatibility:
+                "membership": effective_membership_state,
             },
             token_id=requester.access_token_id,
             txn_id=txn_id,
@@ -432,202 +442,181 @@ class RoomMemberHandler(BaseHandler):
                 errcode=Codes.BAD_STATE
             )
 
-        yield msg_handler.send_event(
+        member_handler = self.hs.get_handlers().room_member_handler
+        yield member_handler.send_membership_event(
             event,
             context,
-            ratelimit=True,
-            is_guest=requester.is_guest
+            is_guest=requester.is_guest,
+            ratelimit=ratelimit,
+            remote_room_hosts=remote_room_hosts,
+            from_client=True,
         )
 
         if action == "forget":
             yield self.forget(requester.user, room_id)
 
     @defer.inlineCallbacks
-    def send_membership_event(self, event, context, is_guest=False):
-        """ Change the membership status of a user in a room.
+    def send_membership_event(
+            self,
+            event,
+            context,
+            is_guest=False,
+            remote_room_hosts=None,
+            ratelimit=True,
+            from_client=True,
+    ):
+        """
+        Change the membership status of a user in a room.
 
         Args:
-            event (SynapseEvent): The membership event
+            event (SynapseEvent): The membership event.
+            context: The context of the event.
+            is_guest (bool): Whether the sender is a guest.
+            room_hosts ([str]): Homeservers which are likely to already be in
+                the room, and could be danced with in order to join this
+                homeserver for the first time.
+            ratelimit (bool): Whether to rate limit this request.
+            from_client (bool): Whether this request is the result of a local
+                client request (rather than over federation). If so, we will
+                perform extra checks, like that this homeserver can act as this
+                client.
         Raises:
             SynapseError if there was a problem changing the membership.
         """
-        target_user_id = event.state_key
+        target_user = UserID.from_string(event.state_key)
+        room_id = event.room_id
 
-        prev_state = context.current_state.get(
-            (EventTypes.Member, target_user_id),
-            None
-        )
+        if from_client:
+            sender = UserID.from_string(event.sender)
+            assert self.hs.is_mine(sender), "Sender must be our own: %s" % (sender,)
 
-        room_id = event.room_id
+        message_handler = self.hs.get_handlers().message_handler
+        prev_event = message_handler.deduplicate_state_event(event, context)
+        if prev_event is not None:
+            return
 
-        # If we're trying to join a room then we have to do this differently
-        # if this HS is not currently in the room, i.e. we have to do the
-        # invite/join dance.
-        if event.membership == Membership.JOIN:
-            if is_guest:
-                guest_access = context.current_state.get(
-                    (EventTypes.GuestAccess, ""),
-                    None
-                )
-                is_guest_access_allowed = (
-                    guest_access
-                    and guest_access.content
-                    and "guest_access" in guest_access.content
-                    and guest_access.content["guest_access"] == "can_join"
-                )
-                if not is_guest_access_allowed:
-                    raise AuthError(403, "Guest access not allowed")
+        action = "send"
 
-            yield self._do_join(event, context)
-        else:
-            if event.membership == Membership.LEAVE:
-                is_host_in_room = yield self.is_host_in_room(room_id, context)
-                if not is_host_in_room:
-                    # Rejecting an invite, rather than leaving a joined room
-                    handler = self.hs.get_handlers().federation_handler
-                    inviter = yield self.get_inviter(event)
-                    if not inviter:
-                        # return the same error as join_room_alias does
-                        raise SynapseError(404, "No known servers")
-                    yield handler.do_remotely_reject_invite(
-                        [inviter.domain],
-                        room_id,
-                        event.user_id
-                    )
-                    defer.returnValue({"room_id": room_id})
-                    return
-
-            # FIXME: This isn't idempotency.
-            if prev_state and prev_state.membership == event.membership:
-                # double same action, treat this event as a NOOP.
-                defer.returnValue({})
-                return
-
-            yield self._do_local_membership_update(
-                event,
-                context=context,
+        if event.membership == Membership.JOIN:
+            if is_guest and not self._can_guest_join(context.current_state):
+                # This should be an auth check, but guests are a local concept,
+                # so don't really fit into the general auth process.
+                raise AuthError(403, "Guest access not allowed")
+            do_remote_join_dance, remote_room_hosts = self._should_do_dance(
+                context,
+                (self.get_inviter(event.state_key, context.current_state)),
+                remote_room_hosts,
             )
+            if do_remote_join_dance:
+                action = "remote_join"
+        elif event.membership == Membership.LEAVE:
+            is_host_in_room = self.is_host_in_room(context.current_state)
+            if not is_host_in_room:
+                action = "remote_reject"
 
-            if prev_state and prev_state.membership == Membership.JOIN:
-                user = UserID.from_string(event.user_id)
-                user_left_room(self.distributor, user, event.room_id)
+        federation_handler = self.hs.get_handlers().federation_handler
 
-        defer.returnValue({"room_id": room_id})
+        if action == "remote_join":
+            if len(remote_room_hosts) == 0:
+                raise SynapseError(404, "No known servers")
 
-    @defer.inlineCallbacks
-    def join_room_alias(self, joinee, room_alias, content={}):
-        directory_handler = self.hs.get_handlers().directory_handler
-        mapping = yield directory_handler.get_association(room_alias)
+            # We don't do an auth check if we are doing an invite
+            # join dance for now, since we're kinda implicitly checking
+            # that we are allowed to join when we decide whether or not we
+            # need to do the invite/join dance.
+            yield federation_handler.do_invite_join(
+                remote_room_hosts,
+                event.room_id,
+                event.user_id,
+                event.content,
+            )
+        elif action == "remote_reject":
+            inviter = self.get_inviter(target_user.to_string(), context.current_state)
+            if not inviter:
+                raise SynapseError(404, "No known servers")
+            yield federation_handler.do_remotely_reject_invite(
+                [inviter.domain],
+                room_id,
+                event.user_id
+            )
+        else:
+            yield self.handle_new_client_event(
+                event,
+                context,
+                extra_users=[target_user],
+                ratelimit=ratelimit,
+            )
 
-        if not mapping:
-            raise SynapseError(404, "No such room alias")
+        prev_member_event = context.current_state.get(
+            (EventTypes.Member, target_user.to_string()),
+            None
+        )
 
-        room_id = mapping["room_id"]
-        hosts = mapping["servers"]
-        if not hosts:
-            raise SynapseError(404, "No known servers")
+        if event.membership == Membership.JOIN:
+            if not prev_member_event or prev_member_event.membership != Membership.JOIN:
+                # Only fire user_joined_room if the user has acutally joined the
+                # room. Don't bother if the user is just changing their profile
+                # info.
+                yield user_joined_room(self.distributor, target_user, room_id)
+        elif event.membership == Membership.LEAVE:
+            if prev_member_event and prev_member_event.membership == Membership.JOIN:
+                user_left_room(self.distributor, target_user, room_id)
+
+    def _can_guest_join(self, current_state):
+        """
+        Returns whether a guest can join a room based on its current state.
+        """
+        guest_access = current_state.get((EventTypes.GuestAccess, ""), None)
+        return (
+            guest_access
+            and guest_access.content
+            and "guest_access" in guest_access.content
+            and guest_access.content["guest_access"] == "can_join"
+        )
 
-        # If event doesn't include a display name, add one.
-        yield collect_presencelike_data(self.distributor, joinee, content)
+    def _should_do_dance(self, context, inviter, room_hosts=None):
+        # TODO: Shouldn't this be remote_room_host?
+        room_hosts = room_hosts or []
 
-        content.update({"membership": Membership.JOIN})
-        builder = self.event_builder_factory.new({
-            "type": EventTypes.Member,
-            "state_key": joinee.to_string(),
-            "room_id": room_id,
-            "sender": joinee.to_string(),
-            "membership": Membership.JOIN,
-            "content": content,
-        })
-        event, context = yield self._create_new_client_event(builder)
+        is_host_in_room = self.is_host_in_room(context.current_state)
+        if is_host_in_room:
+            return False, room_hosts
 
-        yield self._do_join(event, context, room_hosts=hosts)
+        if inviter and not self.hs.is_mine(inviter):
+            room_hosts.append(inviter.domain)
 
-        defer.returnValue({"room_id": room_id})
+        return True, room_hosts
 
     @defer.inlineCallbacks
-    def _do_join(self, event, context, room_hosts=None):
-        room_id = event.room_id
-
-        # XXX: We don't do an auth check if we are doing an invite
-        # join dance for now, since we're kinda implicitly checking
-        # that we are allowed to join when we decide whether or not we
-        # need to do the invite/join dance.
-
-        is_host_in_room = yield self.is_host_in_room(room_id, context)
-        if is_host_in_room:
-            should_do_dance = False
-        elif room_hosts:  # TODO: Shouldn't this be remote_room_host?
-            should_do_dance = True
-        else:
-            inviter = yield self.get_inviter(event)
-            if not inviter:
-                # return the same error as join_room_alias does
-                raise SynapseError(404, "No known servers")
-            should_do_dance = not self.hs.is_mine(inviter)
-            room_hosts = [inviter.domain]
+    def lookup_room_alias(self, room_alias):
+        """
+        Get the room ID associated with a room alias.
 
-        if should_do_dance:
-            handler = self.hs.get_handlers().federation_handler
-            yield handler.do_invite_join(
-                room_hosts,
-                room_id,
-                event.user_id,
-                event.content,
-            )
-        else:
-            logger.debug("Doing normal join")
+        Args:
+            room_alias (RoomAlias): The alias to look up.
+        Returns:
+            A tuple of:
+                The room ID as a RoomID object.
+                Hosts likely to be participating in the room ([str]).
+        Raises:
+            SynapseError if room alias could not be found.
+        """
+        directory_handler = self.hs.get_handlers().directory_handler
+        mapping = yield directory_handler.get_association(room_alias)
 
-            yield self._do_local_membership_update(
-                event,
-                context=context,
-            )
+        if not mapping:
+            raise SynapseError(404, "No such room alias")
 
-        prev_state = context.current_state.get((event.type, event.state_key))
-        if not prev_state or prev_state.membership != Membership.JOIN:
-            # Only fire user_joined_room if the user has acutally joined the
-            # room. Don't bother if the user is just changing their profile
-            # info.
-            user = UserID.from_string(event.user_id)
-            yield user_joined_room(self.distributor, user, room_id)
+        room_id = mapping["room_id"]
+        servers = mapping["servers"]
 
-    @defer.inlineCallbacks
-    def get_inviter(self, event):
-        # TODO(markjh): get prev_state from snapshot
-        prev_state = yield self.store.get_room_member(
-            event.user_id, event.room_id
-        )
+        defer.returnValue((RoomID.from_string(room_id), servers))
 
+    def get_inviter(self, user_id, current_state):
+        prev_state = current_state.get((EventTypes.Member, user_id))
         if prev_state and prev_state.membership == Membership.INVITE:
-            defer.returnValue(UserID.from_string(prev_state.user_id))
-            return
-        elif "third_party_invite" in event.content:
-            if "sender" in event.content["third_party_invite"]:
-                inviter = UserID.from_string(
-                    event.content["third_party_invite"]["sender"]
-                )
-                defer.returnValue(inviter)
-        defer.returnValue(None)
-
-    @defer.inlineCallbacks
-    def is_host_in_room(self, room_id, context):
-        is_host_in_room = yield self.auth.check_host_in_room(
-            room_id,
-            self.hs.hostname
-        )
-        if not is_host_in_room:
-            # is *anyone* in the room?
-            room_member_keys = [
-                v for (k, v) in context.current_state.keys() if (
-                    k == "m.room.member"
-                )
-            ]
-            if len(room_member_keys) == 0:
-                # has the room been created so we can join it?
-                create_event = context.current_state.get(("m.room.create", ""))
-                if create_event:
-                    is_host_in_room = True
-        defer.returnValue(is_host_in_room)
+            return UserID.from_string(prev_state.user_id)
+        return None
 
     @defer.inlineCallbacks
     def get_joined_rooms_for_user(self, user):
@@ -645,18 +634,6 @@ class RoomMemberHandler(BaseHandler):
         defer.returnValue(room_ids)
 
     @defer.inlineCallbacks
-    def _do_local_membership_update(self, event, context):
-        yield run_on_reactor()
-
-        target_user = UserID.from_string(event.state_key)
-
-        yield self.handle_new_client_event(
-            event,
-            context,
-            extra_users=[target_user],
-        )
-
-    @defer.inlineCallbacks
     def do_3pid_invite(
             self,
             room_id,
@@ -664,7 +641,7 @@ class RoomMemberHandler(BaseHandler):
             medium,
             address,
             id_server,
-            token_id,
+            requester,
             txn_id
     ):
         invitee = yield self._lookup_3pid(
@@ -672,19 +649,12 @@ class RoomMemberHandler(BaseHandler):
         )
 
         if invitee:
-            # make sure it looks like a user ID; it'll throw if it's invalid.
-            UserID.from_string(invitee)
-            yield self.hs.get_handlers().message_handler.create_and_send_event(
-                {
-                    "type": EventTypes.Member,
-                    "content": {
-                        "membership": unicode("invite")
-                    },
-                    "room_id": room_id,
-                    "sender": inviter.to_string(),
-                    "state_key": invitee,
-                },
-                token_id=token_id,
+            handler = self.hs.get_handlers().room_member_handler
+            yield handler.update_membership(
+                requester,
+                UserID.from_string(invitee),
+                room_id,
+                "invite",
                 txn_id=txn_id,
             )
         else:
@@ -694,7 +664,7 @@ class RoomMemberHandler(BaseHandler):
                 address,
                 room_id,
                 inviter,
-                token_id,
+                requester.access_token_id,
                 txn_id=txn_id
             )
 
@@ -805,7 +775,7 @@ class RoomMemberHandler(BaseHandler):
             )
         )
         msg_handler = self.hs.get_handlers().message_handler
-        yield msg_handler.create_and_send_event(
+        yield msg_handler.create_and_send_nonmember_event(
             {
                 "type": EventTypes.ThirdPartyInvite,
                 "content": {