Merge branch 'develop' into pushers

13 files changed, 130 insertions, 75 deletions

diff --git a/synapse/handlers/__init__.py b/synapse/handlers/__init__.py
index d5df3c630b..fe071a4bc2 100644
--- a/synapse/handlers/__init__.py
+++ b/synapse/handlers/__init__.py
@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014 OpenMarket Ltd
+# Copyright 2014, 2015 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py
index 97ebd98917..38af034b4d 100644
--- a/synapse/handlers/_base.py
+++ b/synapse/handlers/_base.py
@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014 OpenMarket Ltd
+# Copyright 2014, 2015 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
diff --git a/synapse/handlers/admin.py b/synapse/handlers/admin.py
index 687b343a1d..1c9e7152c7 100644
--- a/synapse/handlers/admin.py
+++ b/synapse/handlers/admin.py
@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014 OpenMarket Ltd
+# Copyright 2014, 2015 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
diff --git a/synapse/handlers/directory.py b/synapse/handlers/directory.py
index 66d3b533d9..91fceda2ac 100644
--- a/synapse/handlers/directory.py
+++ b/synapse/handlers/directory.py
@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014 OpenMarket Ltd
+# Copyright 2014, 2015 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -155,4 +155,4 @@ class DirectoryHandler(BaseHandler):
             "room_id": room_id,
             "sender": user_id,
             "content": {"aliases": aliases},
-        })
+        }, ratelimit=False)
diff --git a/synapse/handlers/events.py b/synapse/handlers/events.py
index 02202692d4..c9ade253dd 100644
--- a/synapse/handlers/events.py
+++ b/synapse/handlers/events.py
@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014 OpenMarket Ltd
+# Copyright 2014, 2015 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -46,7 +46,8 @@ class EventStreamHandler(BaseHandler):
 
     @defer.inlineCallbacks
     @log_function
-    def get_stream(self, auth_user_id, pagin_config, timeout=0):
+    def get_stream(self, auth_user_id, pagin_config, timeout=0,
+                   as_client_event=True):
         auth_user = self.hs.parse_userid(auth_user_id)
 
         try:
@@ -78,7 +79,9 @@ class EventStreamHandler(BaseHandler):
                     auth_user, room_ids, pagin_config, timeout
                 )
 
