Merge branch 'develop' into request_logging

Conflicts:
	setup.py
	synapse/storage/_base.py
	synapse/util/async.py

7 files changed, 559 insertions, 288 deletions

diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py
index de4d23bbb3..30c6733063 100644
--- a/synapse/handlers/_base.py
+++ b/synapse/handlers/_base.py
@@ -14,7 +14,18 @@
 # limitations under the License.
 
 from twisted.internet import defer
+
 from synapse.api.errors import LimitExceededError
+from synapse.util.async import run_on_reactor
+from synapse.crypto.event_signing import add_hashes_and_signatures
+from synapse.api.events.room import RoomMemberEvent
+from synapse.api.constants import Membership
+
+import logging
+
+
+logger = logging.getLogger(__name__)
+
 
 class BaseHandler(object):
 
@@ -30,6 +41,9 @@ class BaseHandler(object):
         self.clock = hs.get_clock()
         self.hs = hs
 
+        self.signing_key = hs.config.signing_key[0]
+        self.server_name = hs.hostname
+
     def ratelimit(self, user_id):
         time_now = self.clock.time()
         allowed, time_allowed = self.ratelimiter.send_message(
@@ -44,16 +58,58 @@ class BaseHandler(object):
 
     @defer.inlineCallbacks
     def _on_new_room_event(self, event, snapshot, extra_destinations=[],
-                           extra_users=[]):
+                           extra_users=[], suppress_auth=False,
+                           do_invite_host=None):
+        yield run_on_reactor()
+
         snapshot.fill_out_prev_events(event)
 
+        yield self.state_handler.annotate_event_with_state(event)
+
+        yield self.auth.add_auth_events(event)
+
+        logger.debug("Signing event...")
+
+        add_hashes_and_signatures(
+            event, self.server_name, self.signing_key
+        )
+
+        logger.debug("Signed event.")
+
+        if not suppress_auth:
+            logger.debug("Authing...")
+            self.auth.check(event, raises=True)
+            logger.debug("Authed")
+        else:
+            logger.debug("Suppressed auth.")
+
+        if do_invite_host:
+            federation_handler = self.hs.get_handlers().federation_handler
+            invite_event = yield federation_handler.send_invite(
+                do_invite_host,
+                event
+            )
+
+            # FIXME: We need to check if the remote changed anything else
+            event.signatures = invite_event.signatures
+
         yield self.store.persist_event(event)
 
         destinations = set(extra_destinations)
         # Send a PDU to all hosts who have joined the room.
-        destinations.update((yield self.store.get_joined_hosts_for_room(
-            event.room_id
-        )))
+
+        for k, s in event.state_events.items():
+            try:
+                if k[0] == RoomMemberEvent.TYPE:
+                    if s.content["membership"] == Membership.JOIN:
+                        destinations.add(
+                            self.hs.parse_userid(s.state_key).domain
+                        )
+            except:
+                logger.warn(
+                    "Failed to get destination from event %s", s.event_id
+                )
+
         event.destinations = list(destinations)
 
         self.notifier.on_new_room_event(event, extra_users=extra_users)
diff --git a/synapse/handlers/directory.py b/synapse/handlers/directory.py
index a56830d520..164363cdc5 100644
--- a/synapse/handlers/directory.py
+++ b/synapse/handlers/directory.py
@@ -147,10 +147,8 @@ class DirectoryHandler(BaseHandler):
             content={"aliases": aliases},
         )
 
-        snapshot = yield self.store.snapshot_room(
-            room_id=room_id,
-            user_id=user_id,
-        )
+        snapshot = yield self.store.snapshot_room(event)
 
-        yield self.state_handler.handle_new_event(event, snapshot)
-        yield self._on_new_room_event(event, snapshot, extra_users=[user_id])
+        yield self._on_new_room_event(
+            event, snapshot, extra_users=[user_id], suppress_auth=True
+        )
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index f52591d2a3..5e096f4652 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -17,13 +17,15 @@
 
 from ._base import BaseHandler
 
-from synapse.api.events.room import InviteJoinEvent, RoomMemberEvent
+from synapse.api.errors import AuthError, FederationError
+from synapse.api.events.room import RoomMemberEvent
 from synapse.api.constants import Membership
 from synapse.util.logutils import log_function
 from synapse.federation.pdu_codec import PduCodec
-from synapse.api.errors import SynapseError
+from synapse.util.async import run_on_reactor
+from synapse.crypto.event_signing import compute_event_signature
 
-from twisted.internet import defer, reactor
+from twisted.internet import defer
 
 import logging
 
@@ -38,6 +40,8 @@ class FederationHandler(BaseHandler):
         of the home server (including auth and state conflict resoultion)
         b) converting events that were produced by local clients that may need
         to be sent to remote home servers.
