summary refs log tree commit diff
path: root/synapse
diff options
context:
space:
mode:
authorErik Johnston <erik@matrix.org>2014-08-22 15:50:23 +0100
committerErik Johnston <erik@matrix.org>2014-08-22 15:50:23 +0100
commite1297c922d7f88eed088b365bd6cf15951443506 (patch)
tree57fb71ff724a4046d3a8e3ea4ce3de4735b9dd69 /synapse
parentAdded initial swagger REST API spec. (diff)
parentMerge branch 'develop' of github.com:matrix-org/synapse into release-v0.0.1 (diff)
downloadsynapse-e1297c922d7f88eed088b365bd6cf15951443506.tar.xz
Merge branch 'release-v0.0.1' of github.com:matrix-org/synapse
Diffstat (limited to 'synapse')
-rw-r--r--synapse/__init__.py2
-rw-r--r--synapse/api/events/__init__.py1
-rw-r--r--synapse/api/events/factory.py17
-rw-r--r--synapse/api/events/room.py30
-rw-r--r--synapse/api/notifier.py3
-rw-r--r--synapse/api/streams/__init__.py20
-rw-r--r--synapse/api/streams/event.py173
-rwxr-xr-xsynapse/app/homeserver.py9
-rw-r--r--synapse/federation/handler.py26
-rw-r--r--synapse/federation/replication.py9
-rw-r--r--synapse/handlers/_base.py1
-rw-r--r--synapse/handlers/events.py8
-rw-r--r--synapse/handlers/federation.py124
-rw-r--r--synapse/handlers/presence.py5
-rw-r--r--synapse/handlers/room.py271
-rw-r--r--synapse/http/server.py26
-rw-r--r--synapse/rest/room.py63
-rw-r--r--synapse/server.py2
-rw-r--r--synapse/state.py2
-rw-r--r--synapse/storage/__init__.py187
-rw-r--r--synapse/storage/_base.py33
-rw-r--r--synapse/storage/feedback.py72
-rw-r--r--synapse/storage/message.py81
-rw-r--r--synapse/storage/pdu.py22
-rw-r--r--synapse/storage/room.py97
-rw-r--r--synapse/storage/roomdata.py85
-rw-r--r--synapse/storage/roommember.py172
-rw-r--r--synapse/storage/schema/im.sql82
-rw-r--r--synapse/storage/stream.py468
-rw-r--r--synapse/types.py8
30 files changed, 1093 insertions, 1006 deletions
diff --git a/synapse/__init__.py b/synapse/__init__.py
index 1e7b2ab272..47fc1b2ea4 100644
--- a/synapse/__init__.py
+++ b/synapse/__init__.py
@@ -15,3 +15,5 @@
 
 """ This is a reference implementation of a synapse home server.
 """
+
+__version__ = "0.0.1"
diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py
index 921fd08832..aa04dbece7 100644
--- a/synapse/api/events/__init__.py
+++ b/synapse/api/events/__init__.py
@@ -51,6 +51,7 @@ class SynapseEvent(JsonEncodedObject):
         "depth",
         "destinations",
         "origin",