-            chunks = [self.hs.serialize_event(e) for e in events]
+            chunks = [
+                self.hs.serialize_event(e, as_client_event) for e in events
+            ]
 
             chunk = {
                 "chunk": chunks,
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index 4aec3563ac..195f7c618a 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014 OpenMarket Ltd
+# Copyright 2014, 2015 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -91,11 +91,12 @@ class FederationHandler(BaseHandler):
 
         yield run_on_reactor()
 
-        yield self.replication_layer.send_pdu(event, destinations)
+        self.replication_layer.send_pdu(event, destinations)
 
     @log_function
     @defer.inlineCallbacks
-    def on_receive_pdu(self, origin, pdu, backfilled, state=None):
+    def on_receive_pdu(self, origin, pdu, backfilled, state=None,
+                       auth_chain=None):
         """ Called by the ReplicationLayer when we have a new pdu. We need to
         do auth checks and put it through the StateHandler.
         """
@@ -150,40 +151,41 @@ class FederationHandler(BaseHandler):
         if not is_in_room and not event.internal_metadata.outlier:
             logger.debug("Got event for room we're not in.")
 
-            replication_layer = self.replication_layer
-            auth_chain = yield replication_layer.get_event_auth(
-                origin,
-                context=event.room_id,
-                event_id=event.event_id,
-            )
+            replication = self.replication_layer
+
+            if not state:
+                state, auth_chain = yield replication.get_state_for_context(
+                    origin, context=event.room_id, event_id=event.event_id,
+                )
+
+            if not auth_chain:
+                auth_chain = yield replication.get_event_auth(
+                    origin,
+                    context=event.room_id,
+                    event_id=event.event_id,
+                )
 
             for e in auth_chain:
                 e.internal_metadata.outlier = True
                 try:
-                    yield self._handle_new_event(e, fetch_missing=False)
+                    yield self._handle_new_event(e, fetch_auth_from=origin)
                 except:
                     logger.exception(
-                        "Failed to parse auth event %s",
+                        "Failed to handle auth event %s",
                         e.event_id,
                     )
 
-            if not state:
-                state = yield replication_layer.get_state_for_context(
-                    origin,
-                    context=event.room_id,
-                    event_id=event.event_id,
-                )
-
             current_state = state
 
         if state:
             for e in state:
+                logging.info("A :) %r", e)
                 e.internal_metadata.outlier = True
                 try:
                     yield self._handle_new_event(e)
                 except:
                     logger.exception(
-                        "Failed to parse state event %s",
+                        "Failed to handle state event %s",
                         e.event_id,
                     )
 
@@ -288,7 +290,7 @@ class FederationHandler(BaseHandler):
 
     @defer.inlineCallbacks
     def on_event_auth(self, event_id):
-        auth = yield self.store.get_auth_chain(event_id)
+        auth = yield self.store.get_auth_chain([event_id])
 
         for event in auth:
             event.signatures.update(
@@ -391,10 +393,10 @@ class FederationHandler(BaseHandler):
             for e in auth_chain:
                 e.internal_metadata.outlier = True
                 try:
-                    yield self._handle_new_event(e, fetch_missing=False)
+                    yield self._handle_new_event(e)
                 except:
                     logger.exception(
-                        "Failed to parse auth event %s",
+                        "Failed to handle auth event %s",
                         e.event_id,
                     )
 
@@ -403,12 +405,11 @@ class FederationHandler(BaseHandler):
                 e.internal_metadata.outlier = True
                 try:
                     yield self._handle_new_event(
-                        e,
-                        fetch_missing=True
+                        e, fetch_auth_from=target_host
                     )
                 except:
                     logger.exception(
-                        "Failed to parse state event %s",
+                        "Failed to handle state event %s",
                         e.event_id,
                     )
 
@@ -526,9 +527,12 @@ class FederationHandler(BaseHandler):
             event.signatures,
         )
 
-        yield self.replication_layer.send_pdu(new_pdu, destinations)
+        self.replication_layer.send_pdu(new_pdu, destinations)
 
-        auth_chain = yield self.store.get_auth_chain(event.event_id)
+        state_ids = [e.event_id for e in context.current_state.values()]
+        auth_chain = yield self.store.get_auth_chain(set(
+            [event.event_id] + state_ids
+        ))
 
         defer.returnValue({
             "state": context.current_state.values(),
@@ -613,13 +617,13 @@ class FederationHandler(BaseHandler):
 
     @defer.inlineCallbacks
     @log_function
-    def on_backfill_request(self, origin, context, pdu_list, limit):
-        in_room = yield self.auth.check_host_in_room(context, origin)
+    def on_backfill_request(self, origin, room_id, pdu_list, limit):
+        in_room = yield self.auth.check_host_in_room(room_id, origin)
         if not in_room:
             raise AuthError(403, "Host not in room.")
 
         events = yield self.store.get_backfill_events(
-            context,
+            room_id,
             pdu_list,
             limit
         )
@@ -678,7 +682,7 @@ class FederationHandler(BaseHandler):
 
     @defer.inlineCallbacks
     def _handle_new_event(self, event, state=None, backfilled=False,
-                          current_state=None, fetch_missing=True):
+                          current_state=None, fetch_auth_from=None):
 
         logger.debug(
             "_handle_new_event: Before annotate: %s, sigs: %s",
@@ -699,11 +703,20 @@ class FederationHandler(BaseHandler):
         known_ids = set(
             [s.event_id for s in context.auth_events.values()]
         )
+
         for e_id, _ in event.auth_events:
             if e_id not in known_ids:
-                e = yield self.store.get_event(
-                    e_id, allow_none=True,
-                )
+                e = yield self.store.get_event(e_id, allow_none=True)
+
+                if not e and fetch_auth_from is not None:
+                    # Grab the auth_chain over federation if we are missing
+                    # auth events.
+                    auth_chain = yield self.replication_layer.get_event_auth(
+                        fetch_auth_from, event.event_id, event.room_id
+                    )
+                    for auth_event in auth_chain:
+                        yield self._handle_new_event(auth_event)
+                    e = yield self.store.get_event(e_id, allow_none=True)
 
                 if not e:
                     # TODO: Do some conflict res to make sure that we're
@@ -713,7 +726,7 @@ class FederationHandler(BaseHandler):
                         event.event_id, e_id, known_ids,
                     )
                     # FIXME: How does raising AuthError work with federation?
-                    raise AuthError(403, "Auth events are stale")
+                    raise AuthError(403, "Cannot find auth event")
 
                 context.auth_events[(e.type, e.state_key)] = e
 
diff --git a/synapse/handlers/login.py b/synapse/handlers/login.py
index c98ae2cfb5..d297d71c03 100644
--- a/synapse/handlers/login.py
+++ b/synapse/handlers/login.py
@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014 OpenMarket Ltd
+# Copyright 2014, 2015 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index 49c0e98113..f2a2f16933 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014 OpenMarket Ltd
+# Copyright 2014, 2015 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -19,6 +19,7 @@ from synapse.api.constants import EventTypes, Membership
 from synapse.api.errors import RoomError
 from synapse.streams.config import PaginationConfig
 from synapse.events.validator import EventValidator
+from synapse.util.logcontext import PreserveLoggingContext
 
 from ._base import BaseHandler
 
@@ -66,7 +67,7 @@ class MessageHandler(BaseHandler):
 
     @defer.inlineCallbacks
     def get_messages(self, user_id=None, room_id=None, pagin_config=None,
-                     feedback=False):
+                     feedback=False, as_client_event=True):
         """Get messages in a room.
 
         Args:
@@ -75,6 +76,7 @@ class MessageHandler(BaseHandler):
             pagin_config (synapse.api.streams.PaginationConfig): The pagination
             config rules to apply, if any.
             feedback (bool): True to get compressed feedback with the messages
+            as_client_event (bool): True to get events in client-server format.
         Returns:
             dict: Pagination API results
         """
@@ -98,7 +100,9 @@ class MessageHandler(BaseHandler):
         )
 
         chunk = {
-            "chunk": [self.hs.serialize_event(e) for e in events],
+            "chunk": [
+                self.hs.serialize_event(e, as_client_event) for e in events
+            ],
             "start": pagin_config.from_token.to_string(),
             "end": next_token.to_string(),
         }