+        c) doing the necessary dances to invite remote users and join remote
+        rooms.
     """
 
     def __init__(self, hs):
@@ -62,6 +66,9 @@ class FederationHandler(BaseHandler):
 
         self.pdu_codec = PduCodec(hs)
 
+        # When joining a room we need to queue any events for that room up
+        self.room_queues = {}
+
     @log_function
     @defer.inlineCallbacks
     def handle_new_event(self, event, snapshot):
@@ -78,6 +85,8 @@ class FederationHandler(BaseHandler):
             processing.
         """
 
+        yield run_on_reactor()
+
         pdu = self.pdu_codec.pdu_from_event(event)
 
         if not hasattr(pdu, "destinations") or not pdu.destinations:
@@ -87,98 +96,91 @@ class FederationHandler(BaseHandler):
 
     @log_function
     @defer.inlineCallbacks
-    def on_receive_pdu(self, pdu, backfilled):
+    def on_receive_pdu(self, pdu, backfilled, state=None):
         """ Called by the ReplicationLayer when we have a new pdu. We need to
-        do auth checks and put it throught the StateHandler.
+        do auth checks and put it through the StateHandler.
         """
         event = self.pdu_codec.event_from_pdu(pdu)
 
         logger.debug("Got event: %s", event.event_id)
 
-        with (yield self.lock_manager.lock(pdu.context)):
-            if event.is_state and not backfilled:
-                is_new_state = yield self.state_handler.handle_new_state(
-                    pdu
-                )
-            else:
-                is_new_state = False
-        # TODO: Implement something in federation that allows us to
-        # respond to PDU.
+        # If we are currently in the process of joining this room, then we
+        # queue up events for later processing.
+        if event.room_id in self.room_queues:
+            self.room_queues[event.room_id].append(pdu)
+            return
 
-        target_is_mine = False
-        if hasattr(event, "target_host"):
-            target_is_mine = event.target_host == self.hs.hostname
-
-        if event.type == InviteJoinEvent.TYPE:
-            if not target_is_mine:
-                logger.debug("Ignoring invite/join event %s", event)
-                return
-
-            # If we receive an invite/join event then we need to join the
-            # sender to the given room.
-            # TODO: We should probably auth this or some such
-            content = event.content
-            content.update({"membership": Membership.JOIN})
-            new_event = self.event_factory.create_event(
-                etype=RoomMemberEvent.TYPE,
-                state_key=event.user_id,
-                room_id=event.room_id,
-                user_id=event.user_id,
-                membership=Membership.JOIN,
-                content=content
-            )
+        logger.debug("Processing event: %s", event.event_id)
+
+        if state:
+            state = [self.pdu_codec.event_from_pdu(p) for p in state]
+
+        is_new_state = yield self.state_handler.annotate_event_with_state(
+            event,
+            old_state=state
+        )
+
+        logger.debug("Event: %s", event)
 