+        "outlier",
     ]
 
     required_keys = [
diff --git a/synapse/api/events/factory.py b/synapse/api/events/factory.py
index 12aa04fc6e..c2cdcddf41 100644
--- a/synapse/api/events/factory.py
+++ b/synapse/api/events/factory.py
@@ -15,7 +15,7 @@
 
 from synapse.api.events.room import (
     RoomTopicEvent, MessageEvent, RoomMemberEvent, FeedbackEvent,
-    InviteJoinEvent, RoomConfigEvent
+    InviteJoinEvent, RoomConfigEvent, RoomNameEvent, GenericEvent,
 )
 
 from synapse.util.stringutils import random_string
@@ -25,6 +25,7 @@ class EventFactory(object):
 
     _event_classes = [
         RoomTopicEvent,
+        RoomNameEvent,
         MessageEvent,
         RoomMemberEvent,
         FeedbackEvent,
@@ -32,20 +33,24 @@ class EventFactory(object):
         RoomConfigEvent
     ]
 
-    def __init__(self):
+    def __init__(self, hs):
         self._event_list = {}  # dict of TYPE to event class
         for event_class in EventFactory._event_classes:
             self._event_list[event_class.TYPE] = event_class
 
+        self.clock = hs.get_clock()
+
     def create_event(self, etype=None, **kwargs):
         kwargs["type"] = etype
         if "event_id" not in kwargs:
             kwargs["event_id"] = random_string(10)
 
-        try:
+        if "ts" not in kwargs:
+            kwargs["ts"] = int(self.clock.time_msec())
+
+        if etype in self._event_list:
             handler = self._event_list[etype]
-        except KeyError:  # unknown event type
-            # TODO allow custom event types.
-            raise NotImplementedError("Unknown etype=%s" % etype)
+        else:
+            handler = GenericEvent
 
         return handler(**kwargs)
diff --git a/synapse/api/events/room.py b/synapse/api/events/room.py
index f3df849af2..42459f3f21 100644
--- a/synapse/api/events/room.py
+++ b/synapse/api/events/room.py
@@ -16,17 +16,45 @@
 from . import SynapseEvent
 
 
+class GenericEvent(SynapseEvent):
+    def get_content_template(self):
+        return {}
+
+
 class RoomTopicEvent(SynapseEvent):
     TYPE = "m.room.topic"
 
+    internal_keys = SynapseEvent.internal_keys + [
+        "topic",
+    ]
+
     def __init__(self, **kwargs):
         kwargs["state_key"] = ""
+        if "topic" in kwargs["content"]:
+            kwargs["topic"] = kwargs["content"]["topic"]
         super(RoomTopicEvent, self).__init__(**kwargs)
 
     def get_content_template(self):
         return {"topic": u"string"}
 
 
+class RoomNameEvent(SynapseEvent):
+    TYPE = "m.room.name"
+
+    internal_keys = SynapseEvent.internal_keys + [
+        "name",
+    ]
+
+    def __init__(self, **kwargs):
+        kwargs["state_key"] = ""
+        if "name" in kwargs["content"]:
+            kwargs["name"] = kwargs["content"]["name"]
+        super(RoomNameEvent, self).__init__(**kwargs)
+
+    def get_content_template(self):
+        return {"name": u"string"}
+
+
 class RoomMemberEvent(SynapseEvent):
     TYPE = "m.room.member"
 
@@ -38,6 +66,8 @@ class RoomMemberEvent(SynapseEvent):
     def __init__(self, **kwargs):
         if "target_user_id" in kwargs:
             kwargs["state_key"] = kwargs["target_user_id"]
+        if "membership" not in kwargs:
+            kwargs["membership"] = kwargs.get("content", {}).get("membership")
         super(RoomMemberEvent, self).__init__(**kwargs)
 
     def get_content_template(self):
diff --git a/synapse/api/notifier.py b/synapse/api/notifier.py
index 65b5a4ebb3..9f622df6bb 100644
--- a/synapse/api/notifier.py
+++ b/synapse/api/notifier.py
@@ -15,6 +15,7 @@
 
 from synapse.api.constants import Membership
 from synapse.api.events.room import RoomMemberEvent
+from synapse.api.streams.event import EventsStreamData
 
 from twisted.internet import defer
 from twisted.internet import reactor
@@ -66,7 +67,7 @@ class Notifier(object):
                 self._notify_and_callback(
                     user_id=user_id,
                     event_data=event.get_dict(),
-                    stream_type=event.type,
+                    stream_type=EventsStreamData.EVENT_TYPE,
                     store_id=store_id)
 
     def on_new_user_event(self, user_id, event_data, stream_type, store_id):
diff --git a/synapse/api/streams/__init__.py b/synapse/api/streams/__init__.py
index 989e63f9ec..d831eafbab 100644
--- a/synapse/api/streams/__init__.py
+++ b/synapse/api/streams/__init__.py
@@ -20,23 +20,24 @@ class PaginationConfig(object):
 
     """A configuration object which stores pagination parameters."""
 
-    def __init__(self, from_tok=None, to_tok=None, limit=0):
+    def __init__(self, from_tok=None, to_tok=None, direction='f', limit=0):
         self.from_tok = from_tok
         self.to_tok = to_tok
+        self.direction = direction
         self.limit = limit
 
     @classmethod
     def from_request(cls, request, raise_invalid_params=True):
         params = {
-            "from_tok": PaginationStream.TOK_START,
-            "to_tok": PaginationStream.TOK_END,
-            "limit": 0
+            "from_tok": "END",
+            "direction": 'f',
         }
 
         query_param_mappings = [  # 3-tuple of qp_key, attribute, rules
             ("from", "from_tok", lambda x: type(x) == str),
             ("to", "to_tok", lambda x: type(x) == str),
-            ("limit", "limit", lambda x: x.isdigit())
+            ("limit", "limit", lambda x: x.isdigit()),
+            ("dir", "direction", lambda x: x == 'f' or x == 'b'),
         ]
 
         for qp, attr, is_valid in query_param_mappings:
@@ -48,12 +49,17 @@ class PaginationConfig(object):
 
         return PaginationConfig(**params)
 
+    def __str__(self):
+        return (
+            "<PaginationConfig from_tok=%s, to_tok=%s, "
+            "direction=%s, limit=%s>"
+        ) % (self.from_tok, self.to_tok, self.direction, self.limit)
+
 
 class PaginationStream(object):
 
     """ An interface for streaming data as chunks. """
 
-    TOK_START = "START"
     TOK_END = "END"
 
     def get_chunk(self, config=None):
@@ -76,7 +82,7 @@ class StreamData(object):
         self.hs = hs
         self.store = hs.get_datastore()
 
-    def get_rows(self, user_id, from_pkey, to_pkey, limit):
+    def get_rows(self, user_id, from_pkey, to_pkey, limit, direction):
         """ Get event stream data between the specified pkeys.
 
         Args:
diff --git a/synapse/api/streams/event.py b/synapse/api/streams/event.py
index 4b6d739e54..a5c8b2b31f 100644
--- a/synapse/api/streams/event.py
+++ b/synapse/api/streams/event.py
@@ -18,6 +18,7 @@
 from twisted.internet import defer
 
 from synapse.api.errors import EventStreamError
+from synapse.api.events import SynapseEvent
 from synapse.api.events.room import (
     RoomMemberEvent, MessageEvent, FeedbackEvent, RoomTopicEvent
 )
@@ -28,17 +29,17 @@ import logging
 logger = logging.getLogger(__name__)
 
 
-class MessagesStreamData(StreamData):
-    EVENT_TYPE = MessageEvent.TYPE
+class EventsStreamData(StreamData):
+    EVENT_TYPE = "EventsStream"
 
     def __init__(self, hs, room_id=None, feedback=False):
-        super(MessagesStreamData, self).__init__(hs)
+        super(EventsStreamData, self).__init__(hs)
         self.room_id = room_id
         self.with_feedback = feedback
 
     @defer.inlineCallbacks
-    def get_rows(self, user_id, from_key, to_key, limit):
-        (data, latest_ver) = yield self.store.get_message_stream(
+    def get_rows(self, user_id, from_key, to_key, limit, direction):
+        data, latest_ver = yield self.store.get_room_events(
             user_id=user_id,
             from_key=from_key,
             to_key=to_key,
@@ -50,74 +51,7 @@ class MessagesStreamData(StreamData):
 
     @defer.inlineCallbacks
     def max_token(self):
-        val = yield self.store.get_max_message_id()
-        defer.returnValue(val)
-
-
-class RoomMemberStreamData(StreamData):
-    EVENT_TYPE = RoomMemberEvent.TYPE
-
-    @defer.inlineCallbacks
-    def get_rows(self, user_id, from_key, to_key, limit):
-        (data, latest_ver) = yield self.store.get_room_member_stream(
-            user_id=user_id,
-            from_key=from_key,
-            to_key=to_key
-        )
-
-        defer.returnValue((data, latest_ver))
-
-    @defer.inlineCallbacks
-    def max_token(self):
-        val = yield self.store.get_max_room_member_id()
-        defer.returnValue(val)
-
-
-class FeedbackStreamData(StreamData):
-    EVENT_TYPE = FeedbackEvent.TYPE
-
-    def __init__(self, hs, room_id=None):
-        super(FeedbackStreamData, self).__init__(hs)
-        self.room_id = room_id
-
-    @defer.inlineCallbacks
-    def get_rows(self, user_id, from_key, to_key, limit):
-        (data, latest_ver) = yield self.store.get_feedback_stream(
-            user_id=user_id,
-            from_key=from_key,
-            to_key=to_key,
-            limit=limit,
-            room_id=self.room_id
-        )
-        defer.returnValue((data, latest_ver))
-
-    @defer.inlineCallbacks
-    def max_token(self):
-        val = yield self.store.get_max_feedback_id()
-        defer.returnValue(val)
-
-
-class RoomDataStreamData(StreamData):
-    EVENT_TYPE = RoomTopicEvent.TYPE  # TODO need multiple event types
-
-    def __init__(self, hs, room_id=None):
-        super(RoomDataStreamData, self).__init__(hs)
-        self.room_id = room_id
-
-    @defer.inlineCallbacks
-    def get_rows(self, user_id, from_key, to_key, limit):
-        (data, latest_ver) = yield self.store.get_room_data_stream(
-            user_id=user_id,
-            from_key=from_key,
-            to_key=to_key,
-            limit=limit,
-            room_id=self.room_id
-        )
-        defer.returnValue((data, latest_ver))
-
-    @defer.inlineCallbacks
-    def max_token(self):
-        val = yield self.store.get_max_room_data_id()
+        val = yield self.store.get_room_events_max_id()
         defer.returnValue(val)
 
 
@@ -136,6 +70,15 @@ class EventStream(PaginationStream):
             pagination_config.from_tok)
         pagination_config.to_tok = yield self.fix_token(
             pagination_config.to_tok)
+
+        if (
+            not pagination_config.to_tok
+            and pagination_config.direction == 'f'
+        ):
+            pagination_config.to_tok = yield self.get_current_max_token()
+
+        logger.debug("pagination_config: %s", pagination_config)
+
         defer.returnValue(pagination_config)
 
     @defer.inlineCallbacks
@@ -147,39 +90,42 @@ class EventStream(PaginationStream):
         Returns:
             The fixed-up token, which may == token.
         """
-        # replace TOK_START and TOK_END with 0_0_0 or -1_-1_-1 depending.
-        replacements = [
-            (PaginationStream.TOK_START, "0"),
-            (PaginationStream.TOK_END, "-1")
-        ]
-        for magic_token, key in replacements:
-            if magic_token == token:
-                token = EventStream.SEPARATOR.join(
-                    [key] * len(self.stream_data)
-                )
-
-        # replace -1 values with an actual pkey
-        token_segments = self._split_token(token)
-        for i, tok in enumerate(token_segments):
-            if tok == -1:
-                # add 1 to the max token because results are EXCLUSIVE from the
-                # latest version.
-                token_segments[i] = 1 + (yield self.stream_data[i].max_token())
-        defer.returnValue(EventStream.SEPARATOR.join(
-            str(x) for x in token_segments
-        ))
+        if token == PaginationStream.TOK_END:
+            new_token = yield self.get_current_max_token()
+
+            logger.debug("fix_token: From %s to %s", token, new_token)
+
+            token = new_token
+
+        defer.returnValue(token)
 
     @defer.inlineCallbacks
-    def get_chunk(self, config=None):
+    def get_current_max_token(self):
+        new_token_parts = []
+        for s in self.stream_data:
+            mx = yield s.max_token()
+            new_token_parts.append(str(mx))
+
+        new_token = EventStream.SEPARATOR.join(new_token_parts)
+
+        logger.debug("get_current_max_token: %s", new_token)
+
+        defer.returnValue(new_token)
+
+    @defer.inlineCallbacks
+    def get_chunk(self, config):
         # no support for limit on >1 streams, makes no sense.
         if config.limit and len(self.stream_data) > 1:
             raise EventStreamError(
                 400, "Limit not supported on multiplexed streams."
             )
 
-        (chunk_data, next_tok) = yield self._get_chunk_data(config.from_tok,
-                                                            config.to_tok,
-                                                            config.limit)
+        chunk_data, next_tok = yield self._get_chunk_data(
+            config.from_tok,
+            config.to_tok,
+            config.limit,
+            config.direction,
+        )
 
         defer.returnValue({
             "chunk": chunk_data,
@@ -188,7 +134,7 @@ class EventStream(PaginationStream):
         })
 
     @defer.inlineCallbacks
-    def _get_chunk_data(self, from_tok, to_tok, limit):
+    def _get_chunk_data(self, from_tok, to_tok, limit, direction):
         """ Get event data between the two tokens.
 
         Tokens are SEPARATOR separated values representing pkey values of
@@ -206,11 +152,12 @@ class EventStream(PaginationStream):
             EventStreamError if something went wrong.
         """
         # sanity check
-        if (from_tok.count(EventStream.SEPARATOR) !=
-                to_tok.count(EventStream.SEPARATOR) or
-                (from_tok.count(EventStream.SEPARATOR) + 1) !=
-                len(self.stream_data)):
-            raise EventStreamError(400, "Token lengths don't match.")
+        if to_tok is not None:
+            if (from_tok.count(EventStream.SEPARATOR) !=
+                    to_tok.count(EventStream.SEPARATOR) or
+                    (from_tok.count(EventStream.SEPARATOR) + 1) !=
+                    len(self.stream_data)):
+                raise EventStreamError(400, "Token lengths don't match.")
 
         chunk = []
         next_ver = []
@@ -224,10 +171,13 @@ class EventStream(PaginationStream):
                 continue
 
             (event_chunk, max_pkey) = yield self.stream_data[i].get_rows(
-                self.user_id, from_pkey, to_pkey, limit
+                self.user_id, from_pkey, to_pkey, limit, direction,
             )
 
-            chunk += event_chunk
+            chunk.extend([
+                e.get_dict() if isinstance(e, SynapseEvent) else e
+                for e in event_chunk
+            ])
             next_ver.append(str(max_pkey))
 
         defer.returnValue((chunk, EventStream.SEPARATOR.join(next_ver)))
@@ -240,9 +190,8 @@ class EventStream(PaginationStream):
         Returns:
             A list of ints.
         """
-        segments = token.split(EventStream.SEPARATOR)
-        try:
-            int_segments = [int(x) for x in segments]
-        except ValueError:
-            raise EventStreamError(400, "Bad token: %s" % token)
-        return int_segments
+        if token:
+            segments = token.split(EventStream.SEPARATOR)
+        else:
+            segments = [None] * len(self.stream_data)
+        return segments
diff --git a/synapse/app/homeserver.py b/synapse/app/homeserver.py
index ca102236cf..495149466c 100755
--- a/synapse/app/homeserver.py
+++ b/synapse/app/homeserver.py
@@ -56,7 +56,7 @@ class SynapseHomeServer(HomeServer):
         return File("webclient")  # TODO configurable?
 
     def build_resource_for_content_repo(self):
-        return ContentRepoResource("uploads", self.auth)
+        return ContentRepoResource(self, self.upload_dir, self.auth)
 
     def build_db_pool(self):
         """ Set up all the dbs. Since all the *.sql have IF NOT EXISTS, so we
@@ -235,8 +235,8 @@ def setup():
     parser.add_argument('--pid-file', dest="pid", help="When running as a "
                         "daemon, the file to store the pid in",
                         default="hs.pid")
-    parser.add_argument("-w", "--webclient", dest="webclient",
-                        action="store_true", help="Host the web client.")
+    parser.add_argument("-W", "--webclient", dest="webclient", default=True,
+                        action="store_false", help="Don't host a web client.")
     args = parser.parse_args()
 
     verbosity = int(args.verbose) if args.verbose else None
@@ -257,7 +257,8 @@ def setup():
 
     hs = SynapseHomeServer(
         args.host,
-        db_name=db_name
+        upload_dir=os.path.abspath("uploads"),
+        db_name=db_name,
     )
 
     # This object doesn't need to be saved because it's set as the handler for
diff --git a/synapse/federation/handler.py b/synapse/federation/handler.py
index 580e591aca..984c1558e9 100644
--- a/synapse/federation/handler.py
+++ b/synapse/federation/handler.py
@@ -63,7 +63,7 @@ class FederationEventHandler(object):
             Deferred: Resolved when it has successfully been queued for
             processing.
         """
-        yield self._fill_out_prev_events(event)
+        yield self.fill_out_prev_events(event)
 
         pdu = self.pdu_codec.pdu_from_event(event)
 
@@ -74,10 +74,18 @@ class FederationEventHandler(object):
 
     @log_function
     @defer.inlineCallbacks
-    def backfill(self, room_id, limit):
-        # TODO: Work out which destinations to ask for backfill
-        # self.replication_layer.backfill(dest, room_id, limit)
-        pass
+    def backfill(self, dest, room_id, limit):
+        pdus = yield self.replication_layer.backfill(dest, room_id, limit)
+
+        if not pdus:
+            defer.returnValue([])
+
+        events = [
+            self.pdu_codec.event_from_pdu(pdu)
+            for pdu in pdus
+        ]
+
+        defer.returnValue(events)
 
     @log_function
     def get_state_for_room(self, destination, room_id):
@@ -87,7 +95,7 @@ class FederationEventHandler(object):
 
     @log_function
     @defer.inlineCallbacks
-    def on_receive_pdu(self, pdu):
+    def on_receive_pdu(self, pdu, backfilled):
         """ Called by the ReplicationLayer when we have a new pdu. We need to
         do auth checks and put it throught the StateHandler.
         """
@@ -95,7 +103,7 @@ class FederationEventHandler(object):
 
         try:
             with (yield self.lock_manager.lock(pdu.context)):
-                if event.is_state:
+                if event.is_state and not backfilled:
                     is_new_state = yield self.state_handler.handle_new_state(
                         pdu
                     )
@@ -104,7 +112,7 @@ class FederationEventHandler(object):
                 else:
                     is_new_state = False
 
-            yield self.event_handler.on_receive(event, is_new_state)
+            yield self.event_handler.on_receive(event, is_new_state, backfilled)
 
         except AuthError:
             # TODO: Implement something in federation that allows us to
@@ -129,7 +137,7 @@ class FederationEventHandler(object):
         yield self.event_handler.on_receive(new_state_event)
 
     @defer.inlineCallbacks
-    def _fill_out_prev_events(self, event):
+    def fill_out_prev_events(self, event):
         if hasattr(event, "prev_events"):
             return
 
diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py
index c9f2e06b7b..8030d0963f 100644
--- a/synapse/federation/replication.py
+++ b/synapse/federation/replication.py
@@ -209,7 +209,7 @@ class ReplicationLayer(object):
 
         pdus = [Pdu(outlier=False, **p) for p in transaction.pdus]
         for pdu in pdus:
-            yield self._handle_new_pdu(pdu)
+            yield self._handle_new_pdu(pdu, backfilled=True)
 
         defer.returnValue(pdus)
 
@@ -416,7 +416,7 @@ class ReplicationLayer(object):
 
     @defer.inlineCallbacks
     @log_function
-    def _handle_new_pdu(self, pdu):
+    def _handle_new_pdu(self, pdu, backfilled=False):
         # We reprocess pdus when we have seen them only as outliers
         existing = yield self._get_persisted_pdu(pdu.pdu_id, pdu.origin)
 
@@ -452,7 +452,10 @@ class ReplicationLayer(object):
         # Persist the Pdu, but don't mark it as processed yet.
         yield self.pdu_actions.persist_received(pdu)
 
-        ret = yield self.handler.on_receive_pdu(pdu)
+        if not backfilled:
+            ret = yield self.handler.on_receive_pdu(pdu, backfilled=backfilled)
+        else:
+            ret = None
 
         yield self.pdu_actions.mark_as_processed(pdu)
 
diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py
index c2f4685c92..3f07b5aa4a 100644
--- a/synapse/handlers/_base.py
+++ b/synapse/handlers/_base.py
@@ -24,4 +24,5 @@ class BaseHandler(object):
         self.notifier = hs.get_notifier()
         self.room_lock = hs.get_room_lock_manager()
         self.state_handler = hs.get_state_handler()
+        self.distributor = hs.get_distributor()
         self.hs = hs
diff --git a/synapse/handlers/events.py b/synapse/handlers/events.py
index 3af7d824a2..6bb797caf2 100644
--- a/synapse/handlers/events.py
+++ b/synapse/handlers/events.py
@@ -17,8 +17,7 @@ from twisted.internet import defer
 
 from ._base import BaseHandler
 from synapse.api.streams.event import (
-    EventStream, MessagesStreamData, RoomMemberStreamData, FeedbackStreamData,
-    RoomDataStreamData
+    EventStream, EventsStreamData
 )
 from synapse.handlers.presence import PresenceStreamData
 
@@ -26,10 +25,7 @@ from synapse.handlers.presence import PresenceStreamData
 class EventStreamHandler(BaseHandler):
 
     stream_data_classes = [
-        MessagesStreamData,
-        RoomMemberStreamData,
-        FeedbackStreamData,
-        RoomDataStreamData,
+        EventsStreamData,
         PresenceStreamData,
     ]
 
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index 7026df90a2..9cff444779 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -32,10 +32,19 @@ logger = logging.getLogger(__name__)
 class FederationHandler(BaseHandler):
 
     """Handles events that originated from federation."""
+    def __init__(self, hs):
+        super(FederationHandler, self).__init__(hs)
+
+        self.distributor.observe(
+            "user_joined_room",
+            self._on_user_joined
+        )
+
+        self.waiting_for_join_list = {}
 
     @log_function
     @defer.inlineCallbacks
-    def on_receive(self, event, is_new_state):
+    def on_receive(self, event, is_new_state, backfilled):
         if hasattr(event, "state_key") and not is_new_state:
             logger.debug("Ignoring old state.")
             return
@@ -70,6 +79,115 @@ class FederationHandler(BaseHandler):
 
         else:
             with (yield self.room_lock.lock(event.room_id)):
-                store_id = yield self.store.persist_event(event)
+                store_id = yield self.store.persist_event(event, backfilled)
+
+            room = yield self.store.get_room(event.room_id)
+
+            if not room:
+                # Huh, let's try and get the current state
+                try:
+                    federation = self.hs.get_federation()
+                    yield federation.get_state_for_room(
+                        event.origin, event.room_id
+                    )
+
+                    hosts = yield self.store.get_joined_hosts_for_room(
+                        event.room_id
+                    )
+                    if self.hs.hostname in hosts:
+                        try:
+                            yield self.store.store_room(
+                                event.room_id,
+                                "",
+                                is_public=False
+                            )
+                        except:
+                            pass
+                except:
+                    logger.exception(
+                        "Failed to get current state for room %s",
+                        event.room_id
+                    )
+
+            if not backfilled:
+                yield self.notifier.on_new_room_event(event, store_id)
+
+        if event.type == RoomMemberEvent.TYPE:
+            if event.membership == Membership.JOIN:
+                user = self.hs.parse_userid(event.target_user_id)
+                self.distributor.fire(
+                    "user_joined_room", user=user, room_id=event.room_id
+                )
+
+
+    @log_function
+    @defer.inlineCallbacks
+    def backfill(self, dest, room_id, limit):
+        events = yield self.hs.get_federation().backfill(dest, room_id, limit)
+
+        for event in events:
+            try:
+                yield self.store.persist_event(event, backfilled=True)
+            except:
+                logger.exception("Failed to persist event: %s", event)
+
+        defer.returnValue(events)
 
-            yield self.notifier.on_new_room_event(event, store_id)
+    @log_function
+    @defer.inlineCallbacks
+    def do_invite_join(self, target_host, room_id, joinee, content):
+        federation = self.hs.get_federation()
+
+        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)
+
+        # First get current state to see if we are already joined.
+        try:
+            yield federation.get_state_for_room(target_host, room_id)
+
+            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
+        )
+
+        new_event.destinations = [target_host]
+
+        yield federation.handle_new_event(new_event)
+
+        # TODO (erikj): Time out here.
+        d = defer.Deferred()
+        self.waiting_for_join_list.setdefault((joinee, room_id), []).append(d)
+        yield d
+
+        try:
+            yield self.store.store_room(
+                event.room_id,
+                "",
+                is_public=False
+            )
+        except:
+            pass
+
+
+        defer.returnValue(True)
+
+
+    @log_function
+    def _on_user_joined(self, user, 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/presence.py b/synapse/handlers/presence.py
index a2152c99cf..540e114b82 100644
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -685,7 +685,10 @@ class PresenceStreamData(StreamData):
         super(PresenceStreamData, self).__init__(hs)
         self.presence = hs.get_handlers().presence_handler
 
-    def get_rows(self, user_id, from_key, to_key, limit):
+    def get_rows(self, user_id, from_key, to_key, limit, direction):
+        from_key = int(from_key)
+        to_key = int(to_key)
+
         cachemap = self.presence._user_cachemap
 
         # TODO(paul): limit, and filter by visibility
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index 5d0379254b..899b653fb7 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -23,7 +23,8 @@ from synapse.api.events.room import (
     RoomTopicEvent, MessageEvent, InviteJoinEvent, RoomMemberEvent,
     RoomConfigEvent
 )
-from synapse.api.streams.event import EventStream, MessagesStreamData
+from synapse.api.streams.event import EventStream, EventsStreamData
+from synapse.handlers.presence import PresenceStreamData
 from synapse.util import stringutils
 from ._base import BaseHandler
 
@@ -59,12 +60,14 @@ class MessageHandler(BaseHandler):
         yield self.auth.check_joined_room(room_id, user_id)
 
         # Pull out the message from the db
-        msg = yield self.store.get_message(room_id=room_id,
-                                           msg_id=msg_id,
-                                           user_id=sender_id)
+#        msg = yield self.store.get_message(
+#            room_id=room_id,
+#            msg_id=msg_id,
+#            user_id=sender_id
+#        )
+
+        # TODO (erikj): Once we work out the correct c-s api we need to think on how to do this.
 
-        if msg:
-            defer.returnValue(msg)
         defer.returnValue(None)
 
     @defer.inlineCallbacks
@@ -114,8 +117,9 @@ class MessageHandler(BaseHandler):
         """
         yield self.auth.check_joined_room(room_id, user_id)
 
-        data_source = [MessagesStreamData(self.hs, room_id=room_id,
-                                          feedback=feedback)]
+        data_source = [
+            EventsStreamData(self.hs, room_id=room_id, feedback=feedback)
+        ]
         event_stream = EventStream(user_id, data_source)
         pagin_config = yield event_stream.fix_tokens(pagin_config)
         data_chunk = yield event_stream.get_chunk(config=pagin_config)
@@ -141,12 +145,7 @@ class MessageHandler(BaseHandler):
             yield self.state_handler.handle_new_event(event)
 
             # store in db
-            store_id = yield self.store.store_room_data(
-                room_id=event.room_id,
-                etype=event.type,
-                state_key=event.state_key,
-                content=json.dumps(event.content)
-            )
+            store_id = yield self.store.persist_event(event)
 
             event.destinations = yield self.store.get_joined_hosts_for_room(
                 event.room_id
@@ -201,19 +200,17 @@ class MessageHandler(BaseHandler):
                 raise RoomError(
                     403, "Member does not meet private room rules.")
 
-        data = yield self.store.get_room_data(room_id, event_type, state_key)
+        data = yield self.store.get_current_state(
+            room_id, event_type, state_key
+        )
         defer.returnValue(data)
 
     @defer.inlineCallbacks
-    def get_feedback(self, room_id=None, msg_sender_id=None, msg_id=None,
-                     user_id=None, fb_sender_id=None, fb_type=None):
-        yield self.auth.check_joined_room(room_id, user_id)
+    def get_feedback(self, event_id):
+        # yield self.auth.check_joined_room(room_id, user_id)
 
         # Pull out the feedback from the db
-        fb = yield self.store.get_feedback(
-            room_id=room_id, msg_id=msg_id, msg_sender_id=msg_sender_id,
-            fb_sender_id=fb_sender_id, fb_type=fb_type
-        )
+        fb = yield self.store.get_feedback(event_id)
 
         if fb:
             defer.returnValue(fb)
@@ -260,20 +257,59 @@ class MessageHandler(BaseHandler):
             user_id=user_id,
             membership_list=[Membership.INVITE, Membership.JOIN]
         )
-        for room_info in room_list:
-            if room_info["membership"] != Membership.JOIN:
+
+        rooms_ret = []
+
+        now_rooms_token = yield self.store.get_room_events_max_id()
+
+        # FIXME (erikj): Fix this.
+        presence_stream = PresenceStreamData(self.hs)
+        now_presence_token = yield presence_stream.max_token()
+        presence = yield presence_stream.get_rows(
+            user_id, 0, now_presence_token, None, None
+        )
+
+        # FIXME (erikj): We need to not generate this token,
+        now_token = "%s_%s" % (now_rooms_token, now_presence_token)
+
+        for event in room_list:
+            d = {
+                "room_id": event.room_id,
+                "membership": event.membership,
+            }
+
+            if event.membership == Membership.INVITE:
+                d["inviter"] = event.user_id
+
+            rooms_ret.append(d)
+
+            if event.membership != Membership.JOIN:
                 continue
             try:
-                event_chunk = yield self.get_messages(
-                    user_id=user_id,
-                    pagin_config=pagin_config,
-                    feedback=feedback,
-                    room_id=room_info["room_id"]
+                messages, token = yield self.store.get_recent_events_for_room(
+                    event.room_id,
+                    limit=10,
+                    end_token=now_rooms_token,
                 )
-                room_info["messages"] = event_chunk
+
+                d["messages"] = {
+                    "chunk": [m.get_dict() for m in messages],
+                    "start": token[0],
+                    "end": token[1],
+                }
+
+                current_state = yield self.store.get_current_state(event.room_id)
+                d["state"] = [c.get_dict() for c in current_state]
             except:
-                pass
-        defer.returnValue(room_list)
+                logger.exception("Failed to get snapshot")
+
+        user = self.hs.parse_userid(user_id)
+
+        ret = {"rooms": rooms_ret, "presence": presence[0], "end": now_token}
+
+        logger.debug("snapshot_all_rooms returning: %s", ret)
+
+        defer.returnValue(ret)
 
 
 class RoomCreationHandler(BaseHandler):
@@ -372,7 +408,6 @@ class RoomCreationHandler(BaseHandler):
 
         yield self.hs.get_handlers().room_member_handler.change_membership(
             join_event,
-            broadcast_msg=True,
             do_auth=False
         )
 
@@ -451,11 +486,11 @@ class RoomMemberHandler(BaseHandler):
 
         member_list = yield self.store.get_room_members(room_id=room_id)
         event_list = [
-            entry.as_event(self.event_factory).get_dict()
+            entry.get_dict()
             for entry in member_list
         ]
         chunk_data = {
-            "start": "START",
+            "start": "START",  # FIXME (erikj): START is no longer a valid value
             "end": "END",
             "chunk": event_list
         }
@@ -484,29 +519,28 @@ class RoomMemberHandler(BaseHandler):
         defer.returnValue(member)
 
     @defer.inlineCallbacks
-    def change_membership(self, event=None, broadcast_msg=False, do_auth=True):
+    def change_membership(self, event=None, do_auth=True):
         """ Change the membership status of a user in a room.
 
         Args:
             event (SynapseEvent): The membership event
-            broadcast_msg (bool): True to inject a membership message into this
-                room on success.
         Raises:
             SynapseError if there was a problem changing the membership.
         """
 
-        #broadcast_msg = False
-
         prev_state = yield self.store.get_room_member(
             event.target_user_id, event.room_id
         )
 
-        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
+        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
 
@@ -514,9 +548,7 @@ class RoomMemberHandler(BaseHandler):
         # if this HS is not currently in the room, i.e. we have to do the
         # invite/join dance.
         if event.membership == Membership.JOIN:
-            yield self._do_join(
-                event, do_auth=do_auth, broadcast_msg=broadcast_msg
-            )
+            yield self._do_join(event, do_auth=do_auth)
         else:
             # This is not a JOIN, so we can handle it normally.
             if do_auth:
@@ -534,7 +566,6 @@ class RoomMemberHandler(BaseHandler):
             yield self._do_local_membership_update(
                 event,
                 membership=event.content["membership"],
-                broadcast_msg=broadcast_msg,
             )
 
         defer.returnValue({"room_id": room_id})
@@ -569,14 +600,14 @@ class RoomMemberHandler(BaseHandler):
         defer.returnValue({"room_id": room_id})
 
     @defer.inlineCallbacks
-    def _do_join(self, event, room_host=None, do_auth=True, broadcast_msg=True):
+    def _do_join(self, event, room_host=None, do_auth=True):
         joinee = self.hs.parse_userid(event.target_user_id)
         # room_id = RoomID.from_string(event.room_id, self.hs)
         room_id = event.room_id
 
         # If event doesn't include a display name, add one.
-        yield self._fill_out_join_content(
-            joinee, event.content
+        yield self.distributor.fire(
+            "collect_presencelike_data", joinee, event.content
         )
 
         # XXX: We don't do an auth check if we are doing an invite
@@ -584,9 +615,9 @@ class RoomMemberHandler(BaseHandler):
         # that we are allowed to join when we decide whether or not we
         # need to do the invite/join dance.
 
-        room = yield self.store.get_room(room_id)
+        hosts = yield self.store.get_joined_hosts_for_room(room_id)
 
-        if room:
+        if self.hs.hostname in hosts:
             should_do_dance = False
         elif room_host:
             should_do_dance = True
@@ -598,7 +629,7 @@ class RoomMemberHandler(BaseHandler):
             if prev_state and prev_state.membership == Membership.INVITE:
                 room = yield self.store.get_room(room_id)
                 inviter = UserID.from_string(
-                    prev_state.sender, self.hs
+                    prev_state.user_id, self.hs
                 )
 
                 should_do_dance = not inviter.is_mine and not room
@@ -606,8 +637,15 @@ class RoomMemberHandler(BaseHandler):
             else:
                 should_do_dance = False
 
+        have_joined = False
+        if should_do_dance:
+            handler = self.hs.get_handlers().federation_handler
+            have_joined = yield handler.do_invite_join(
+                room_host, room_id, event.user_id, event.content
+            )
+
         # We want to do the _do_update inside the room lock.
-        if not should_do_dance:
+        if not have_joined:
             logger.debug("Doing normal join")
 
             if do_auth:
@@ -617,16 +655,6 @@ class RoomMemberHandler(BaseHandler):
             yield self._do_local_membership_update(
                 event,
                 membership=event.content["membership"],
-                broadcast_msg=broadcast_msg,
-            )
-
-
-        if should_do_dance:
-            yield self._do_invite_join_dance(
-                room_id=room_id,
-                joinee=event.user_id,
-                target_host=room_host,
-                content=event.content,
             )
 
         user = self.hs.parse_userid(event.user_id)
@@ -635,32 +663,6 @@ class RoomMemberHandler(BaseHandler):
         )
 
     @defer.inlineCallbacks
-    def _fill_out_join_content(self, user_id, content):
-        # If event doesn't include a display name, add one.
-        profile_handler = self.hs.get_handlers().profile_handler
-        if "displayname" not in content:
-            try:
-                display_name = yield profile_handler.get_displayname(
-                    user_id
-                )
-
-                if display_name:
-                    content["displayname"] = display_name
-            except:
-                logger.exception("Failed to set display_name")
-
-        if "avatar_url" not in content:
-            try:
-                avatar_url = yield profile_handler.get_avatar_url(
-                    user_id
-                )
-
-                if avatar_url:
-                    content["avatar_url"] = avatar_url
-            except:
-                logger.exception("Failed to set display_name")
-
-    @defer.inlineCallbacks
     def _should_invite_join(self, room_id, prev_state, do_auth):
         logger.debug("_should_invite_join: room_id: %s", room_id)
 
@@ -694,18 +696,12 @@ class RoomMemberHandler(BaseHandler):
             user_id=user.to_string(), membership_list=membership_list
         )
 
-        defer.returnValue([r["room_id"] for r in rooms])
+        defer.returnValue([r.room_id for r in rooms])
 
     @defer.inlineCallbacks
-    def _do_local_membership_update(self, event, membership, broadcast_msg):
+    def _do_local_membership_update(self, event, membership):
         # store membership
-        store_id = yield self.store.store_room_member(
-            user_id=event.target_user_id,
-            sender=event.user_id,
-            room_id=event.room_id,
-            content=event.content,
-            membership=membership
-        )
+        store_id = yield self.store.persist_event(event)
 
         # Send a PDU to all hosts who have joined the room.
         destinations = yield self.store.get_joined_hosts_for_room(
@@ -732,78 +728,11 @@ class RoomMemberHandler(BaseHandler):
         yield self.hs.get_federation().handle_new_event(event)
         self.notifier.on_new_room_event(event, store_id)
 
-        if broadcast_msg:
-            yield self._inject_membership_msg(
-                source=event.user_id,
-                target=event.target_user_id,
-                room_id=event.room_id,
-                membership=event.content["membership"]
-            )
-
-    @defer.inlineCallbacks
-    def _do_invite_join_dance(self, room_id, joinee, target_host, content):
-        logger.debug("Doing remote join dance")
-
-        # do invite join dance
-        federation = self.hs.get_federation()
-        new_event = self.event_factory.create_event(
-            etype=InviteJoinEvent.TYPE,
-            target_host=target_host,
-            room_id=room_id,
-            user_id=joinee,
-            content=content
-        )
-
-        new_event.destinations = [target_host]
-
-        yield self.store.store_room(
-            room_id, "", is_public=False
-        )
-
-        #yield self.state_handler.handle_new_event(event)
-        yield federation.handle_new_event(new_event)
-        yield federation.get_state_for_room(
-            target_host, room_id
-        )
-
-    @defer.inlineCallbacks
-    def _inject_membership_msg(self, room_id=None, source=None, target=None,
-                               membership=None):
-        # TODO this should be a different type of message, not m.text
-        if membership == Membership.INVITE:
-            body = "%s invited %s to the room." % (source, target)
-        elif membership == Membership.JOIN:
-            body = "%s joined the room." % (target)
-        elif membership == Membership.LEAVE:
-            body = "%s left the room." % (target)
-        else:
-            raise RoomError(500, "Unknown membership value %s" % membership)
-
-        membership_json = {
-            "msgtype": u"m.text",
-            "body": body,
-            "membership_source": source,
-            "membership_target": target,
-            "membership": membership,
-        }
-
-        msg_id = "m%s" % int(self.clock.time_msec())
-
-        event = self.event_factory.create_event(
-            etype=MessageEvent.TYPE,
-            room_id=room_id,
-            user_id="_homeserver_",
-            msg_id=msg_id,
-            content=membership_json
-        )
-
-        handler = self.hs.get_handlers().message_handler
-        yield handler.send_message(event, suppress_auth=True)
-
 
 class RoomListHandler(BaseHandler):
 
     @defer.inlineCallbacks
     def get_public_room_list(self):
-        chunk = yield self.store.get_rooms(is_public=True, with_topics=True)
+        chunk = yield self.store.get_rooms(is_public=True)
+        # FIXME (erikj): START is no longer a valid value
         defer.returnValue({"start": "START", "end": "END", "chunk": chunk})
diff --git a/synapse/http/server.py b/synapse/http/server.py
index c28d9a33f9..d1f99460c1 100644
--- a/synapse/http/server.py
+++ b/synapse/http/server.py
@@ -212,8 +212,9 @@ class ContentRepoResource(resource.Resource):
     """
     isLeaf = True
 
-    def __init__(self, directory, auth):
+    def __init__(self, hs, directory, auth):
         resource.Resource.__init__(self)
+        self.hs = hs
         self.directory = directory
         self.auth = auth
 
@@ -250,7 +251,8 @@ class ContentRepoResource(resource.Resource):
                 file_ext = re.sub("[^a-z]", "", file_ext)
                 suffix += "." + file_ext
 
-        file_path = os.path.join(self.directory, prefix + main_part + suffix)
+        file_name = prefix + main_part + suffix
+        file_path = os.path.join(self.directory, file_name)
         logger.info("User %s is uploading a file to path %s",
                     auth_user.to_string(),
                     file_path)
@@ -259,8 +261,8 @@ class ContentRepoResource(resource.Resource):
         attempts = 0
         while os.path.exists(file_path):
             main_part = random_string(24)
-            file_path = os.path.join(self.directory,
-                                     prefix + main_part + suffix)
+            file_name = prefix + main_part + suffix
+            file_path = os.path.join(self.directory, file_name)
             attempts += 1
             if attempts > 25:  # really? Really?
                 raise SynapseError(500, "Unable to create file.")
@@ -272,11 +274,14 @@ class ContentRepoResource(resource.Resource):
         # servers.
 
         # TODO: A little crude here, we could do this better.
-        filename = request.path.split(self.directory + "/")[1]
+        filename = request.path.split('/')[-1]
         # be paranoid
         filename = re.sub("[^0-9A-z.-_]", "", filename)
 
         file_path = self.directory + "/" + filename
+
+        logger.debug("Searching for %s", file_path)
+
         if os.path.isfile(file_path):
             # filename has the content type
             base64_contentype = filename.split(".")[1]
@@ -304,6 +309,10 @@ class ContentRepoResource(resource.Resource):
         self._async_render(request)
         return server.NOT_DONE_YET
 
+    def render_OPTIONS(self, request):
+        respond_with_json_bytes(request, 200, {}, send_cors=True)
+        return server.NOT_DONE_YET
+
     @defer.inlineCallbacks
     def _async_render(self, request):
         try:
@@ -313,8 +322,13 @@ class ContentRepoResource(resource.Resource):
             with open(fname, "wb") as f:
                 f.write(request.content.read())
 
+
+            # FIXME (erikj): These should use constants.
+            file_name = os.path.basename(fname)
+            url = "http://%s/matrix/content/%s" % (self.hs.hostname, file_name)
+
             respond_with_json_bytes(request, 200,
-                                    json.dumps({"content_token": fname}),
+                                    json.dumps({"content_token": url}),
                                     send_cors=True)
 
         except CodeMessageException as e:
diff --git a/synapse/rest/room.py b/synapse/rest/room.py
index db8f18e8b3..f5b547b963 100644
--- a/synapse/rest/room.py
+++ b/synapse/rest/room.py
@@ -115,7 +115,7 @@ class RoomTopicRestServlet(RestServlet):
 
         if not data:
             raise SynapseError(404, "Topic not found.", errcode=Codes.NOT_FOUND)
-        defer.returnValue((200, json.loads(data.content)))
+        defer.returnValue((200, data.content))
 
     @defer.inlineCallbacks
     def on_PUT(self, request, room_id):
@@ -177,7 +177,7 @@ class RoomMemberRestServlet(RestServlet):
         if not member:
             raise SynapseError(404, "Member not found.",
                                errcode=Codes.NOT_FOUND)
-        defer.returnValue((200, json.loads(member.content)))
+        defer.returnValue((200, member.content))
 
     @defer.inlineCallbacks
     def on_DELETE(self, request, roomid, target_user_id):
@@ -193,7 +193,7 @@ class RoomMemberRestServlet(RestServlet):
             )
 
         handler = self.handlers.room_member_handler
-        yield handler.change_membership(event, broadcast_msg=True)
+        yield handler.change_membership(event)
         defer.returnValue((200, ""))
 
     @defer.inlineCallbacks
@@ -220,7 +220,7 @@ class RoomMemberRestServlet(RestServlet):
             )
 
         handler = self.handlers.room_member_handler
-        yield handler.change_membership(event, broadcast_msg=True)
+        yield handler.change_membership(event)
         defer.returnValue((200, ""))
 
 
@@ -287,25 +287,28 @@ class FeedbackRestServlet(RestServlet):
                feedback_type):
         user = yield (self.auth.get_user_by_req(request))
 
-        if feedback_type not in Feedback.LIST:
-            raise SynapseError(400, "Bad feedback type.",
-                               errcode=Codes.BAD_JSON)
-
-        msg_handler = self.handlers.message_handler
-        feedback = yield msg_handler.get_feedback(
-            room_id=urllib.unquote(room_id),
-            msg_sender_id=msg_sender_id,
-            msg_id=msg_id,
-            user_id=user.to_string(),
-            fb_sender_id=fb_sender_id,
-            fb_type=feedback_type
-        )
-
-        if not feedback:
-            raise SynapseError(404, "Feedback not found.",
-                               errcode=Codes.NOT_FOUND)
+        # TODO (erikj): Implement this?
+        raise NotImplementedError("Getting feedback is not supported")
 
-        defer.returnValue((200, json.loads(feedback.content)))
+#        if feedback_type not in Feedback.LIST:
+#            raise SynapseError(400, "Bad feedback type.",
+#                               errcode=Codes.BAD_JSON)
+#
+#        msg_handler = self.handlers.message_handler
+#        feedback = yield msg_handler.get_feedback(
+#            room_id=urllib.unquote(room_id),
+#            msg_sender_id=msg_sender_id,
+#            msg_id=msg_id,
+#            user_id=user.to_string(),
+#            fb_sender_id=fb_sender_id,
+#            fb_type=feedback_type
+#        )
+#
+#        if not feedback:
+#            raise SynapseError(404, "Feedback not found.",
+#                               errcode=Codes.NOT_FOUND)
+#
+#        defer.returnValue((200, json.loads(feedback.content)))
 
     @defer.inlineCallbacks
     def on_PUT(self, request, room_id, sender_id, msg_id, fb_sender_id,
@@ -382,6 +385,21 @@ class RoomMessageListRestServlet(RestServlet):
         defer.returnValue((200, msgs))
 
 
+class RoomTriggerBackfill(RestServlet):
+    PATTERN = client_path_pattern("/rooms/(?P<room_id>[^/]*)/backfill$")
+
+    @defer.inlineCallbacks
+    def on_GET(self, request, room_id):
+        remote_server = urllib.unquote(request.args["remote"][0])
+        room_id = urllib.unquote(room_id)
+        limit = int(request.args["limit"][0])
+
+        handler = self.handlers.federation_handler
+        events = yield handler.backfill(remote_server, room_id, limit)
+
+        res = [event.get_dict() for event in events]
+        defer.returnValue((200, res))
+
 def _parse_json(request):
     try:
         content = json.loads(request.content.read())
@@ -402,3 +420,4 @@ def register_servlets(hs, http_server):
     RoomMemberListRestServlet(hs).register(http_server)
     RoomMessageListRestServlet(hs).register(http_server)
     JoinRoomAliasServlet(hs).register(http_server)
+    RoomTriggerBackfill(hs).register(http_server)
diff --git a/synapse/server.py b/synapse/server.py
index d4c2481483..c5b0a32757 100644
--- a/synapse/server.py
+++ b/synapse/server.py
@@ -159,7 +159,7 @@ class HomeServer(BaseHomeServer):
         return DataStore(self)
 
     def build_event_factory(self):
-        return EventFactory()
+        return EventFactory(self)
 
     def build_handlers(self):
         return Handlers(self)
diff --git a/synapse/state.py b/synapse/state.py
index 4f8b4d9760..ca8e1ca630 100644
--- a/synapse/state.py
+++ b/synapse/state.py
@@ -86,7 +86,7 @@ class StateHandler(object):
         else:
             event.depth = 0
 
-        current_state = yield self.store.get_current_state(
+        current_state = yield self.store.get_current_state_pdu(
             key.context, key.type, key.state_key
         )
 
diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py
index 5d5b5f7c44..d06033b980 100644
--- a/synapse/storage/__init__.py
+++ b/synapse/storage/__init__.py
@@ -13,30 +13,35 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from twisted.internet import defer
 
 from synapse.api.events.room import (
     RoomMemberEvent, MessageEvent, RoomTopicEvent, FeedbackEvent,
-    RoomConfigEvent
+    RoomConfigEvent, RoomNameEvent,
 )
 
+from synapse.util.logutils import log_function
+
 from .directory import DirectoryStore
 from .feedback import FeedbackStore
-from .message import MessageStore
 from .presence import PresenceStore
 from .profile import ProfileStore
 from .registration import RegistrationStore
 from .room import RoomStore
 from .roommember import RoomMemberStore
-from .roomdata import RoomDataStore
 from .stream import StreamStore
 from .pdu import StatePduStore, PduStore
 from .transactions import TransactionStore
 
 import json
+import logging
 import os
 
 
-class DataStore(RoomDataStore, RoomMemberStore, MessageStore, RoomStore,
+logger = logging.getLogger(__name__)
+
+
+class DataStore(RoomMemberStore, RoomStore,
                 RegistrationStore, StreamStore, ProfileStore, FeedbackStore,
                 PresenceStore, PduStore, StatePduStore, TransactionStore,
                 DirectoryStore):
@@ -44,51 +49,147 @@ class DataStore(RoomDataStore, RoomMemberStore, MessageStore, RoomStore,
     def __init__(self, hs):
         super(DataStore, self).__init__(hs)
         self.event_factory = hs.get_event_factory()
+        self.hs = hs
 
-    def persist_event(self, event):
-        if event.type == MessageEvent.TYPE:
-            return self.store_message(
-                user_id=event.user_id,
-                room_id=event.room_id,
-                msg_id=event.msg_id,
-                content=json.dumps(event.content)
-            )
-        elif event.type == RoomMemberEvent.TYPE:
-            return self.store_room_member(
-                user_id=event.target_user_id,
-                sender=event.user_id,
-                room_id=event.room_id,
-                content=event.content,
-                membership=event.content["membership"]
-            )
+        self.min_token_deferred = self._get_min_token()
+        self.min_token = None
+
+    @defer.inlineCallbacks
+    @log_function
+    def persist_event(self, event, backfilled=False):
+        if event.type == RoomMemberEvent.TYPE:
+            yield self._store_room_member(event)
         elif event.type == FeedbackEvent.TYPE:
-            return self.store_feedback(
-                room_id=event.room_id,
-                msg_id=event.msg_id,
-                msg_sender_id=event.msg_sender_id,
-                fb_sender_id=event.user_id,
-                fb_type=event.feedback_type,
-                content=json.dumps(event.content)
-            )
+            yield self._store_feedback(event)
+#        elif event.type == RoomConfigEvent.TYPE:
+#            yield self._store_room_config(event)
+        elif event.type == RoomNameEvent.TYPE:
+            yield self._store_room_name(event)
         elif event.type == RoomTopicEvent.TYPE:
-            return self.store_room_data(
-                room_id=event.room_id,
-                etype=event.type,
-                state_key=event.state_key,
-                content=json.dumps(event.content)
+            yield self._store_room_topic(event)
+
+        ret = yield self._store_event(event, backfilled)
+        defer.returnValue(ret)
+
+    @defer.inlineCallbacks
+    def get_event(self, event_id):
+        events_dict = yield self._simple_select_one(
+            "events",
+            {"event_id": event_id},
+            [
+                "event_id",
+                "type",
+                "sender",
+                "room_id",
+                "content",
+                "unrecognized_keys"
+            ],
+        )
+
+        event = self._parse_event_from_row(events_dict)
+        defer.returnValue(event)
+
+    @defer.inlineCallbacks
+    @log_function
+    def _store_event(self, event, backfilled):
+        # FIXME (erikj): This should be removed when we start amalgamating
+        # event and pdu storage
+        yield self.hs.get_federation().fill_out_prev_events(event)
+
+        vals = {
+            "topological_ordering": event.depth,
+            "event_id": event.event_id,
+            "type": event.type,
+            "room_id": event.room_id,
+            "content": json.dumps(event.content),
+            "processed": True,
+        }
+
+        if hasattr(event, "outlier"):
+            vals["outlier"] = event.outlier
+        else:
+            vals["outlier"] = False
+
+        if backfilled:
+            if not self.min_token_deferred.called:
+                yield self.min_token_deferred
+            self.min_token -= 1
+            vals["stream_ordering"] = self.min_token
+
+        unrec = {
+            k: v
+            for k, v in event.get_full_dict().items()
+            if k not in vals.keys()
+        }
+        vals["unrecognized_keys"] = json.dumps(unrec)
+
+        try:
+            yield self._simple_insert("events", vals)
+        except:
+            logger.exception(
+                "Failed to persist, probably duplicate: %s",
+                event.event_id
+            )
+            return
+
+        if not backfilled and hasattr(event, "state_key"):
+            vals = {
+                "event_id": event.event_id,
+                "room_id": event.room_id,
+                "type": event.type,
+                "state_key": event.state_key,
+            }
+
+            if hasattr(event, "prev_state"):
+                vals["prev_state"] = event.prev_state
+
+            yield self._simple_insert("state_events", vals)
+
+            yield self._simple_insert(
+                "current_state_events",
+                {
+                    "event_id": event.event_id,
+                    "room_id": event.room_id,
+                    "type": event.type,
+                    "state_key": event.state_key,
+                }
             )
-        elif event.type == RoomConfigEvent.TYPE:
-            if "visibility" in event.content:
-                visibility = event.content["visibility"]
-                return self.store_room_config(
-                    room_id=event.room_id,
-                    visibility=visibility
-                )
 
+        latest = yield self.get_room_events_max_id()
+        defer.returnValue(latest)
+
+    @defer.inlineCallbacks
+    def get_current_state(self, room_id, event_type=None, state_key=""):
+        sql = (
+            "SELECT e.* FROM events as e "
+            "INNER JOIN current_state_events as c ON e.event_id = c.event_id "
+            "INNER JOIN state_events as s ON e.event_id = s.event_id "
+            "WHERE c.room_id = ? "
+        )
+
+        if event_type:
+            sql += " AND s.type = ? AND s.state_key = ? "
+            args = (room_id, event_type, state_key)
         else:
-            raise NotImplementedError(
-                "Don't know how to persist type=%s" % event.type
-            )
+            args = (room_id, )
+
+        results = yield self._execute_and_decode(sql, *args)
+
+        defer.returnValue([self._parse_event_from_row(r) for r in results])
+
+    @defer.inlineCallbacks
+    def _get_min_token(self):
+        row = yield self._execute(
+            None,
+            "SELECT MIN(stream_ordering) FROM events"
+        )
+
+        self.min_token = row[0][0] if row and row[0] and row[0][0] else -1
+        self.min_token = min(self.min_token, -1)
+
+        logger.debug("min_token is: %s", self.min_token)
+
+        defer.returnValue(self.min_token)
 
 
 def schema_path(schema):
diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index bf1800f4bf..75aab2d3b9 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -12,7 +12,6 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-
 import logging
 
 from twisted.internet import defer
@@ -20,6 +19,9 @@ from twisted.internet import defer
 from synapse.api.errors import StoreError
 
 import collections
+import copy
+import json
+
 
 logger = logging.getLogger(__name__)
 
@@ -29,6 +31,7 @@ class SQLBaseStore(object):
     def __init__(self, hs):
         self.hs = hs
         self._db_pool = hs.get_db_pool()
+        self.event_factory = hs.get_event_factory()
         self._clock = hs.get_clock()
 
     def cursor_to_dict(self, cursor):
@@ -57,14 +60,22 @@ class SQLBaseStore(object):
             The result of decoder(results)
         """
         logger.debug(
-            "[SQL] %s  Args=%s Func=%s", query, args, decoder.__name__
+            "[SQL] %s  Args=%s Func=%s",
+            query, args, decoder.__name__ if decoder else None
         )
 
         def interaction(txn):
             cursor = txn.execute(query, args)
-            return decoder(cursor)
+            if decoder:
+                return decoder(cursor)
+            else:
+                return cursor.fetchall()
+
         return self._db_pool.runInteraction(interaction)
 
+    def _execute_and_decode(self, query, *args):
+        return self._execute(self.cursor_to_dict, query, *args)
+
     # "Simple" SQL API methods that operate on a single table with no JOINs,
     # no complex WHERE clauses, just a dict of values for columns.
 
@@ -281,6 +292,22 @@ class SQLBaseStore(object):
 
         return self._db_pool.runInteraction(func)
 
+    def _parse_event_from_row(self, row_dict):
+        d = copy.deepcopy({k: v for k, v in row_dict.items() if v})
+
+        d.pop("stream_ordering", None)
+        d.pop("topological_ordering", None)
+        d.pop("processed", None)
+
+        d.update(json.loads(row_dict["unrecognized_keys"]))
+        d["content"] = json.loads(d["content"])
+        del d["unrecognized_keys"]
+
+        return self.event_factory.create_event(
+            etype=d["type"],
+            **d
+        )
+
 
 class Table(object):
     """ A base class used to store information about a particular table.
diff --git a/synapse/storage/feedback.py b/synapse/storage/feedback.py
index 9bd562c762..e60f98d1e1 100644
--- a/synapse/storage/feedback.py
+++ b/synapse/storage/feedback.py
@@ -13,6 +13,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from twisted.internet import defer
+
 from ._base import SQLBaseStore, Table
 from synapse.api.events.room import FeedbackEvent
 
@@ -22,54 +24,28 @@ import json
 
 class FeedbackStore(SQLBaseStore):
 
-    def store_feedback(self, room_id, msg_id, msg_sender_id,
-                       fb_sender_id, fb_type, content):
-        return self._simple_insert(FeedbackTable.table_name, dict(
-            room_id=room_id,
-            msg_id=msg_id,
-            msg_sender_id=msg_sender_id,
-            fb_sender_id=fb_sender_id,
-            fb_type=fb_type,
-            content=content,
-        ))
-
-    def get_feedback(self, room_id=None, msg_id=None, msg_sender_id=None,
-                     fb_sender_id=None, fb_type=None):
-        query = FeedbackTable.select_statement(
-            "msg_sender_id = ? AND room_id = ? AND msg_id = ? " +
-            "AND fb_sender_id = ? AND feedback_type = ? " +
-            "ORDER BY id DESC LIMIT 1")
-        return self._execute(
-            FeedbackTable.decode_single_result,
-            query, msg_sender_id, room_id, msg_id, fb_sender_id, fb_type,
+    def _store_feedback(self, event):
+        return self._simple_insert("feedback", {
+            "event_id": event.event_id,
+            "feedback_type": event.feedback_type,
+            "room_id": event.room_id,
+            "target_event_id": event.target_event,
+            "sender": event.user_id,
+        })
+
+    @defer.inlineCallbacks
+    def get_feedback_for_event(self, event_id):
+        sql = (
+            "SELECT events.* FROM events INNER JOIN feedback "
+            "ON events.event_id = feedback.event_id "
+            "WHERE feedback.target_event_id = ? "
         )
 
-    def get_max_feedback_id(self):
-        return self._simple_max_id(FeedbackTable.table_name)
-
-
-class FeedbackTable(Table):
-    table_name = "feedback"
+        rows = yield self._execute_and_decode(sql, event_id)
 
-    fields = [
-        "id",
-        "content",
-        "feedback_type",
-        "fb_sender_id",
-        "msg_id",
-        "room_id",
-        "msg_sender_id"
-    ]
-
-    class EntryType(collections.namedtuple("FeedbackEntry", fields)):
-
-        def as_event(self, event_factory):
-            return event_factory.create_event(
-                etype=FeedbackEvent.TYPE,
-                room_id=self.room_id,
-                msg_id=self.msg_id,
-                msg_sender_id=self.msg_sender_id,
-                user_id=self.fb_sender_id,
-                feedback_type=self.feedback_type,
-                content=json.loads(self.content),
-            )
+        defer.returnValue(
+            [
+                self._parse_event_from_row(r)
+                for r in rows
+            ]
+        )
diff --git a/synapse/storage/message.py b/synapse/storage/message.py
deleted file mode 100644
index 7bb69c1384..0000000000
--- a/synapse/storage/message.py
+++ /dev/null
@@ -1,81 +0,0 @@
-# -*- coding: utf-8 -*-
-# Copyright 2014 matrix.org
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-from ._base import SQLBaseStore, Table
-from synapse.api.events.room import MessageEvent
-
-import collections
-import json
-
-
-class MessageStore(SQLBaseStore):
-
-    def get_message(self, user_id, room_id, msg_id):
-        """Get a message from the store.
-
-        Args:
-            user_id (str): The ID of the user who sent the message.
-            room_id (str): The room the message was sent in.
-            msg_id (str): The unique ID for this user/room combo.
-        """
-        query = MessagesTable.select_statement(
-            "user_id = ? AND room_id = ? AND msg_id = ? " +
-            "ORDER BY id DESC LIMIT 1")
-        return self._execute(
-            MessagesTable.decode_single_result,
-            query, user_id, room_id, msg_id,
-        )
-
-    def store_message(self, user_id, room_id, msg_id, content):
-        """Store a message in the store.
-
-        Args:
-            user_id (str): The ID of the user who sent the message.
-            room_id (str): The room the message was sent in.
-            msg_id (str): The unique ID for this user/room combo.
-            content (str): The content of the message (JSON)
-        """
-        return self._simple_insert(MessagesTable.table_name, dict(
-            user_id=user_id,
-            room_id=room_id,
-            msg_id=msg_id,
-            content=content,
-        ))
-
-    def get_max_message_id(self):
-        return self._simple_max_id(MessagesTable.table_name)
-
-
-class MessagesTable(Table):
-    table_name = "messages"
-
-    fields = [
-        "id",
-        "user_id",
-        "room_id",
-        "msg_id",
-        "content"
-    ]
-
-    class EntryType(collections.namedtuple("MessageEntry", fields)):
-
-        def as_event(self, event_factory):
-            return event_factory.create_event(
-                etype=MessageEvent.TYPE,
-                room_id=self.room_id,
-                user_id=self.user_id,
-                msg_id=self.msg_id,
-                content=json.loads(self.content),
-            )
diff --git a/synapse/storage/pdu.py b/synapse/storage/pdu.py
index 13adc581e1..7655f43ede 100644
--- a/synapse/storage/pdu.py
+++ b/synapse/storage/pdu.py
@@ -13,6 +13,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from twisted.internet import defer
+
 from ._base import SQLBaseStore, Table, JoinHelper
 
 from synapse.util.logutils import log_function
@@ -319,6 +321,7 @@ class PduStore(SQLBaseStore):
 
         return [(row[0], row[1], row[2]) for row in results]
 
+    @defer.inlineCallbacks
     def get_oldest_pdus_in_context(self, context):
         """Get a list of Pdus that we haven't backfilled beyond yet (and haven't    
         seen). This list is used when we want to backfill backwards and is the 
@@ -331,17 +334,14 @@ class PduStore(SQLBaseStore):
         Returns:
             list: A list of PduIdTuple.
         """
-        return self._db_pool.runInteraction(
-            self._get_oldest_pdus_in_context, context
-        )
-
-    def _get_oldest_pdus_in_context(self, txn, context):
-        txn.execute(
+        results = yield self._execute(
+            None,
             "SELECT pdu_id, origin FROM %(back)s WHERE context = ?"
             % {"back": PduBackwardExtremitiesTable.table_name, },
-            (context,)
+            context
         )
-        return [PduIdTuple(i, o) for i, o in txn.fetchall()]
+
+        defer.returnValue([PduIdTuple(i, o) for i, o in results])
 
     def is_pdu_new(self, pdu_id, origin, context, depth):
         """For a given Pdu, try and figure out if it's 'new', i.e., if it's
@@ -580,7 +580,7 @@ class StatePduStore(SQLBaseStore):
 
         txn.execute(query, query_args)
 
-    def get_current_state(self, context, pdu_type, state_key):
+    def get_current_state_pdu(self, context, pdu_type, state_key):
         """For a given context, pdu_type, state_key 3-tuple, return what is
         currently considered the current state.
 
@@ -595,10 +595,10 @@ class StatePduStore(SQLBaseStore):
         """
 
         return self._db_pool.runInteraction(
-            self._get_current_state, context, pdu_type, state_key
+            self._get_current_state_pdu, context, pdu_type, state_key
         )
 
-    def _get_current_state(self, txn, context, pdu_type, state_key):
+    def _get_current_state_pdu(self, txn, context, pdu_type, state_key):
         return self._get_current_interaction(txn, context, pdu_type, state_key)
 
     def _get_current_interaction(self, txn, context, pdu_type, state_key):
diff --git a/synapse/storage/room.py b/synapse/storage/room.py
index a97162831b..22f2dcca45 100644
--- a/synapse/storage/room.py
+++ b/synapse/storage/room.py
@@ -76,49 +76,80 @@ class RoomStore(SQLBaseStore):
         )
 
     @defer.inlineCallbacks