@@ -106,7 +110,7 @@ class MessageHandler(BaseHandler):
         defer.returnValue(chunk)
 
     @defer.inlineCallbacks
-    def create_and_send_event(self, event_dict):
+    def create_and_send_event(self, event_dict, ratelimit=True):
         """ Given a dict from a client, create and handle a new event.
 
         Creates an FrozenEvent object, filling out auth_events, prev_events,
@@ -123,7 +127,8 @@ class MessageHandler(BaseHandler):
 
         self.validator.validate_new(builder)
 
-        self.ratelimit(builder.user_id)
+        if ratelimit:
+            self.ratelimit(builder.user_id)
         # TODO(paul): Why does 'event' not have a 'user' object?
         user = self.hs.parse_userid(builder.user_id)
         assert self.hs.is_mine(user), "User must be our own: %s" % (user,)
@@ -152,6 +157,11 @@ class MessageHandler(BaseHandler):
                 context=context,
             )
 
+        if event.type == EventTypes.Message:
+            presence = self.hs.get_handlers().presence_handler
+            with PreserveLoggingContext():
+                presence.bump_presence_active_time(user)
+
         defer.returnValue(event)
 
     @defer.inlineCallbacks
@@ -204,7 +214,7 @@ class MessageHandler(BaseHandler):
 
     @defer.inlineCallbacks
     def snapshot_all_rooms(self, user_id=None, pagin_config=None,
-                           feedback=False):
+                           feedback=False, as_client_event=True):
         """Retrieve a snapshot of all rooms the user is invited or has joined.
 
         This snapshot may include messages for all rooms where the user is