-            yield self.hs.get_handlers().room_member_handler.change_membership(
-                new_event,
-                do_auth=False,
+        try:
+            self.auth.check(event, raises=True)
+        except AuthError as e:
+            raise FederationError(
+                "ERROR",
+                e.code,
+                e.msg,
+                affected=event.event_id,
             )
 
-        else:
-            with (yield self.room_lock.lock(event.room_id)):
-                yield self.store.persist_event(
-                    event,
-                    backfilled,
-                    is_new_state=is_new_state
-                )
+        is_new_state = is_new_state and not backfilled
 
-            room = yield self.store.get_room(event.room_id)
+        # TODO: Implement something in federation that allows us to
+        # respond to PDU.
 
-            if not room:
-                # Huh, let's try and get the current state
-                try:
-                    yield self.replication_layer.get_state_for_context(
-                        event.origin, event.room_id
-                    )
+        yield self.store.persist_event(
+            event,
+            backfilled,
+            is_new_state=is_new_state
+        )
 
-                    hosts = yield self.store.get_joined_hosts_for_room(
-                        event.room_id
-                    )
-                    if self.hs.hostname in hosts:
-                        try:
-                            yield self.store.store_room(
-                                room_id=event.room_id,
-                                room_creator_user_id="",
-                                is_public=False,
-                            )
-                        except:
-                            pass
-                except:
-                    logger.exception(
-                        "Failed to get current state for room %s",
-                        event.room_id
-                    )
+        room = yield self.store.get_room(event.room_id)
 
-            if not backfilled:
-                extra_users = []
-                if event.type == RoomMemberEvent.TYPE:
-                    target_user_id = event.state_key
-                    target_user = self.hs.parse_userid(target_user_id)
-                    extra_users.append(target_user)
+        if not room:
+            # Huh, let's try and get the current state
+            try:
+                yield self.replication_layer.get_state_for_context(
+                    event.origin, event.room_id, event.event_id,
+                )
 
-                yield self.notifier.on_new_room_event(
-                    event, extra_users=extra_users
+                hosts = yield self.store.get_joined_hosts_for_room(
+                    event.room_id
+                )
+                if self.hs.hostname in hosts:
+                    try:
+                        yield self.store.store_room(
+                            room_id=event.room_id,
+                            room_creator_user_id="",
+                            is_public=False,
+                        )
+                    except:
+                        pass
+            except:
+                logger.exception(
+                    "Failed to get current state for room %s",
+                    event.room_id
                 )
 
+        if not backfilled:
+            extra_users = []
+            if event.type == RoomMemberEvent.TYPE:
+                target_user_id = event.state_key
+                target_user = self.hs.parse_userid(target_user_id)
+                extra_users.append(target_user)
+
+            yield self.notifier.on_new_room_event(
+                event, extra_users=extra_users
+            )
+
         if event.type == RoomMemberEvent.TYPE:
             if event.membership == Membership.JOIN:
                 user = self.hs.parse_userid(event.state_key)
@@ -189,79 +191,366 @@ class FederationHandler(BaseHandler):
     @log_function
     @defer.inlineCallbacks
     def backfill(self, dest, room_id, limit):
-        pdus = yield self.replication_layer.backfill(dest, room_id, limit)
+        """ Trigger a backfill request to `dest` for the given `room_id`
+        """
+        extremities = yield self.store.get_oldest_events_in_room(room_id)
+
+        pdus = yield self.replication_layer.backfill(
+            dest,
+            room_id,
+            limit,
+            extremities=extremities,
+        )
 
         events = []
 
         for pdu in pdus:
             event = self.pdu_codec.event_from_pdu(pdu)
+
+            # FIXME (erikj): Not sure this actually works :/
+            yield self.state_handler.annotate_event_with_state(event)
+
             events.append(event)
+
             yield self.store.persist_event(event, backfilled=True)
 
         defer.returnValue(events)
 
+    @defer.inlineCallbacks
+    def send_invite(self, target_host, event):
+        """ Sends the invite to the remote server for signing.
+
+        Invites must be signed by the invitee's server before distribution.
+        """
+        pdu = yield self.replication_layer.send_invite(
+            destination=target_host,
+            context=event.room_id,
+            event_id=event.event_id,
+            pdu=self.pdu_codec.pdu_from_event(event)
+        )
+
+        defer.returnValue(self.pdu_codec.event_from_pdu(pdu))
+
+    @defer.inlineCallbacks
+    def on_event_auth(self, event_id):
+        auth = yield self.store.get_auth_chain(event_id)
+        defer.returnValue([self.pdu_codec.pdu_from_event(e) for e in auth])
+
     @log_function
     @defer.inlineCallbacks
     def do_invite_join(self, target_host, room_id, joinee, content, snapshot):
+        """ Attempts to join the `joinee` to the room `room_id` via the
+        server `target_host`.
+
+        This first triggers a /make_join/ request that returns a partial
+        event that we can fill out and sign. This is then sent to the
+        remote server via /send_join/ which responds with the state at that
+        event and the auth_chains.
+
+        We suspend processing of any received events from this room until we
+        have finished processing the join.
+        """
+        pdu = yield self.replication_layer.make_join(
+            target_host,
+            room_id,
+            joinee
+        )
+
+        logger.debug("Got response to make_join: %s", pdu)
 
-        hosts = yield self.store.get_joined_hosts_for_room(room_id)
-        if self.hs.hostname in hosts:
-            # We are already in the room.
-            logger.debug("We're already in the room apparently")
-            defer.returnValue(False)
+        event = self.pdu_codec.event_from_pdu(pdu)
+
+        # We should assert some things.
+        assert(event.type == RoomMemberEvent.TYPE)
+        assert(event.user_id == joinee)
+        assert(event.state_key == joinee)
+        assert(event.room_id == room_id)
+
+        event.outlier = False
+
+        self.room_queues[room_id] = []
 
-        # First get current state to see if we are already joined.
         try:
-            yield self.replication_layer.get_state_for_context(
-                target_host, room_id
+            event.event_id = self.event_factory.create_event_id()
+            event.content = content
+
+            state = yield self.replication_layer.send_join(
+                target_host,
+                self.pdu_codec.pdu_from_event(event)
             )
 
-            hosts = yield self.store.get_joined_hosts_for_room(room_id)
-            if self.hs.hostname in hosts:
-                # Oh, we were actually in the room already.
-                logger.debug("We're already in the room apparently")
-                defer.returnValue(False)
-        except Exception:
-            logger.exception("Failed to get current state")
-
-        new_event = self.event_factory.create_event(
-            etype=InviteJoinEvent.TYPE,
-            target_host=target_host,
-            room_id=room_id,
-            user_id=joinee,
-            content=content
-        )
+            state = [self.pdu_codec.event_from_pdu(p) for p in state]
 
-        new_event.destinations = [target_host]
+            logger.debug("do_invite_join state: %s", state)
 
-        snapshot.fill_out_prev_events(new_event)
-        yield self.handle_new_event(new_event, snapshot)
+            yield self.state_handler.annotate_event_with_state(
+                event,
+                old_state=state
+            )
 
-        # TODO (erikj): Time out here.
-        d = defer.Deferred()
-        self.waiting_for_join_list.setdefault((joinee, room_id), []).append(d)
-        reactor.callLater(10, d.cancel)
+            logger.debug("do_invite_join event: %s", event)
 
-        try:
-            yield d
-        except defer.CancelledError:
-            raise SynapseError(500, "Unable to join remote room")
+            try:
+                yield self.store.store_room(
+                    room_id=room_id,
+                    room_creator_user_id="",
+                    is_public=False
+                )
+            except:
+                # FIXME
+                pass
 
-        try:
-            yield self.store.store_room(
-                room_id=room_id,
-                room_creator_user_id="",
-                is_public=False
+            for e in state:
+                # FIXME: Auth these.
+                e.outlier = True
+
+                yield self.state_handler.annotate_event_with_state(
+                    e,
+                )
+
+                yield self.store.persist_event(
+                    e,
+                    backfilled=False,
+                    is_new_state=True
+                )
+
+            yield self.store.persist_event(
+                event,
+                backfilled=False,
+                is_new_state=True
             )
-        except:
-            pass
+        finally:
+            room_queue = self.room_queues[room_id]
+            del self.room_queues[room_id]
 
+            for p in room_queue:
+                try:
+                    yield self.on_receive_pdu(p, backfilled=False)
+                except:
+                    pass
 
         defer.returnValue(True)
 
+    @defer.inlineCallbacks
+    @log_function
+    def on_make_join_request(self, context, user_id):
+        """ We've received a /make_join/ request, so we create a partial
+        join event for the room and return that. We don *not* persist or
+        process it until the other server has signed it and sent it back.
+        """
+        event = self.event_factory.create_event(
+            etype=RoomMemberEvent.TYPE,
+            content={"membership": Membership.JOIN},
+            room_id=context,
+            user_id=user_id,
+            state_key=user_id,
+        )
+
+        snapshot = yield self.store.snapshot_room(event)
+        snapshot.fill_out_prev_events(event)
+
+        yield self.state_handler.annotate_event_with_state(event)
+        yield self.auth.add_auth_events(event)
+        self.auth.check(event, raises=True)
+
+        pdu = self.pdu_codec.pdu_from_event(event)
+
+        defer.returnValue(pdu)
+
+    @defer.inlineCallbacks
+    @log_function
+    def on_send_join_request(self, origin, pdu):
+        """ We have received a join event for a room. Fully process it and
+        respond with the current state and auth chains.
+        """
+        event = self.pdu_codec.event_from_pdu(pdu)
+
+        event.outlier = False
+
+        is_new_state = yield self.state_handler.annotate_event_with_state(event)
+        self.auth.check(event, raises=True)
+
+        # FIXME (erikj):  All this is duplicated above :(
+
+        yield self.store.persist_event(
+            event,
+            backfilled=False,
+            is_new_state=is_new_state
+        )
+
+        extra_users = []
+        if event.type == RoomMemberEvent.TYPE:
+            target_user_id = event.state_key
+            target_user = self.hs.parse_userid(target_user_id)
+            extra_users.append(target_user)
+
+        yield self.notifier.on_new_room_event(
+            event, extra_users=extra_users
+        )
+
+        if event.type == RoomMemberEvent.TYPE:
+            if event.membership == Membership.JOIN:
+                user = self.hs.parse_userid(event.state_key)
+                self.distributor.fire(
+                    "user_joined_room", user=user, room_id=event.room_id
+                )
+
+        new_pdu = self.pdu_codec.pdu_from_event(event)
+
+        destinations = set()
+
+        for k, s in event.state_events.items():
+            try:
+                if k[0] == RoomMemberEvent.TYPE:
+                    if s.content["membership"] == Membership.JOIN:
+                        destinations.add(
+                            self.hs.parse_userid(s.state_key).domain
+                        )
+            except:
+                logger.warn(
+                    "Failed to get destination from event %s", s.event_id
+                )
+
+        new_pdu.destinations = list(destinations)
+
+        yield self.replication_layer.send_pdu(new_pdu)
+
+        auth_chain = yield self.store.get_auth_chain(event.event_id)
+        pdu_auth_chain = [
+            self.pdu_codec.pdu_from_event(e)
+            for e in auth_chain
+        ]
+
+        defer.returnValue({
+            "state": [
+                self.pdu_codec.pdu_from_event(e)
+                for e in event.state_events.values()
+            ],
+            "auth_chain": pdu_auth_chain,
+        })
+
+    @defer.inlineCallbacks
+    def on_invite_request(self, origin, pdu):
+        """ We've got an invite event. Process and persist it. Sign it.
+
+        Respond with the now signed event.
+        """
+        event = self.pdu_codec.event_from_pdu(pdu)
+
+        event.outlier = True
+
+        event.signatures.update(
+            compute_event_signature(
+                event,
+                self.hs.hostname,
+                self.hs.config.signing_key[0]
+            )
+        )
+
+        yield self.state_handler.annotate_event_with_state(event)
+
+        yield self.store.persist_event(
+            event,
+            backfilled=False,
+        )
+
+        target_user = self.hs.parse_userid(event.state_key)
+        yield self.notifier.on_new_room_event(
+            event, extra_users=[target_user],
+        )
+
+        defer.returnValue(self.pdu_codec.pdu_from_event(event))
+
+    @defer.inlineCallbacks
+    def get_state_for_pdu(self, origin, room_id, event_id):
+        yield run_on_reactor()
+
+        in_room = yield self.auth.check_host_in_room(room_id, origin)
+        if not in_room:
+            raise AuthError(403, "Host not in room.")
+
+        state_groups = yield self.store.get_state_groups(
+            [event_id]
+        )
+
+        if state_groups:
+            _, state = state_groups.items().pop()
+            results = {
+                (e.type, e.state_key): e for e in state
+            }
+
+            event = yield self.store.get_event(event_id)
+            if hasattr(event, "state_key"):
+                # Get previous state
+                if hasattr(event, "replaces_state") and event.replaces_state:
+                    prev_event = yield self.store.get_event(
+                        event.replaces_state
+                    )
+                    results[(event.type, event.state_key)] = prev_event
+                else:
+                    del results[(event.type, event.state_key)]
+
+            defer.returnValue(
+                [
+                    self.pdu_codec.pdu_from_event(s)
+                    for s in results.values()
+                ]
+            )
+        else:
+            defer.returnValue([])
+
+    @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)
+        if not in_room:
+            raise AuthError(403, "Host not in room.")
+
+        events = yield self.store.get_backfill_events(
+            context,
+            pdu_list,
+            limit
+        )
+
+        defer.returnValue([
+            self.pdu_codec.pdu_from_event(e)
+            for e in events
+        ])
+
+    @defer.inlineCallbacks
+    @log_function
+    def get_persisted_pdu(self, origin, event_id):
+        """ Get a PDU from the database with given origin and id.
+
+        Returns:
+            Deferred: Results in a `Pdu`.
+        """
+        event = yield self.store.get_event(
+            event_id,
+            allow_none=True,
+        )
+
+        if event:
+            in_room = yield self.auth.check_host_in_room(
+                event.room_id,
+                origin
+            )
+            if not in_room:
+                raise AuthError(403, "Host not in room.")
+
+            defer.returnValue(self.pdu_codec.pdu_from_event(event))
+        else:
+            defer.returnValue(None)
+
+    @log_function
+    def get_min_depth_for_context(self, context):
+        return self.store.get_min_depth(context)
 
     @log_function
     def _on_user_joined(self, user, room_id):
-        waiters = self.waiting_for_join_list.get((user.to_string(), room_id), [])
+        waiters = self.waiting_for_join_list.get(
+            (user.to_string(), room_id),
+            []
+        )
         while waiters:
             waiters.pop().callback(None)
diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index 72894869ea..d8f8ea80cc 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -16,7 +16,6 @@
 from twisted.internet import defer
 
 from synapse.api.constants import Membership
-from synapse.api.events.room import RoomTopicEvent
 from synapse.api.errors import RoomError
 from synapse.streams.config import PaginationConfig
 from ._base import BaseHandler
@@ -26,7 +25,6 @@ import logging
 logger = logging.getLogger(__name__)
 
 
-
 class MessageHandler(BaseHandler):
 
     def __init__(self, hs):
@@ -59,7 +57,8 @@ class MessageHandler(BaseHandler):
 #            user_id=sender_id
 #        )
 
-        # TODO (erikj): Once we work out the correct c-s api we need to think on how to do this.
+        # TODO (erikj): Once we work out the correct c-s api we need to think
+        # on how to do this.
 
         defer.returnValue(None)
 
@@ -81,12 +80,11 @@ class MessageHandler(BaseHandler):
         user = self.hs.parse_userid(event.user_id)
         assert user.is_mine, "User must be our own: %s" % (user,)
 
-        snapshot = yield self.store.snapshot_room(event.room_id, event.user_id)
-
-        if not suppress_auth:
-            yield self.auth.check(event, snapshot, raises=True)
+        snapshot = yield self.store.snapshot_room(event)
 
-        yield self._on_new_room_event(event, snapshot)
+        yield self._on_new_room_event(
+            event, snapshot, suppress_auth=suppress_auth
+        )
 
         self.hs.get_handlers().presence_handler.bump_presence_active_time(
             user
@@ -111,7 +109,9 @@ class MessageHandler(BaseHandler):
         data_source = self.hs.get_event_sources().sources["room"]
 
         if not pagin_config.from_token:
-            pagin_config.from_token = yield self.hs.get_event_sources().get_current_token()
+            pagin_config.from_token = (
+                yield self.hs.get_event_sources().get_current_token()
+            )
 
         user = self.hs.parse_userid(user_id)
 
@@ -142,66 +142,27 @@ class MessageHandler(BaseHandler):
             SynapseError if something went wrong.
         """
 
-        snapshot = yield self.store.snapshot_room(
-            event.room_id,
-            event.user_id,
-            state_type=event.type,
-            state_key=event.state_key,
-        )
-
-        yield self.auth.check(event, snapshot, raises=True)
-
-        yield self.state_handler.handle_new_event(event, snapshot)
+        snapshot = yield self.store.snapshot_room(event)
 
         yield self._on_new_room_event(event, snapshot)
 
     @defer.inlineCallbacks
     def get_room_data(self, user_id=None, room_id=None,
-                      event_type=None, state_key="",
-                      public_room_rules=[],
-                      private_room_rules=["join"]):
+                      event_type=None, state_key=""):
         """ Get data from a room.
 
         Args:
             event : The room path event
-            public_room_rules : A list of membership states the user can be in,
-            in order to read this data IN A PUBLIC ROOM. An empty list means
-            'any state'.
-            private_room_rules : A list of membership states the user can be
-            in, in order to read this data IN A PRIVATE ROOM. An empty list
-            means 'any state'.
         Returns:
             The path data content.
         Raises:
             SynapseError if something went wrong.
         """
-        if event_type == RoomTopicEvent.TYPE:
-            # anyone invited/joined can read the topic
-            private_room_rules = ["invite", "join"]
-
-        # does this room exist
-        room = yield self.store.get_room(room_id)
-        if not room:
-            raise RoomError(403, "Room does not exist.")
-
-        # does this user exist in this room
-        member = yield self.store.get_room_member(
-            room_id=room_id,
-            user_id="" if not user_id else user_id)
-
-        member_state = member.membership if member else None
-
-        if room.is_public and public_room_rules:
-            # make sure the user meets public room rules
-            if member_state not in public_room_rules:
-                raise RoomError(403, "Member does not meet public room rules.")
-        elif not room.is_public and private_room_rules:
-            # make sure the user meets private room rules
-            if member_state not in private_room_rules:
-                raise RoomError(
-                    403, "Member does not meet private room rules.")
-
-        data = yield self.store.get_current_state(
+        have_joined = yield self.auth.check_joined_room(room_id, user_id)
+        if not have_joined:
+            raise RoomError(403, "User not in room.")
+
+        data = yield self.state_handler.get_current_state(
             room_id, event_type, state_key
         )
         defer.returnValue(data)
@@ -219,9 +180,7 @@ class MessageHandler(BaseHandler):
 
     @defer.inlineCallbacks
     def send_feedback(self, event):
-        snapshot = yield self.store.snapshot_room(event.room_id, event.user_id)
-
-        yield self.auth.check(event, snapshot, raises=True)
+        snapshot = yield self.store.snapshot_room(event)
 
         # store message in db
         yield self._on_new_room_event(event, snapshot)
@@ -239,7 +198,7 @@ class MessageHandler(BaseHandler):
         yield self.auth.check_joined_room(room_id, user_id)
 
         # TODO: This is duplicating logic from snapshot_all_rooms
-        current_state = yield self.store.get_current_state(room_id)
+        current_state = yield self.state_handler.get_current_state(room_id)
         defer.returnValue([self.hs.serialize_event(c) for c in current_state])
 
     @defer.inlineCallbacks
@@ -289,8 +248,10 @@ class MessageHandler(BaseHandler):
             d = {
                 "room_id": event.room_id,
                 "membership": event.membership,
-                "visibility": ("public" if event.room_id in
-                              public_room_ids else "private"),
+                "visibility": (
+                    "public" if event.room_id in public_room_ids
+                    else "private"
+                ),
             }
 
             if event.membership == Membership.INVITE:
@@ -316,10 +277,12 @@ class MessageHandler(BaseHandler):
                     "end": end_token.to_string(),
                 }
 
-                current_state = yield self.store.get_current_state(
+                current_state = yield self.state_handler.get_current_state(
                     event.room_id
                 )
-                d["state"] = [self.hs.serialize_event(c) for c in current_state]
+                d["state"] = [
+                    self.hs.serialize_event(c) for c in current_state
+                ]
             except:
                 logger.exception("Failed to get snapshot")
 
@@ -330,5 +293,3 @@ class MessageHandler(BaseHandler):
         }
 
         defer.returnValue(ret)
-
-
diff --git a/synapse/handlers/profile.py b/synapse/handlers/profile.py
index dab9b03f04..834b37f5f3 100644
--- a/synapse/handlers/profile.py
+++ b/synapse/handlers/profile.py
@@ -17,7 +17,6 @@ from twisted.internet import defer
 
 from synapse.api.errors import SynapseError, AuthError, CodeMessageException
 from synapse.api.constants import Membership
-from synapse.api.events.room import RoomMemberEvent
 
 from ._base import BaseHandler
 
@@ -153,10 +152,13 @@ class ProfileHandler(BaseHandler):
         if not user.is_mine:
             defer.returnValue(None)
 
-        (displayname, avatar_url) = yield defer.gatherResults([
-            self.store.get_profile_displayname(user.localpart),
-            self.store.get_profile_avatar_url(user.localpart),
-        ])
+        (displayname, avatar_url) = yield defer.gatherResults(
+            [
+                self.store.get_profile_displayname(user.localpart),
+                self.store.get_profile_avatar_url(user.localpart),
+            ],
+            consumeErrors=True
+        )
 
         state["displayname"] = displayname
         state["avatar_url"] = avatar_url
@@ -196,10 +198,7 @@ class ProfileHandler(BaseHandler):
         )
 
         for j in joins:
-            snapshot = yield self.store.snapshot_room(
-                j.room_id, j.state_key, RoomMemberEvent.TYPE,
-                j.state_key
-            )
+            snapshot = yield self.store.snapshot_room(j)
 
             content = {
                 "membership": j.content["membership"],
@@ -218,5 +217,6 @@ class ProfileHandler(BaseHandler):
                 user_id=j.state_key,
             )
 
-            yield self.state_handler.handle_new_event(new_event, snapshot)
-            yield self._on_new_room_event(new_event, snapshot)
+            yield self._on_new_room_event(
+                new_event, snapshot, suppress_auth=True
+            )
diff --git a/synapse/handlers/register.py b/synapse/handlers/register.py
index 88eb51a8ed..7df9d9b82d 100644
--- a/synapse/handlers/register.py
+++ b/synapse/handlers/register.py
@@ -15,7 +15,6 @@
 
 """Contains functions for registering clients."""
 from twisted.internet import defer
-from twisted.python import log
 
 from synapse.types import UserID
 from synapse.api.errors import (
@@ -129,7 +128,7 @@ class RegistrationHandler(BaseHandler):
             try:
                 threepid = yield self._threepid_from_creds(c)
             except:
-                log.err()
+                logger.exception("Couldn't validate 3pid")
                 raise RegistrationError(400, "Couldn't validate 3pid")
 
             if not threepid:
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index 81ce1a5907..825957f721 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -21,10 +21,10 @@ from synapse.api.constants import Membership, JoinRules
 from synapse.api.errors import StoreError, SynapseError
 from synapse.api.events.room import (
     RoomMemberEvent, RoomCreateEvent, RoomPowerLevelsEvent,
-    RoomJoinRulesEvent, RoomAddStateLevelEvent, RoomTopicEvent,
-    RoomSendEventLevelEvent, RoomOpsPowerLevelsEvent, RoomNameEvent,
+    RoomTopicEvent, RoomNameEvent, RoomJoinRulesEvent,
 )
 from synapse.util import stringutils
+from synapse.util.async import run_on_reactor
 from ._base import BaseHandler
 
 import logging
@@ -122,15 +122,13 @@ class RoomCreationHandler(BaseHandler):
 
         @defer.inlineCallbacks
         def handle_event(event):
-            snapshot = yield self.store.snapshot_room(
-                room_id=room_id,
-                user_id=user_id,
-            )
+            snapshot = yield self.store.snapshot_room(event)
 
             logger.debug("Event: %s", event)
 
-            yield self.state_handler.handle_new_event(event, snapshot)
-            yield self._on_new_room_event(event, snapshot, extra_users=[user])
+            yield self._on_new_room_event(
+                event, snapshot, extra_users=[user], suppress_auth=True
+            )
 
         for event in creation_events:
             yield handle_event(event)
@@ -141,7 +139,6 @@ class RoomCreationHandler(BaseHandler):
                 etype=RoomNameEvent.TYPE,
                 room_id=room_id,
                 user_id=user_id,
-                required_power_level=50,
                 content={"name": name},
             )
 
@@ -153,7 +150,6 @@ class RoomCreationHandler(BaseHandler):
                 etype=RoomTopicEvent.TYPE,
                 room_id=room_id,
                 user_id=user_id,
-                required_power_level=50,
                 content={"topic": topic},
             )
 
@@ -198,7 +194,6 @@ class RoomCreationHandler(BaseHandler):
         event_keys = {
             "room_id": room_id,
             "user_id": creator.to_string(),
-            "required_power_level": 100,
         }
 
         def create(etype, **content):
@@ -215,7 +210,21 @@ class RoomCreationHandler(BaseHandler):
 
         power_levels_event = self.event_factory.create_event(
             etype=RoomPowerLevelsEvent.TYPE,
-            content={creator.to_string(): 100, "default": 0},
+            content={
+                "users": {
+                    creator.to_string(): 100,
+                },
+                "users_default": 0,
+                "events": {
+                    RoomNameEvent.TYPE: 100,
+                    RoomPowerLevelsEvent.TYPE: 100,
+                },
+                "events_default": 0,
+                "state_default": 50,
+                "ban": 50,
+                "kick": 50,
+                "redact": 50
+            },
             **event_keys
         )
 
@@ -225,30 +234,10 @@ class RoomCreationHandler(BaseHandler):
             join_rule=join_rule,
         )
 
-        add_state_event = create(
-            etype=RoomAddStateLevelEvent.TYPE,
-            level=100,
-        )
-
-        send_event = create(
-            etype=RoomSendEventLevelEvent.TYPE,
-            level=0,
-        )
-
-        ops = create(
-            etype=RoomOpsPowerLevelsEvent.TYPE,
-            ban_level=50,
-            kick_level=50,
-            redact_level=50,
-        )
-
         return [
             creation_event,
             power_levels_event,
             join_rules_event,
-            add_state_event,
-            send_event,
-            ops,
         ]
 
 
@@ -363,10 +352,8 @@ class RoomMemberHandler(BaseHandler):
         """
         target_user_id = event.state_key
 
-        snapshot = yield self.store.snapshot_room(
-            event.room_id, event.user_id,
-            RoomMemberEvent.TYPE, target_user_id
-        )
+        snapshot = yield self.store.snapshot_room(event)
+
         ## TODO(markjh): get prev state from snapshot.
         prev_state = yield self.store.get_room_member(
             target_user_id, event.room_id
@@ -375,13 +362,6 @@ class RoomMemberHandler(BaseHandler):
         if prev_state:
             event.content["prev"] = prev_state.membership
 
-#        if prev_state and prev_state.membership == event.membership:
-#            # treat this event as a NOOP.
-#            if do_auth:  # This is mainly to fix a unit test.
-#                yield self.auth.check(event, raises=True)
-#            defer.returnValue({})
-#            return
-
         room_id = event.room_id
 
         # If we're trying to join a room then we have to do this differently
@@ -391,29 +371,17 @@ class RoomMemberHandler(BaseHandler):
             yield self._do_join(event, snapshot, do_auth=do_auth)
         else:
             # This is not a JOIN, so we can handle it normally.
-            if do_auth:
-                yield self.auth.check(event, snapshot, raises=True)
-
-            # If we're banning someone, set a req power level
-            if event.membership == Membership.BAN:
-                if not hasattr(event, "required_power_level") or event.required_power_level is None:
-                    # Add some default required_power_level
-                    user_level = yield self.store.get_power_level(
-                        event.room_id,
-                        event.user_id,
-                    )
-                    event.required_power_level = user_level
 
             if prev_state and prev_state.membership == event.membership:
                 # double same action, treat this event as a NOOP.
                 defer.returnValue({})
                 return
 
-            yield self.state_handler.handle_new_event(event, snapshot)
             yield self._do_local_membership_update(
                 event,
                 membership=event.content["membership"],
                 snapshot=snapshot,
+                do_auth=do_auth,
             )
 
         defer.returnValue({"room_id": room_id})
@@ -443,10 +411,7 @@ class RoomMemberHandler(BaseHandler):
             content=content,
         )
 
-        snapshot = yield self.store.snapshot_room(
-            room_id, joinee.to_string(), RoomMemberEvent.TYPE,
-            joinee.to_string()
-        )
+        snapshot = yield self.store.snapshot_room(new_event)
 
         yield self._do_join(new_event, snapshot, room_host=host, do_auth=True)
 
@@ -468,9 +433,12 @@ class RoomMemberHandler(BaseHandler):
         # that we are allowed to join when we decide whether or not we
         # need to do the invite/join dance.
 
-        hosts = yield self.store.get_joined_hosts_for_room(room_id)
+        is_host_in_room = yield self.auth.check_host_in_room(
+            event.room_id,
+            self.hs.hostname
+        )
 
-        if self.hs.hostname in hosts:
+        if is_host_in_room:
             should_do_dance = False
         elif room_host:
             should_do_dance = True
@@ -502,14 +470,11 @@ class RoomMemberHandler(BaseHandler):
         if not have_joined:
             logger.debug("Doing normal join")
 
-            if do_auth:
-                yield self.auth.check(event, snapshot, raises=True)
-
-            yield self.state_handler.handle_new_event(event, snapshot)
             yield self._do_local_membership_update(
                 event,
                 membership=event.content["membership"],
                 snapshot=snapshot,
+                do_auth=do_auth,
             )
 
         user = self.hs.parse_userid(event.user_id)
@@ -553,26 +518,29 @@ class RoomMemberHandler(BaseHandler):
 
         defer.returnValue([r.room_id for r in rooms])
 
-    def _do_local_membership_update(self, event, membership, snapshot):
-        destinations = []
+    @defer.inlineCallbacks
+    def _do_local_membership_update(self, event, membership, snapshot,
+                                    do_auth):
+        yield run_on_reactor()
 
         # If we're inviting someone, then we should also send it to that
         # HS.
         target_user_id = event.state_key
         target_user = self.hs.parse_userid(target_user_id)
-        if membership == Membership.INVITE:
-            host = target_user.domain
-            destinations.append(host)
-
-        # Always include target domain
-        host = target_user.domain
-        destinations.append(host)
-
-        return self._on_new_room_event(
-            event, snapshot, extra_destinations=destinations,
-            extra_users=[target_user]
+        if membership == Membership.INVITE and not target_user.is_mine:
+            do_invite_host = target_user.domain
+        else:
+            do_invite_host = None
+
+        yield self._on_new_room_event(
+            event,
+            snapshot,
+            extra_users=[target_user],
+            suppress_auth=(not do_auth),
+            do_invite_host=do_invite_host,
         )
 
+
 class RoomListHandler(BaseHandler):
 
     @defer.inlineCallbacks