-    def get_rooms(self, is_public, with_topics):
+    def get_rooms(self, is_public):
         """Retrieve a list of all public rooms.
 
         Args:
             is_public (bool): True if the rooms returned should be public.
-            with_topics (bool): True to include the current topic for the room
-            in the response.
         Returns:
-            A list of room dicts containing at least a "room_id" key, and a
-            "topic" key if one is set and with_topic=True.
+            A list of room dicts containing at least a "room_id" key, a
+            "topic" key if one is set, and a "name" key if one is set
         """
-        room_data_type = RoomTopicEvent.TYPE
-        public = 1 if is_public else 0
-
-        latest_topic = ("SELECT max(room_data.id) FROM room_data WHERE "
-                        + "room_data.type = ? GROUP BY room_id")
-
-        query = ("SELECT rooms.*, room_data.content, room_alias FROM rooms "
-                 + "LEFT JOIN "
-                 + "room_aliases ON room_aliases.room_id = rooms.room_id "
-                 + "LEFT JOIN "
-                 + "room_data ON rooms.room_id = room_data.room_id WHERE "
-                 + "(room_data.id IN (" + latest_topic + ") "
-                 + "OR room_data.id IS NULL) AND rooms.is_public = ?")
-
-        res = yield self._execute(
-            self.cursor_to_dict, query, room_data_type, public
+
+        topic_subquery = (
+            "SELECT topics.event_id as event_id, "
+            "topics.room_id as room_id, topic "
+            "FROM topics "
+            "INNER JOIN current_state_events as c "
+            "ON c.event_id = topics.event_id "
         )
 
-        # return only the keys the specification expects
-        ret_keys = ["room_id", "topic", "room_alias"]
+        name_subquery = (
+            "SELECT room_names.event_id as event_id, "
+            "room_names.room_id as room_id, name "
+            "FROM room_names "
+            "INNER JOIN current_state_events as c "
+            "ON c.event_id = room_names.event_id "
+        )
 
-        # extract topic from the json (icky) FIXME
-        for i, room_row in enumerate(res):
-            try:
-                content_json = json.loads(room_row["content"])
-                room_row["topic"] = content_json["topic"]
-            except:
-                pass  # no topic set
-            # filter the dict based on ret_keys
-            res[i] = {k: v for k, v in room_row.iteritems() if k in ret_keys}
+        # We use non printing ascii character US () as a seperator
+        sql = (
+            "SELECT r.room_id, n.name, t.topic, "
+            "group_concat(a.room_alias, '') "
+            "FROM rooms AS r "
+            "LEFT JOIN (%(topic)s) AS t ON t.room_id = r.room_id "
+            "LEFT JOIN (%(name)s) AS n ON n.room_id = r.room_id "
+            "INNER JOIN room_aliases AS a ON a.room_id = r.room_id "
+            "WHERE r.is_public = ? "
+            "GROUP BY r.room_id "
+        ) % {
+            "topic": topic_subquery,
+            "name": name_subquery,
+        }
+
+        rows = yield self._execute(None, sql, is_public)
+
+        ret = [
+            {
+                "room_id": r[0],
+                "name": r[1],
+                "topic": r[2],
+                "aliases": r[3].split(""),
+            }
+            for r in rows
+        ]
+
+        defer.returnValue(ret)
+
+    def _store_room_topic(self, event):
+        return self._simple_insert(
+            "topics",
+            {
+                "event_id": event.event_id,
+                "room_id": event.room_id,
+                "topic": event.topic,
+            }
+        )
 
-        defer.returnValue(res)
+    def _store_room_name(self, event):
+        return self._simple_insert(
+            "room_names",
+            {
+                "event_id": event.event_id,
+                "room_id": event.room_id,
+                "name": event.name,
+            }
+        )
 
 
 class RoomsTable(Table):
diff --git a/synapse/storage/roomdata.py b/synapse/storage/roomdata.py
deleted file mode 100644
index cc04d1ba14..0000000000
--- a/synapse/storage/roomdata.py
+++ /dev/null
@@ -1,85 +0,0 @@
-# -*- coding: utf-8 -*-
-# Copyright 2014 matrix.org
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-from ._base import SQLBaseStore, Table
-
-import collections
-import json
-
-
-class RoomDataStore(SQLBaseStore):
-
-    """Provides various CRUD operations for Room Events. """
-
-    def get_room_data(self, room_id, etype, state_key=""):
-        """Retrieve the data stored under this type and state_key.
-
-        Args:
-            room_id (str)
-            etype (str)
-            state_key (str)
-        Returns:
-            namedtuple: Or None if nothing exists at this path.
-        """
-        query = RoomDataTable.select_statement(
-            "room_id = ? AND type = ? AND state_key = ? "
-            "ORDER BY id DESC LIMIT 1"
-        )
-        return self._execute(
-            RoomDataTable.decode_single_result,
-            query, room_id, etype, state_key,
-        )
-
-    def store_room_data(self, room_id, etype, state_key="", content=None):
-        """Stores room specific data.
-
-        Args:
-            room_id (str)
-            etype (str)
-            state_key (str)
-            data (str)- The data to store for this path in JSON.
-        Returns:
-            The store ID for this data.
-        """
-        return self._simple_insert(RoomDataTable.table_name, dict(
-            etype=etype,
-            state_key=state_key,
-            room_id=room_id,
-            content=content,
-        ))
-
-    def get_max_room_data_id(self):
-        return self._simple_max_id(RoomDataTable.table_name)
-
-
-class RoomDataTable(Table):
-    table_name = "room_data"
-
-    fields = [
-        "id",
-        "room_id",
-        "type",
-        "state_key",
-        "content"
-    ]
-
-    class EntryType(collections.namedtuple("RoomDataEntry", fields)):
-
-        def as_event(self, event_factory):
-            return event_factory.create_event(
-                etype=self.type,
-                room_id=self.room_id,
-                content=json.loads(self.content),
-            )
diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py
index c45d128f1b..89c87290cf 100644
--- a/synapse/storage/roommember.py
+++ b/synapse/storage/roommember.py
@@ -31,6 +31,38 @@ logger = logging.getLogger(__name__)
 
 class RoomMemberStore(SQLBaseStore):
 
+    @defer.inlineCallbacks
+    def _store_room_member(self, event):
+        """Store a room member in the database.
+        """
+        domain = self.hs.parse_userid(event.target_user_id).domain
+
+        yield self._simple_insert(
+            "room_memberships",
+            {
+                "event_id": event.event_id,
+                "user_id": event.target_user_id,
+                "sender": event.user_id,
+                "room_id": event.room_id,
+                "membership": event.membership,
+            }
+        )
+
+        # Update room hosts table
+        if event.membership == Membership.JOIN:
+            sql = (
+                "INSERT OR IGNORE INTO room_hosts (room_id, host) "
+                "VALUES (?, ?)"
+            )
+            yield self._execute(None, sql, event.room_id, domain)
+        else:
+            sql = (
+                "DELETE FROM room_hosts WHERE room_id = ? AND host = ?"
+            )
+
+            yield self._execute(None, sql, event.room_id, domain)
+
+    @defer.inlineCallbacks
     def get_room_member(self, user_id, room_id):
         """Retrieve the current state of a room member.
 
@@ -38,36 +70,15 @@ class RoomMemberStore(SQLBaseStore):
             user_id (str): The member's user ID.
             room_id (str): The room the member is in.
         Returns:
-            namedtuple: The room member from the database, or None if this
-            member does not exist.
+            Deferred: Results in a MembershipEvent or None.
         """
-        query = RoomMemberTable.select_statement(
-            "room_id = ? AND user_id = ? ORDER BY id DESC LIMIT 1")
-        return self._execute(
-            RoomMemberTable.decode_single_result,
-            query, room_id, user_id,
-        )
+        rows = yield self._get_members_by_dict({
+            "e.room_id": room_id,
+            "m.user_id": user_id,
+        })
 
-    def store_room_member(self, user_id, sender, room_id, membership, content):
-        """Store a room member in the database.
+        defer.returnValue(rows[0] if rows else None)
 
-        Args:
-            user_id (str): The member's user ID.
-            room_id (str): The room in relation to the member.
-            membership (synapse.api.constants.Membership): The new membership
-            state.
-            content (dict): The content of the membership (JSON).
-        """
-        content_json = json.dumps(content)
-        return self._simple_insert(RoomMemberTable.table_name, dict(
-            user_id=user_id,
-            sender=sender,
-            room_id=room_id,
-            membership=membership,
-            content=content_json,
-        ))
-
-    @defer.inlineCallbacks
     def get_room_members(self, room_id, membership=None):
         """Retrieve the current room member list for a room.
 
@@ -79,17 +90,12 @@ class RoomMemberStore(SQLBaseStore):
         Returns:
             list of namedtuples representing the members in this room.
         """
-        query = RoomMemberTable.select_statement(
-            "id IN (SELECT MAX(id) FROM " + RoomMemberTable.table_name
-            + " WHERE room_id = ? GROUP BY user_id)"
-        )
-        res = yield self._execute(
-            RoomMemberTable.decode_results, query, room_id,
-        )
-        # strip memberships which don't match
+
+        where = {"m.room_id": room_id}
         if membership:
-            res = [entry for entry in res if entry.membership == membership]
-        defer.returnValue(res)
+            where["m.membership"] = membership
+
+        return self._get_members_by_dict(where)
 
     def get_rooms_for_user_where_membership_is(self, user_id, membership_list):
         """ Get all the rooms for this user where the membership for this user
@@ -106,70 +112,40 @@ class RoomMemberStore(SQLBaseStore):
             return defer.succeed(None)
 
         args = [user_id]
-        membership_placeholder = ["membership=?"] * len(membership_list)
-        where_membership = "(" + " OR ".join(membership_placeholder) + ")"
-        for membership in membership_list:
-            args.append(membership)
-
-        # sub-select finds the row ID for the most recent (i.e. current)
-        # state change of this user per room, then the outer select finds those
-        query = ("SELECT room_id, membership FROM room_memberships"
-                 + " WHERE id IN (SELECT MAX(id) FROM room_memberships"
-                 + "   WHERE user_id=? GROUP BY room_id)"
-                 + " AND " + where_membership)
-        return self._execute(
-            self.cursor_to_dict, query, *args
-        )
+        args.extend(membership_list)
 
-    @defer.inlineCallbacks
-    def get_joined_hosts_for_room(self, room_id):
-        query = RoomMemberTable.select_statement(
-            "id IN (SELECT MAX(id) FROM " + RoomMemberTable.table_name
-            + " WHERE room_id = ? GROUP BY user_id)"
-        )
-
-        res = yield self._execute(
-            RoomMemberTable.decode_results, query, room_id,
+        where_clause = "user_id = ? AND (%s)" % (
+            " OR ".join(["membership = ?" for _ in membership_list]),
         )
 
-        def host_from_user_id_string(user_id):
-            domain = UserID.from_string(entry.user_id, self.hs).domain
-            return domain
-
-        # strip memberships which don't match
-        hosts = [
-            host_from_user_id_string(entry.user_id)
-            for entry in res
-            if entry.membership == Membership.JOIN
-        ]
+        return self._get_members_query(where_clause, args)
 
-        logger.debug("Returning hosts: %s from results: %s", hosts, res)
-
-        defer.returnValue(hosts)
-
-    def get_max_room_member_id(self):
-        return self._simple_max_id(RoomMemberTable.table_name)
-
-
-class RoomMemberTable(Table):
-    table_name = "room_memberships"
-
-    fields = [
-        "id",
-        "user_id",
-        "sender",
-        "room_id",
-        "membership",
-        "content"
-    ]
+    def get_joined_hosts_for_room(self, room_id):
+        return self._simple_select_onecol(
+            "room_hosts",
+            {"room_id": room_id},
+            "host"
+        )
 
-    class EntryType(collections.namedtuple("RoomMemberEntry", fields)):
+    def _get_members_by_dict(self, where_dict):
+        clause = " AND ".join("%s = ?" % k for k in where_dict.keys())
+        vals = where_dict.values()
+        return self._get_members_query(clause, vals)
 
-        def as_event(self, event_factory):
-            return event_factory.create_event(
-                etype=RoomMemberEvent.TYPE,
-                room_id=self.room_id,
-                target_user_id=self.user_id,
-                user_id=self.sender,
-                content=json.loads(self.content),
-            )
+    @defer.inlineCallbacks
+    def _get_members_query(self, where_clause, where_values):
+        sql = (
+            "SELECT e.* FROM events as e "
+            "INNER JOIN room_memberships as m "
+            "ON e.event_id = m.event_id "
+            "INNER JOIN current_state_events as c "
+            "ON m.event_id = c.event_id "
+            "WHERE %s "
+        ) % (where_clause,)
+
+        rows = yield self._execute_and_decode(sql, *where_values)
+
+        logger.debug("_get_members_query Got rows %s", rows)
+
+        results = [self._parse_event_from_row(r) for r in rows]
+        defer.returnValue(results)
diff --git a/synapse/storage/schema/im.sql b/synapse/storage/schema/im.sql
index 77096546b2..39a1ed703e 100644
--- a/synapse/storage/schema/im.sql
+++ b/synapse/storage/schema/im.sql
@@ -12,43 +12,71 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-CREATE TABLE IF NOT EXISTS rooms(
-    room_id TEXT PRIMARY KEY NOT NULL,
-    is_public INTEGER,
-    creator TEXT
+
+CREATE TABLE IF NOT EXISTS events(
+    stream_ordering INTEGER PRIMARY KEY AUTOINCREMENT,
+    topological_ordering INTEGER NOT NULL,
+    event_id TEXT NOT NULL,
+    type TEXT NOT NULL,
+    room_id TEXT NOT NULL,
+    content TEXT NOT NULL,
+    unrecognized_keys TEXT,
+    processed BOOL NOT NULL,
+    outlier BOOL NOT NULL,
+    CONSTRAINT ev_uniq UNIQUE (event_id)
 );
 
-CREATE TABLE IF NOT EXISTS room_memberships(
-    id INTEGER PRIMARY KEY AUTOINCREMENT,
-    user_id TEXT NOT NULL, -- no foreign key to users table, it could be an id belonging to another home server
-    sender TEXT NOT NULL,
+CREATE TABLE IF NOT EXISTS state_events(
+    event_id TEXT NOT NULL,
     room_id TEXT NOT NULL,
-    membership TEXT NOT NULL,
-    content TEXT NOT NULL
+    type TEXT NOT NULL,
+    state_key TEXT NOT NULL,
+    prev_state TEXT
 );
 
-CREATE TABLE IF NOT EXISTS messages(
-    id INTEGER PRIMARY KEY AUTOINCREMENT,
-    user_id TEXT, 
-    room_id TEXT,
-    msg_id TEXT,
-    content TEXT
+CREATE TABLE IF NOT EXISTS current_state_events(
+    event_id TEXT NOT NULL,
+    room_id TEXT NOT NULL,
+    type TEXT NOT NULL,
+    state_key TEXT NOT NULL,
+    CONSTRAINT curr_uniq UNIQUE (room_id, type, state_key) ON CONFLICT REPLACE
+);
+
+CREATE TABLE IF NOT EXISTS room_memberships(
+    event_id TEXT NOT NULL,
+    user_id TEXT NOT NULL,
+    sender TEXT NOT NULL,
+    room_id TEXT NOT NULL,
+    membership TEXT NOT NULL
 );
 
 CREATE TABLE IF NOT EXISTS feedback(
-    id INTEGER PRIMARY KEY AUTOINCREMENT,
-    content TEXT,
+    event_id TEXT NOT NULL,
     feedback_type TEXT,
-    fb_sender_id TEXT,
-    msg_id TEXT,
-    room_id TEXT,
-    msg_sender_id TEXT
+    target_event_id TEXT,
+    sender TEXT,
+    room_id TEXT
 );
 
-CREATE TABLE IF NOT EXISTS room_data(
-    id INTEGER PRIMARY KEY AUTOINCREMENT,
+CREATE TABLE IF NOT EXISTS topics(
+    event_id TEXT NOT NULL,
     room_id TEXT NOT NULL,
-    type TEXT NOT NULL,
-    state_key TEXT NOT NULL,
-    content TEXT
+    topic TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS room_names(
+    event_id TEXT NOT NULL,
+    room_id TEXT NOT NULL,
+    name TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS rooms(
+    room_id TEXT PRIMARY KEY NOT NULL,
+    is_public INTEGER,
+    creator TEXT
+);
+
+CREATE TABLE IF NOT EXISTS room_hosts(
+    room_id TEXT NOT NULL,
+    host TEXT NOT NULL
 );
diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py
index 47a1f2c45a..87ae961ccd 100644
--- a/synapse/storage/stream.py
+++ b/synapse/storage/stream.py
@@ -13,267 +13,287 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+""" This module is responsible for getting events from the DB for pagination
+and event streaming.
+
+The order it returns events in depend on whether we are streaming forwards or
+are paginating backwards. We do this because we want to handle out of order
+messages nicely, while still returning them in the correct order when we
+paginate bacwards.
+
+This is implemented by keeping two ordering columns: stream_ordering and
+topological_ordering. Stream ordering is basically insertion/received order
+(except for events from backfill requests). The topolgical_ordering is a
+weak ordering of events based on the pdu graph.
+
+This means that we have to have two different types of tokens, depending on
+what sort order was used:
+    - stream tokens are of the form: "s%d", which maps directly to the column
+    - topological tokems: "t%d-%d", where the integers map to the topological
+      and stream ordering columns respectively.
+"""
+
+from twisted.internet import defer
 
 from ._base import SQLBaseStore
-from .message import MessagesTable
-from .feedback import FeedbackTable
-from .roomdata import RoomDataTable
-from .roommember import RoomMemberTable
+from synapse.api.errors import SynapseError
+from synapse.api.constants import Membership
+from synapse.util.logutils import log_function
 
 import json
 import logging
 
+
 logger = logging.getLogger(__name__)
 
 
+MAX_STREAM_SIZE = 1000
+
+
+_STREAM_TOKEN = "stream"
+_TOPOLOGICAL_TOKEN = "topological"
+
+
+def _parse_stream_token(string):
+    try:
+        if string[0] != 's':
+            raise
+        return int(string[1:])
+    except:
+        raise SynapseError(400, "Invalid token")
+
+
+def _parse_topological_token(string):
+    try:
+        if string[0] != 't':
+            raise
+        parts = string[1:].split('-', 1)
+        return (int(parts[0]), int(parts[1]))
+    except:
+        raise SynapseError(400, "Invalid token")
+
+
+def is_stream_token(string):
+    try:
+        _parse_stream_token(string)
+        return True
+    except:
+        return False
+
+
+def is_topological_token(string):
+    try:
+        _parse_topological_token(string)
+        return True
+    except:
+        return False
+
+
+def _get_token_bound(token, comparison):
+    try:
+        s = _parse_stream_token(token)
+        return "%s %s %d" % ("stream_ordering", comparison, s)
+    except:
+        pass
+
+    try:
+        top, stream = _parse_topological_token(token)
+        return "%s %s %d AND %s %s %d" % (
+            "topological_ordering", comparison, top,
+            "stream_ordering", comparison, stream,
+        )
+    except:
+        pass
+
+    raise SynapseError(400, "Invalid token")
+
+
 class StreamStore(SQLBaseStore):
+    @log_function
+    def get_room_events(self, user_id, from_key, to_key, room_id, limit=0,
+                        direction='f', with_feedback=False):
+        # We deal with events request in two different ways depending on if
+        # this looks like an /events request or a pagination request.
+        is_events = (
+            direction == 'f'
+            and user_id
+            and is_stream_token(from_key)
+            and to_key and is_stream_token(to_key)
+        )
 
-    def get_message_stream(self, user_id, from_key, to_key, room_id, limit=0,
-                           with_feedback=False):
-        """Get all messages for this user between the given keys.
-
-        Args:
-            user_id (str): The user who is requesting messages.
-            from_key (int): The ID to start returning results from (exclusive).
-            to_key (int): The ID to stop returning results (exclusive).
-            room_id (str): Gets messages only for this room. Can be None, in
-            which case all room messages will be returned.
-        Returns:
-            A tuple of rows (list of namedtuples), new_id(int)
-        """
-        if with_feedback and room_id:  # with fb MUST specify a room ID
-            return self._db_pool.runInteraction(
-                self._get_message_rows_with_feedback,
-                user_id, from_key, to_key, room_id, limit
+        if is_events:
+            return self.get_room_events_stream(
+                user_id=user_id,
+                from_key=from_key,
+                to_key=to_key,
+                room_id=room_id,
+                limit=limit,
+                with_feedback=with_feedback,
             )
         else:
-            return self._db_pool.runInteraction(
-                self._get_message_rows,
-                user_id, from_key, to_key, room_id, limit
+            return self.paginate_room_events(
+                from_key=from_key,
+                to_key=to_key,
+                room_id=room_id,
+                limit=limit,
+                with_feedback=with_feedback,
             )
 
-    def _get_message_rows(self, txn, user_id, from_pkey, to_pkey, room_id,
-                          limit):
-        # work out which rooms this user is joined in on and join them with
-        # the room id on the messages table, bounded by the specified pkeys
+    @defer.inlineCallbacks
+    @log_function
+    def get_room_events_stream(self, user_id, from_key, to_key, room_id,
+                               limit=0, with_feedback=False):
+        # TODO (erikj): Handle compressed feedback
 
-        # get all messages where the *current* membership state is 'join' for
-        # this user in that room.
-        query = ("SELECT messages.* FROM messages WHERE ? IN"
-                 + " (SELECT membership from room_memberships WHERE user_id=?"
-                 + " AND room_id = messages.room_id ORDER BY id DESC LIMIT 1)")
-        query_args = ["join", user_id]
+        current_room_membership_sql = (
+            "SELECT m.room_id FROM room_memberships as m "
+            "INNER JOIN current_state_events as c ON m.event_id = c.event_id "
+            "WHERE m.user_id = ?"
+        )
 
-        if room_id:
-            query += " AND messages.room_id=?"
-            query_args.append(room_id)
+        # We also want to get any membership events about that user, e.g.
+        # invites or leave notifications.
+        membership_sql = (
+            "SELECT m.event_id FROM room_memberships as m "
+            "INNER JOIN current_state_events as c ON m.event_id = c.event_id "
+            "WHERE m.user_id = ? "
+        )
 
-        (query, query_args) = self._append_stream_operations(
-            "messages", query, query_args, from_pkey, to_pkey, limit=limit
+        if limit:
+            limit = max(limit, MAX_STREAM_SIZE)
+        else:
+            limit = MAX_STREAM_SIZE
+
+        # From and to keys should be integers from ordering.
+        from_id = _parse_stream_token(from_key)
+        to_id = _parse_stream_token(to_key)
+
+        if from_key == to_key:
+            defer.returnValue(([], to_key))
+            return
+
+        sql = (
+            "SELECT * FROM events as e WHERE "
+            "((room_id IN (%(current)s)) OR "
+            "(event_id IN (%(invites)s))) "
+            "AND e.stream_ordering > ? AND e.stream_ordering < ? "
+            "AND e.outlier = 0 "
+            "ORDER BY stream_ordering ASC LIMIT %(limit)d "
+        ) % {
+            "current": current_room_membership_sql,
+            "invites": membership_sql,
+            "limit": limit
+        }
+
+        rows = yield self._execute_and_decode(
+            sql,
+            user_id, user_id, from_id, to_id
         )
 
-        cursor = txn.execute(query, query_args)
-        return self._as_events(cursor, MessagesTable, from_pkey)
+        ret = [self._parse_event_from_row(r) for r in rows]
 
-    def _get_message_rows_with_feedback(self, txn, user_id, from_pkey, to_pkey,
-                                        room_id, limit):
-        # this col represents the compressed feedback JSON as per spec
-        compressed_feedback_col = (
-            "'[' || group_concat('{\"sender_id\":\"' || f.fb_sender_id"
-            + " || '\",\"feedback_type\":\"' || f.feedback_type"
-            + " || '\",\"content\":' || f.content || '}') || ']'"
-        )
+        if rows:
+            key = "s%d" % max([r["stream_ordering"] for r in rows])
+        else:
+            # Assume we didn't get anything because there was nothing to get.
+            key = to_key
+
+        defer.returnValue((ret, key))
 
-        global_msg_id_join = ("f.room_id = messages.room_id"
-                              + " and f.msg_id = messages.msg_id"
-                              + " and messages.user_id = f.msg_sender_id")
+    @defer.inlineCallbacks
+    @log_function
+    def paginate_room_events(self, room_id, from_key, to_key=None,
+                             direction='b', limit=-1,
+                             with_feedback=False):
+        # TODO (erikj): Handle compressed feedback
 
-        select_query = (
-            "SELECT messages.*, f.content AS fb_content, f.fb_sender_id"
-            + ", " + compressed_feedback_col + " AS compressed_fb"
-            + " FROM messages LEFT JOIN feedback f ON " + global_msg_id_join)
+        from_comp = '<' if direction =='b' else '>'
+        to_comp = '>' if direction =='b' else '<'
+        order = "DESC" if direction == 'b' else "ASC"
 
-        current_membership_sub_query = (
-            "(SELECT membership from room_memberships rm"
-            + " WHERE user_id=? AND room_id = rm.room_id"
-            + " ORDER BY id DESC LIMIT 1)")
+        args = [room_id]
 
-        where = (" WHERE ? IN " + current_membership_sub_query
-                 + " AND messages.room_id=?")
+        bounds = _get_token_bound(from_key, from_comp)
+        if to_key:
+            bounds = "%s AND %s" % (bounds, _get_token_bound(to_key, to_comp))
+
+        if int(limit) > 0:
+            args.append(int(limit))
+            limit_str = " LIMIT ?"
+        else:
+            limit_str = ""
 
-        query = select_query + where
-        query_args = ["join", user_id, room_id]
+        sql = (
+            "SELECT * FROM events "
+            "WHERE outlier = 0 AND room_id = ? AND %(bounds)s "
+            "ORDER BY topological_ordering %(order)s, stream_ordering %(order)s %(limit)s "
+        ) % {"bounds": bounds, "order": order, "limit": limit_str}
 
-        (query, query_args) = self._append_stream_operations(
-            "messages", query, query_args, from_pkey, to_pkey,
-            limit=limit, group_by=" GROUP BY messages.id "
+        rows = yield self._execute_and_decode(
+            sql,
+            *args
         )
 
-        cursor = txn.execute(query, query_args)
-
-        # convert the result set into events
-        entries = self.cursor_to_dict(cursor)
-        events = []
-        for entry in entries:
-            # TODO we should spec the cursor > event mapping somewhere else.
-            event = {}
-            straight_mappings = ["msg_id", "user_id", "room_id"]
-            for key in straight_mappings:
-                event[key] = entry[key]
-            event["content"] = json.loads(entry["content"])
-            if entry["compressed_fb"]:
-                event["feedback"] = json.loads(entry["compressed_fb"])
-            events.append(event)
-
-        latest_pkey = from_pkey if len(entries) == 0 else entries[-1]["id"]
-
-        return (events, latest_pkey)
-
-    def get_room_member_stream(self, user_id, from_key, to_key):
-        """Get all room membership events for this user between the given keys.
-
-        Args:
-            user_id (str): The user who is requesting membership events.
-            from_key (int): The ID to start returning results from (exclusive).
-            to_key (int): The ID to stop returning results (exclusive).
-        Returns:
-            A tuple of rows (list of namedtuples), new_id(int)
-        """
-        return self._db_pool.runInteraction(
-            self._get_room_member_rows, user_id, from_key, to_key
+        if rows:
+            topo = rows[-1]["topological_ordering"]
+            toke = rows[-1]["stream_ordering"]
+            next_token = "t%s-%s" % (topo, toke)
+        else:
+            # TODO (erikj): We should work out what to do here instead.
+            next_token = to_key if to_key else from_key
+
+        defer.returnValue(
+            (
+                [self._parse_event_from_row(r) for r in rows],
+                next_token
+            )
         )
 
-    def _get_room_member_rows(self, txn, user_id, from_pkey, to_pkey):
-        # get all room membership events for rooms which the user is
-        # *currently* joined in on, or all invite events for this user.
-        current_membership_sub_query = (
-            "(SELECT membership FROM room_memberships"
-            + " WHERE user_id=? AND room_id = rm.room_id"
-            + " ORDER BY id DESC LIMIT 1)")
-
-        query = ("SELECT rm.* FROM room_memberships rm "
-                 # all membership events for rooms you've currently joined.
-                 + " WHERE (? IN " + current_membership_sub_query
-                 # all invite membership events for this user
-                 + " OR rm.membership=? AND user_id=?)"
-                 + " AND rm.id > ?")
-        query_args = ["join", user_id, "invite", user_id, from_pkey]
-
-        if to_pkey != -1:
-            query += " AND rm.id < ?"
-            query_args.append(to_pkey)
-
-        cursor = txn.execute(query, query_args)
-        return self._as_events(cursor, RoomMemberTable, from_pkey)
-
-    def get_feedback_stream(self, user_id, from_key, to_key, room_id, limit=0):
-        return self._db_pool.runInteraction(
-            self._get_feedback_rows,
-            user_id, from_key, to_key, room_id, limit
+    @defer.inlineCallbacks
+    def get_recent_events_for_room(self, room_id, limit, end_token,
+                                   with_feedback=False):
+        # TODO (erikj): Handle compressed feedback
+
+        sql = (
+            "SELECT * FROM events "
+            "WHERE room_id = ? AND stream_ordering <= ? "
+            "ORDER BY topological_ordering, stream_ordering DESC LIMIT ? "
         )
 
-    def _get_feedback_rows(self, txn, user_id, from_pkey, to_pkey, room_id,
-                           limit):
-        # work out which rooms this user is joined in on and join them with
-        # the room id on the feedback table, bounded by the specified pkeys
-
-        # get all messages where the *current* membership state is 'join' for
-        # this user in that room.
-        query = (
-            "SELECT feedback.* FROM feedback WHERE ? IN "
-            + "(SELECT membership from room_memberships WHERE user_id=?"
-            + " AND room_id = feedback.room_id ORDER BY id DESC LIMIT 1)")
-        query_args = ["join", user_id]
-
-        if room_id:
-            query += " AND feedback.room_id=?"
-            query_args.append(room_id)
-
-        (query, query_args) = self._append_stream_operations(
-            "feedback", query, query_args, from_pkey, to_pkey, limit=limit
+        rows = yield self._execute_and_decode(
+            sql,
+            room_id, end_token, limit
         )
 
-        cursor = txn.execute(query, query_args)
-        return self._as_events(cursor, FeedbackTable, from_pkey)
+        rows.reverse()  # As we selected with reverse ordering
+
+        if rows:
+            topo = rows[0]["topological_ordering"]
+            toke = rows[0]["stream_ordering"]
+            start_token = "t%s-%s" % (topo, toke)
 
-    def get_room_data_stream(self, user_id, from_key, to_key, room_id,
-                             limit=0):
-        return self._db_pool.runInteraction(
-            self._get_room_data_rows,
-            user_id, from_key, to_key, room_id, limit
+            token = (start_token, end_token)
+        else:
+            token = (end_token, end_token)
+
+        defer.returnValue(
+            (
+                [self._parse_event_from_row(r) for r in rows],
+                token
+            )
         )
 
-    def _get_room_data_rows(self, txn, user_id, from_pkey, to_pkey, room_id,
-                            limit):
-        # work out which rooms this user is joined in on and join them with
-        # the room id on the feedback table, bounded by the specified pkeys
-
-        # get all messages where the *current* membership state is 'join' for
-        # this user in that room.
-        query = (
-            "SELECT room_data.* FROM room_data WHERE ? IN "
-            + "(SELECT membership from room_memberships WHERE user_id=?"
-            + " AND room_id = room_data.room_id ORDER BY id DESC LIMIT 1)")
-        query_args = ["join", user_id]
-
-        if room_id:
-            query += " AND room_data.room_id=?"
-            query_args.append(room_id)
-
-        (query, query_args) = self._append_stream_operations(
-            "room_data", query, query_args, from_pkey, to_pkey, limit=limit
+    @defer.inlineCallbacks
+    def get_room_events_max_id(self):
+        res = yield self._execute_and_decode(
+            "SELECT MAX(stream_ordering) as m FROM events"
         )
 
-        cursor = txn.execute(query, query_args)
-        return self._as_events(cursor, RoomDataTable, from_pkey)
-
-    def _append_stream_operations(self, table_name, query, query_args,
-                                  from_pkey, to_pkey, limit=None,
-                                  group_by=""):
-        LATEST_ROW = -1
-        order_by = ""
-        if to_pkey > from_pkey:
-            if from_pkey != LATEST_ROW:
-                # e.g. from=5 to=9 >> from 5 to 9 >> id>5 AND id<9
-                query += (" AND %s.id > ? AND %s.id < ?" %
-                         (table_name, table_name))
-                query_args.append(from_pkey)
-                query_args.append(to_pkey)
-            else:
-                # e.g. from=-1 to=5 >> from now to 5 >> id>5 ORDER BY id DESC
-                query += " AND %s.id > ? " % table_name
-                order_by = "ORDER BY id DESC"
-                query_args.append(to_pkey)
-        elif from_pkey > to_pkey:
-            if to_pkey != LATEST_ROW:
-                # from=9 to=5 >> from 9 to 5 >> id>5 AND id<9 ORDER BY id DESC
-                query += (" AND %s.id > ? AND %s.id < ? " %
-                          (table_name, table_name))
-                order_by = "ORDER BY id DESC"
-                query_args.append(to_pkey)
-                query_args.append(from_pkey)
-            else:
-                # from=5 to=-1 >> from 5 to now >> id>5
-                query += " AND %s.id > ?" % table_name
-                query_args.append(from_pkey)
-
-        query += group_by + order_by
-
-        if limit and limit > 0:
-            query += " LIMIT ?"
-            query_args.append(str(limit))
-
-        return (query, query_args)
-
-    def _as_events(self, cursor, table, from_pkey):
-        data_entries = table.decode_results(cursor)
-        last_pkey = from_pkey
-        if data_entries:
-            last_pkey = data_entries[-1].id
-
-        events = [
-            entry.as_event(self.event_factory).get_dict()
-            for entry in data_entries
-        ]
-
-        return (events, last_pkey)
+        logger.debug("get_room_events_max_id: %s", res)
+
+        if not res or not res[0] or not res[0]["m"]:
+            defer.returnValue("s1")
+            return
+
+        key = res[0]["m"] + 1
+        defer.returnValue("s%d" % (key,))
diff --git a/synapse/types.py b/synapse/types.py
index b8e191bb3c..fd6a3d1d72 100644
--- a/synapse/types.py
+++ b/synapse/types.py
@@ -38,6 +38,14 @@ class DomainSpecificString(
     def __iter__(self):
         raise ValueError("Attempted to iterate a %s" % (type(self).__name__))
 
+    # Because this class is a namedtuple of strings and booleans, it is deeply
+    # immutable.
+    def __copy__(self):
+        return self
+
+    def __deepcopy__(self, memo):
+        return self
+
     @classmethod
     def from_string(cls, s, hs):
         """Parse the string given by 's' into a structure object."""