@@ -215,6 +225,7 @@ class MessageHandler(BaseHandler):
             pagin_config (synapse.api.streams.PaginationConfig): The pagination
             config used to determine how many messages *PER ROOM* to return.
             feedback (bool): True to get feedback along with these messages.
+            as_client_event (bool): True to get events in client-server format.
         Returns:
             A list of dicts with "room_id" and "membership" keys for all rooms
             the user is currently invited or joined in on. Rooms where the user
@@ -256,7 +267,7 @@ class MessageHandler(BaseHandler):
             }
 
             if event.membership == Membership.INVITE:
-                d["inviter"] = event.user_id
+                d["inviter"] = event.sender
 
             rooms_ret.append(d)
 
@@ -273,7 +284,10 @@ class MessageHandler(BaseHandler):
                 end_token = now_token.copy_and_replace("room_key", token[1])
 
                 d["messages"] = {
-                    "chunk": [self.hs.serialize_event(m) for m in messages],
+                    "chunk": [
+                        self.hs.serialize_event(m, as_client_event)
+                        for m in messages
+                    ],
                     "start": start_token.to_string(),
                     "end": end_token.to_string(),
                 }
diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py
index f3abfecdee..8aeed99274 100644
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014 OpenMarket Ltd
+# Copyright 2014, 2015 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
diff --git a/synapse/handlers/profile.py b/synapse/handlers/profile.py
index 33a2c167ec..7777d3cc94 100644
--- a/synapse/handlers/profile.py
+++ b/synapse/handlers/profile.py
@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014 OpenMarket Ltd
+# Copyright 2014, 2015 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -16,7 +16,7 @@
 from twisted.internet import defer
 
 from synapse.api.errors import SynapseError, AuthError, CodeMessageException
-from synapse.api.constants import Membership
+from synapse.api.constants import EventTypes, Membership
 from synapse.util.logcontext import PreserveLoggingContext
 
 from ._base import BaseHandler
@@ -194,6 +194,8 @@ class ProfileHandler(BaseHandler):
         if not self.hs.is_mine(user):
             return
 
+        self.ratelimit(user.to_string())
+
         joins = yield self.store.get_rooms_for_user_where_membership_is(
             user.to_string(),
             [Membership.JOIN],
@@ -201,7 +203,7 @@ class ProfileHandler(BaseHandler):
 
         for j in joins:
             content = {
-                "membership": j.content["membership"],
+                "membership": Membership.JOIN,
             }
 
             yield self.distributor.fire(
@@ -210,9 +212,9 @@ class ProfileHandler(BaseHandler):
 
             msg_handler = self.hs.get_handlers().message_handler
             yield msg_handler.create_and_send_event({
-                "type": j.type,
+                "type": EventTypes.Member,
                 "room_id": j.room_id,
-                "state_key": j.state_key,
+                "state_key": user.to_string(),
                 "content": content,
-                "sender": j.state_key,
-            })
+                "sender": user.to_string()
+            }, ratelimit=False)
diff --git a/synapse/handlers/register.py b/synapse/handlers/register.py
index 15d8716455..732652c228 100644
--- a/synapse/handlers/register.py
+++ b/synapse/handlers/register.py
@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014 OpenMarket Ltd
+# Copyright 2014, 2015 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index 8567d7409d..6d0db18e51 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014 OpenMarket Ltd
+# Copyright 2014, 2015 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -130,6 +130,7 @@ class RoomCreationHandler(BaseHandler):
                 "type": EventTypes.Name,
                 "room_id": room_id,
                 "sender": user_id,
+                "state_key": "",
                 "content": {"name": name},
             })
 
@@ -139,6 +140,7 @@ class RoomCreationHandler(BaseHandler):
                 "type": EventTypes.Topic,
                 "room_id": room_id,
                 "sender": user_id,
+                "state_key": "",
                 "content": {"topic": topic},
             })
 
@@ -147,7 +149,7 @@ class RoomCreationHandler(BaseHandler):
                 "type": EventTypes.Member,
                 "state_key": invitee,
                 "room_id": room_id,
-                "user_id": user_id,
+                "sender": user_id,
                 "content": {"membership": Membership.INVITE},
             })
 
@@ -243,14 +245,12 @@ class RoomMemberHandler(BaseHandler):
         self.distributor.declare("user_left_room")
 
     @defer.inlineCallbacks
-    def get_room_members(self, room_id, membership=Membership.JOIN):
+    def get_room_members(self, room_id):
         hs = self.hs
 
-        memberships = yield self.store.get_room_members(
-            room_id=room_id, membership=membership
-        )
+        users = yield self.store.get_users_in_room(room_id)
 
-        defer.returnValue([hs.parse_userid(m.user_id) for m in memberships])
+        defer.returnValue([hs.parse_userid(u) for u in users])
 
     @defer.inlineCallbacks
     def fetch_room_distributions_into(self, room_id, localusers=None,
@@ -390,6 +390,11 @@ class RoomMemberHandler(BaseHandler):
 
         host = hosts[0]
 
+        # If event doesn't include a display name, add one.
+        yield self.distributor.fire(
+            "collect_presencelike_data", joinee, content
+        )
+
         content.update({"membership": Membership.JOIN})
         builder = self.event_builder_factory.new({
             "type": EventTypes.Member,
@@ -420,10 +425,22 @@ class RoomMemberHandler(BaseHandler):
             event.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
 
         if is_host_in_room:
             should_do_dance = False
-        elif room_host:
+        elif room_host:  # TODO: Shouldn't this be remote_room_host?
             should_do_dance = True
         else:
             # TODO(markjh): get prev_state from snapshot
@@ -437,7 +454,8 @@ class RoomMemberHandler(BaseHandler):
                 should_do_dance = not self.hs.is_mine(inviter)
                 room_host = inviter.domain
             else:
-                should_do_dance = False
+                # return the same error as join_room_alias does
+                raise SynapseError(404, "No known servers")
 
         if should_do_dance:
             handler = self.hs.get_handlers().federation_handler
@@ -524,11 +542,10 @@ class RoomListHandler(BaseHandler):
     def get_public_room_list(self):
         chunk = yield self.store.get_rooms(is_public=True)
         for room in chunk:
-            joined_members = yield self.store.get_room_members(
+            joined_users = yield self.store.get_users_in_room(
                 room_id=room["room_id"],
-                membership=Membership.JOIN
             )
-            room["num_joined_members"] = len(joined_members)
+            room["num_joined_members"] = len(joined_users)
         # FIXME (erikj): START is no longer a valid value
         defer.returnValue({"start": "START", "end": "END", "chunk": chunk})
 
diff --git a/synapse/handlers/typing.py b/synapse/handlers/typing.py
index 7626b07280..22ce7873d0 100644
--- a/synapse/handlers/typing.py
+++ b/synapse/handlers/typing.py
@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014 OpenMarket Ltd
+# Copyright 2014, 2015 OpenMarket Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -83,9 +83,15 @@ class TypingNotificationHandler(BaseHandler):
         if member in self._member_typing_timer:
             self.clock.cancel_call_later(self._member_typing_timer[member])
 
+        def _cb():
+            logger.debug(
+                "%s has timed out in %s", target_user.to_string(), room_id
+            )
+            self._stopped_typing(member)
+
         self._member_typing_until[member] = until
         self._member_typing_timer[member] = self.clock.call_later(
-            timeout / 1000, lambda: self._stopped_typing(member)
+            timeout / 1000.0, _cb
         )
 
         if was_present: