summary refs log tree commit diff
path: root/synapse
diff options
context:
space:
mode:
Diffstat (limited to 'synapse')
-rw-r--r--synapse/api/auth.py51
-rw-r--r--synapse/api/constants.py11
-rwxr-xr-xsynapse/app/homeserver.py55
-rwxr-xr-xsynapse/app/synctl.py12
-rw-r--r--synapse/config/_base.py45
-rw-r--r--synapse/config/appservice.py2
-rw-r--r--synapse/config/captcha.py2
-rw-r--r--synapse/config/database.py2
-rw-r--r--synapse/config/key.py2
-rw-r--r--synapse/config/logger.py2
-rw-r--r--synapse/config/metrics.py8
-rw-r--r--synapse/config/ratelimiting.py2
-rw-r--r--synapse/config/registration.py2
-rw-r--r--synapse/config/repository.py2
-rw-r--r--synapse/config/saml2.py2
-rw-r--r--synapse/config/server.py2
-rw-r--r--synapse/config/tls.py2
-rw-r--r--synapse/config/voip.py2
-rw-r--r--synapse/handlers/message.py206
-rw-r--r--synapse/handlers/room.py39
-rw-r--r--synapse/python_dependencies.py23
-rw-r--r--synapse/rest/client/v1/initial_sync.py2
-rw-r--r--synapse/rest/client/v1/room.py20
-rw-r--r--synapse/state.py3
-rw-r--r--synapse/storage/__init__.py23
-rw-r--r--synapse/storage/events.py64
-rw-r--r--synapse/storage/registration.py13
-rw-r--r--synapse/storage/roommember.py6
-rw-r--r--synapse/storage/schema/delta/24/stats_reporting.sql22
-rw-r--r--synapse/storage/stream.py42
-rw-r--r--synapse/streams/config.py11
31 files changed, 523 insertions, 157 deletions
diff --git a/synapse/api/auth.py b/synapse/api/auth.py
index df788230fa..847ff60671 100644
--- a/synapse/api/auth.py
+++ b/synapse/api/auth.py
@@ -119,6 +119,20 @@ class Auth(object):
 
     @defer.inlineCallbacks
     def check_joined_room(self, room_id, user_id, current_state=None):
+        """Check if the user is currently joined in the room
+        Args:
+            room_id(str): The room to check.
+            user_id(str): The user to check.
+            current_state(dict): Optional map of the current state of the room.
+                If provided then that map is used to check whether they are a
+                member of the room. Otherwise the current membership is
+                loaded from the database.
+        Raises:
+            AuthError if the user is not in the room.
+        Returns:
+            A deferred membership event for the user if the user is in
+            the room.
+        """
         if current_state:
             member = current_state.get(
                 (EventTypes.Member, user_id),
@@ -135,6 +149,43 @@ class Auth(object):
         defer.returnValue(member)
 
     @defer.inlineCallbacks
+    def check_user_was_in_room(self, room_id, user_id, current_state=None):
+        """Check if the user was in the room at some point.
+        Args:
+            room_id(str): The room to check.
+            user_id(str): The user to check.
+            current_state(dict): Optional map of the current state of the room.
+                If provided then that map is used to check whether they are a
+                member of the room. Otherwise the current membership is
+                loaded from the database.
+        Raises:
+            AuthError if the user was never in the room.
+        Returns:
+            A deferred membership event for the user if the user was in the
+            room. This will be the join event if they are currently joined to
+            the room. This will be the leave event if they have left the room.
+        """
+        if current_state:
+            member = current_state.get(
+                (EventTypes.Member, user_id),
+                None
+            )
+        else:
+            member = yield self.state.get_current_state(
+                room_id=room_id,
+                event_type=EventTypes.Member,
+                state_key=user_id
+            )
+        membership = member.membership if member else None
+
+        if membership not in (Membership.JOIN, Membership.LEAVE):
+            raise AuthError(403, "User %s not in room %s" % (
+                user_id, room_id
+            ))
+
+        defer.returnValue(member)
+
+    @defer.inlineCallbacks
     def check_host_in_room(self, room_id, host):
         curr_state = yield self.state.get_current_state(room_id)
 
diff --git a/synapse/api/constants.py b/synapse/api/constants.py
index 1423986c1e..3385664394 100644
--- a/synapse/api/constants.py
+++ b/synapse/api/constants.py
@@ -27,16 +27,6 @@ class Membership(object):
     LIST = (INVITE, JOIN, KNOCK, LEAVE, BAN)
 
 
-class Feedback(object):
-
-    """Represents the types of feedback a user can send in response to a
-    message."""
-
-    DELIVERED = u"delivered"
-    READ = u"read"
-    LIST = (DELIVERED, READ)
-
-
 class PresenceState(object):
     """Represents the presence state of a user."""
     OFFLINE = u"offline"
@@ -73,7 +63,6 @@ class EventTypes(object):
     PowerLevels = "m.room.power_levels"
     Aliases = "m.room.aliases"
     Redaction = "m.room.redaction"
-    Feedback = "m.room.message.feedback"
 
     RoomHistoryVisibility = "m.room.history_visibility"
     CanonicalAlias = "m.room.canonical_alias"
diff --git a/synapse/app/homeserver.py b/synapse/app/homeserver.py
index 68d37e5bda..21840e4a28 100755
--- a/synapse/app/homeserver.py
+++ b/synapse/app/homeserver.py
@@ -16,10 +16,23 @@
 
 import sys
 sys.dont_write_bytecode = True
-from synapse.python_dependencies import check_requirements, DEPENDENCY_LINKS
+from synapse.python_dependencies import (
+    check_requirements, DEPENDENCY_LINKS, MissingRequirementError
+)
 
 if __name__ == '__main__':
-    check_requirements()
+    try:
+        check_requirements()
+    except MissingRequirementError as e:
+        message = "\n".join([
+            "Missing Requirement: %s" % (e.message,),
+            "To install run:",
+            "    pip install --upgrade --force \"%s\"" % (e.dependency,),
+            "",
+        ])
+        sys.stderr.writelines(message)
+        sys.exit(1)
+
 
 from synapse.storage.engines import create_engine, IncorrectDatabaseSetup
 from synapse.storage import (
@@ -29,7 +42,7 @@ from synapse.storage import (
 from synapse.server import HomeServer
 
 
-from twisted.internet import reactor
+from twisted.internet import reactor, task, defer
 from twisted.application import service
 from twisted.enterprise import adbapi
 from twisted.web.resource import Resource, EncodingResourceWrapper
@@ -664,6 +677,42 @@ def run(hs):
         ThreadPool._worker = profile(ThreadPool._worker)
         reactor.run = profile(reactor.run)
 
+    start_time = hs.get_clock().time()
+
+    @defer.inlineCallbacks
+    def phone_stats_home():
+        now = int(hs.get_clock().time())
+        uptime = int(now - start_time)
+        if uptime < 0:
+            uptime = 0
+
+        stats = {}
+        stats["homeserver"] = hs.config.server_name
+        stats["timestamp"] = now
+        stats["uptime_seconds"] = uptime
+        stats["total_users"] = yield hs.get_datastore().count_all_users()
+
+        all_rooms = yield hs.get_datastore().get_rooms(False)
+        stats["total_room_count"] = len(all_rooms)
+
+        stats["daily_active_users"] = yield hs.get_datastore().count_daily_users()
+        daily_messages = yield hs.get_datastore().count_daily_messages()
+        if daily_messages is not None:
+            stats["daily_messages"] = daily_messages
+
+        logger.info("Reporting stats to matrix.org: %s" % (stats,))
+        try:
+            yield hs.get_simple_http_client().put_json(
+                "https://matrix.org/report-usage-stats/push",
+                stats
+            )
+        except Exception as e:
+            logger.warn("Error reporting stats: %s", e)
+
+    if hs.config.report_stats:
+        phone_home_task = task.LoopingCall(phone_stats_home)
+        phone_home_task.start(60 * 60 * 24, now=False)
+
     def in_thread():
         with LoggingContext("run"):
             change_resource_limit(hs.config.soft_file_limit)
diff --git a/synapse/app/synctl.py b/synapse/app/synctl.py
index 1f7d543c31..6bcc437591 100755
--- a/synapse/app/synctl.py
+++ b/synapse/app/synctl.py
@@ -25,6 +25,7 @@ SYNAPSE = ["python", "-B", "-m", "synapse.app.homeserver"]
 CONFIGFILE = "homeserver.yaml"
 
 GREEN = "\x1b[1;32m"
+RED = "\x1b[1;31m"
 NORMAL = "\x1b[m"
 
 if not os.path.exists(CONFIGFILE):
@@ -45,8 +46,15 @@ def start():
     print "Starting ...",
     args = SYNAPSE
     args.extend(["--daemonize", "-c", CONFIGFILE])
-    subprocess.check_call(args)
-    print GREEN + "started" + NORMAL
+    try:
+        subprocess.check_call(args)
+        print GREEN + "started" + NORMAL
+    except subprocess.CalledProcessError as e:
+        print (
+            RED +
+            "error starting (exit code: %d); see above for logs" % e.returncode +
+            NORMAL
+        )
 
 
 def stop():
diff --git a/synapse/config/_base.py b/synapse/config/_base.py
index 8a75c48733..ceef309afc 100644
--- a/synapse/config/_base.py
+++ b/synapse/config/_base.py
@@ -26,6 +26,16 @@ class ConfigError(Exception):
 
 class Config(object):
 
+    stats_reporting_begging_spiel = (
+        "We would really appreciate it if you could help our project out by"
+        " reporting anonymized usage statistics from your homeserver. Only very"
+        " basic aggregate data (e.g. number of users) will be reported, but it"
+        " helps us to track the growth of the Matrix community, and helps us to"
+        " make Matrix a success, as well as to convince other networks that they"
+        " should peer with us."
+        "\nThank you."
+    )
+
     @staticmethod
     def parse_size(value):
         if isinstance(value, int) or isinstance(value, long):
@@ -111,11 +121,14 @@ class Config(object):
                 results.append(getattr(cls, name)(self, *args, **kargs))
         return results
 
-    def generate_config(self, config_dir_path, server_name):
+    def generate_config(self, config_dir_path, server_name, report_stats=None):
         default_config = "# vim:ft=yaml\n"
 
         default_config += "\n\n".join(dedent(conf) for conf in self.invoke_all(
-            "default_config", config_dir_path, server_name
+            "default_config",
+            config_dir_path=config_dir_path,
+            server_name=server_name,
+            report_stats=report_stats,
         ))
 
         config = yaml.load(default_config)
@@ -140,6 +153,12 @@ class Config(object):
             help="Generate a config file for the server name"
         )
         config_parser.add_argument(
+            "--report-stats",
+            action="store",
+            help="Stuff",
+            choices=["yes", "no"]
+        )
+        config_parser.add_argument(
             "--generate-keys",
             action="store_true",
             help="Generate any missing key files then exit"
@@ -189,6 +208,11 @@ class Config(object):
                     config_files.append(config_path)
 
         if config_args.generate_config:
+            if config_args.report_stats is None:
+                config_parser.error(
+                    "Please specify either --report-stats=yes or --report-stats=no\n\n" +
+                    cls.stats_reporting_begging_spiel
+                )
             if not config_files:
                 config_parser.error(
                     "Must supply a config file.\nA config file can be automatically"
@@ -211,7 +235,9 @@ class Config(object):
                     os.makedirs(config_dir_path)
                 with open(config_path, "wb") as config_file:
                     config_bytes, config = obj.generate_config(
-                        config_dir_path, server_name
+                        config_dir_path=config_dir_path,
+                        server_name=server_name,
+                        report_stats=(config_args.report_stats == "yes"),
                     )
                     obj.invoke_all("generate_files", config)
                     config_file.write(config_bytes)
@@ -261,9 +287,20 @@ class Config(object):
             specified_config.update(yaml_config)
 
         server_name = specified_config["server_name"]
-        _, config = obj.generate_config(config_dir_path, server_name)
+        _, config = obj.generate_config(
+            config_dir_path=config_dir_path,
+            server_name=server_name
+        )
         config.pop("log_config")
         config.update(specified_config)
+        if "report_stats" not in config:
+            sys.stderr.write(
+                "Please opt in or out of reporting anonymized homeserver usage "
+                "statistics, by setting the report_stats key in your config file "
+                " ( " + config_path + " ) " +
+                "to either True or False.\n\n" +
+                Config.stats_reporting_begging_spiel + "\n")
+            sys.exit(1)
 
         if generate_keys:
             obj.invoke_all("generate_files", config)
diff --git a/synapse/config/appservice.py b/synapse/config/appservice.py
index 38f41933b7..b8d301995e 100644
--- a/synapse/config/appservice.py
+++ b/synapse/config/appservice.py
@@ -20,7 +20,7 @@ class AppServiceConfig(Config):
     def read_config(self, config):
         self.app_service_config_files = config.get("app_service_config_files", [])
 
-    def default_config(cls, config_dir_path, server_name):
+    def default_config(cls, **kwargs):
         return """\
         # A list of application service config file to use
         app_service_config_files: []
diff --git a/synapse/config/captcha.py b/synapse/config/captcha.py
index 15a132b4e3..dd92fcd0dc 100644
--- a/synapse/config/captcha.py
+++ b/synapse/config/captcha.py
@@ -24,7 +24,7 @@ class CaptchaConfig(Config):
         self.captcha_bypass_secret = config.get("captcha_bypass_secret")
         self.recaptcha_siteverify_api = config["recaptcha_siteverify_api"]
 
-    def default_config(self, config_dir_path, server_name):
+    def default_config(self, **kwargs):
         return """\
         ## Captcha ##
 
diff --git a/synapse/config/database.py b/synapse/config/database.py
index f0611e8884..baeda8f300 100644
--- a/synapse/config/database.py
+++ b/synapse/config/database.py
@@ -45,7 +45,7 @@ class DatabaseConfig(Config):
 
         self.set_databasepath(config.get("database_path"))
 
-    def default_config(self, config, config_dir_path):
+    def default_config(self, **kwargs):
         database_path = self.abspath("homeserver.db")
         return """\
         # Database configuration
diff --git a/synapse/config/key.py b/synapse/config/key.py
index 23ac8a3fca..2c187065e5 100644
--- a/synapse/config/key.py
+++ b/synapse/config/key.py
@@ -40,7 +40,7 @@ class KeyConfig(Config):
             config["perspectives"]
         )
 
-    def default_config(self, config_dir_path, server_name):
+    def default_config(self, config_dir_path, server_name, **kwargs):
         base_key_name = os.path.join(config_dir_path, server_name)
         return """\
         ## Signing Keys ##
diff --git a/synapse/config/logger.py b/synapse/config/logger.py
index daca698d0c..bd0c17c861 100644
--- a/synapse/config/logger.py
+++ b/synapse/config/logger.py
@@ -70,7 +70,7 @@ class LoggingConfig(Config):
         self.log_config = self.abspath(config.get("log_config"))
         self.log_file = self.abspath(config.get("log_file"))
 
-    def default_config(self, config_dir_path, server_name):
+    def default_config(self, config_dir_path, server_name, **kwargs):
         log_file = self.abspath("homeserver.log")
         log_config = self.abspath(
             os.path.join(config_dir_path, server_name + ".log.config")
diff --git a/synapse/config/metrics.py b/synapse/config/metrics.py
index ae5a691527..825fec9a38 100644
--- a/synapse/config/metrics.py
+++ b/synapse/config/metrics.py
@@ -19,13 +19,15 @@ from ._base import Config
 class MetricsConfig(Config):
     def read_config(self, config):
         self.enable_metrics = config["enable_metrics"]
+        self.report_stats = config.get("report_stats", None)
         self.metrics_port = config.get("metrics_port")
         self.metrics_bind_host = config.get("metrics_bind_host", "127.0.0.1")
 
-    def default_config(self, config_dir_path, server_name):
-        return """\
+    def default_config(self, report_stats=None, **kwargs):
+        suffix = "" if report_stats is None else "report_stats: %(report_stats)s\n"
+        return ("""\
         ## Metrics ###
 
         # Enable collection and rendering of performance metrics
         enable_metrics: False
-        """
+        """ + suffix) % locals()
diff --git a/synapse/config/ratelimiting.py b/synapse/config/ratelimiting.py
index 76d9970e5b..611b598ec7 100644
--- a/synapse/config/ratelimiting.py
+++ b/synapse/config/ratelimiting.py
@@ -27,7 +27,7 @@ class RatelimitConfig(Config):
         self.federation_rc_reject_limit = config["federation_rc_reject_limit"]
         self.federation_rc_concurrent = config["federation_rc_concurrent"]
 
-    def default_config(self, config_dir_path, server_name):
+    def default_config(self, **kwargs):
         return """\
         ## Ratelimiting ##
 
diff --git a/synapse/config/registration.py b/synapse/config/registration.py
index 62de4b399f..fa98eced34 100644
--- a/synapse/config/registration.py
+++ b/synapse/config/registration.py
@@ -34,7 +34,7 @@ class RegistrationConfig(Config):
         self.registration_shared_secret = config.get("registration_shared_secret")
         self.macaroon_secret_key = config.get("macaroon_secret_key")
 
-    def default_config(self, config_dir, server_name):
+    def default_config(self, **kwargs):
         registration_shared_secret = random_string_with_symbols(50)
         macaroon_secret_key = random_string_with_symbols(50)
         return """\
diff --git a/synapse/config/repository.py b/synapse/config/repository.py
index 64644b9a7a..2fcf872449 100644
--- a/synapse/config/repository.py
+++ b/synapse/config/repository.py
@@ -60,7 +60,7 @@ class ContentRepositoryConfig(Config):
             config["thumbnail_sizes"]
         )
 
-    def default_config(self, config_dir_path, server_name):
+    def default_config(self, **kwargs):
         media_store = self.default_path("media_store")
         uploads_path = self.default_path("uploads")
         return """
diff --git a/synapse/config/saml2.py b/synapse/config/saml2.py
index 1532036876..4c6133cf22 100644
--- a/synapse/config/saml2.py
+++ b/synapse/config/saml2.py
@@ -41,7 +41,7 @@ class SAML2Config(Config):
             self.saml2_config_path = None
             self.saml2_idp_redirect_url = None
 
-    def default_config(self, config_dir_path, server_name):
+    def default_config(self, config_dir_path, server_name, **kwargs):
         return """
         # Enable SAML2 for registration and login. Uses pysaml2
         # config_path:      Path to the sp_conf.py configuration file
diff --git a/synapse/config/server.py b/synapse/config/server.py
index a03e55c223..4d12d49857 100644
--- a/synapse/config/server.py
+++ b/synapse/config/server.py
@@ -117,7 +117,7 @@ class ServerConfig(Config):
 
         self.content_addr = content_addr
 
-    def default_config(self, config_dir_path, server_name):
+    def default_config(self, server_name, **kwargs):
         if ":" in server_name:
             bind_port = int(server_name.split(":")[1])
             unsecure_port = bind_port - 400
diff --git a/synapse/config/tls.py b/synapse/config/tls.py
index e6023a718d..0ac2698293 100644
--- a/synapse/config/tls.py
+++ b/synapse/config/tls.py
@@ -50,7 +50,7 @@ class TlsConfig(Config):
             "use_insecure_ssl_client_just_for_testing_do_not_use"
         )
 
-    def default_config(self, config_dir_path, server_name):
+    def default_config(self, config_dir_path, server_name, **kwargs):
         base_key_name = os.path.join(config_dir_path, server_name)
 
         tls_certificate_path = base_key_name + ".tls.crt"
diff --git a/synapse/config/voip.py b/synapse/config/voip.py
index a1707223d3..a093354ccd 100644
--- a/synapse/config/voip.py
+++ b/synapse/config/voip.py
@@ -22,7 +22,7 @@ class VoipConfig(Config):
         self.turn_shared_secret = config["turn_shared_secret"]
         self.turn_user_lifetime = self.parse_duration(config["turn_user_lifetime"])
 
-    def default_config(self, config_dir_path, server_name):
+    def default_config(self, **kwargs):
         return """\
         ## Turn ##
 
diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index 23b779ad7c..bda8eb5f3f 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -16,13 +16,13 @@
 from twisted.internet import defer
 
 from synapse.api.constants import EventTypes, Membership
-from synapse.api.errors import RoomError, SynapseError
+from synapse.api.errors import SynapseError
 from synapse.streams.config import PaginationConfig
 from synapse.events.utils import serialize_event
 from synapse.events.validator import EventValidator
 from synapse.util import unwrapFirstError
 from synapse.util.logcontext import PreserveLoggingContext
-from synapse.types import UserID, RoomStreamToken
+from synapse.types import UserID, RoomStreamToken, StreamToken
 
 from ._base import BaseHandler
 
@@ -71,7 +71,7 @@ class MessageHandler(BaseHandler):
 
     @defer.inlineCallbacks
     def get_messages(self, user_id=None, room_id=None, pagin_config=None,
-                     feedback=False, as_client_event=True):
+                     as_client_event=True):
         """Get messages in a room.
 
         Args:
@@ -79,26 +79,52 @@ class MessageHandler(BaseHandler):
             room_id (str): The room they want messages from.
             pagin_config (synapse.api.streams.PaginationConfig): The pagination
             config rules to apply, if any.
-            feedback (bool): True to get compressed feedback with the messages
             as_client_event (bool): True to get events in client-server format.
         Returns:
             dict: Pagination API results
         """
-        yield self.auth.check_joined_room(room_id, user_id)
+        member_event = yield self.auth.check_user_was_in_room(room_id, user_id)
 
         data_source = self.hs.get_event_sources().sources["room"]
 
-        if not pagin_config.from_token:
+        if pagin_config.from_token:
+            room_token = pagin_config.from_token.room_key
+        else:
             pagin_config.from_token = (
                 yield self.hs.get_event_sources().get_current_token(
                     direction='b'
                 )
             )
+            room_token = pagin_config.from_token.room_key
 
-        room_token = RoomStreamToken.parse(pagin_config.from_token.room_key)
+        room_token = RoomStreamToken.parse(room_token)
         if room_token.topological is None:
             raise SynapseError(400, "Invalid token")
 
+        pagin_config.from_token = pagin_config.from_token.copy_and_replace(
+            "room_key", str(room_token)
+        )
+
+        source_config = pagin_config.get_source_config("room")
+
+        if member_event.membership == Membership.LEAVE:
+            # If they have left the room then clamp the token to be before
+            # they left the room
+            leave_token = yield self.store.get_topological_token_for_event(
+                member_event.event_id
+            )
+            leave_token = RoomStreamToken.parse(leave_token)
+            if leave_token.topological < room_token.topological:
+                source_config.from_key = str(leave_token)
+
+            if source_config.direction == "f":
+                if source_config.to_key is None:
+                    source_config.to_key = str(leave_token)
+                else:
+                    to_token = RoomStreamToken.parse(source_config.to_key)
+                    if leave_token.topological < to_token.topological:
+                        source_config.to_key = str(leave_token)
+
         yield self.hs.get_handlers().federation_handler.maybe_backfill(
             room_id, room_token.topological
         )
@@ -106,7 +132,7 @@ class MessageHandler(BaseHandler):
         user = UserID.from_string(user_id)
 
         events, next_key = yield data_source.get_pagination_rows(
-            user, pagin_config.get_source_config("room"), room_id
+            user, source_config, room_id
         )
 
         next_token = pagin_config.from_token.copy_and_replace(
@@ -255,29 +281,26 @@ class MessageHandler(BaseHandler):
         Raises:
             SynapseError if something went wrong.
         """
-        have_joined = yield self.auth.check_joined_room(room_id, user_id)
-        if not have_joined:
-            raise RoomError(403, "User not in room.")
+        member_event = yield self.auth.check_user_was_in_room(room_id, user_id)
 
-        data = yield self.state_handler.get_current_state(
-            room_id, event_type, state_key
-        )
-        defer.returnValue(data)
-
-    @defer.inlineCallbacks
-    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(event_id)
+        if member_event.membership == Membership.JOIN:
+            data = yield self.state_handler.get_current_state(
+                room_id, event_type, state_key
+            )
+        elif member_event.membership == Membership.LEAVE:
+            key = (event_type, state_key)
+            room_state = yield self.store.get_state_for_events(
+                room_id, [member_event.event_id], [key]
+            )
+            data = room_state[member_event.event_id].get(key)
 
-        if fb:
-            defer.returnValue(fb)
-        defer.returnValue(None)
+        defer.returnValue(data)
 
     @defer.inlineCallbacks
     def get_state_events(self, user_id, room_id):
-        """Retrieve all state events for a given room.
+        """Retrieve all state events for a given room. If the user is
+        joined to the room then return the current state. If the user has
+        left the room return the state events from when they left.
 
         Args:
             user_id(str): The user requesting state events.
@@ -285,18 +308,23 @@ class MessageHandler(BaseHandler):
         Returns:
             A list of dicts representing state events. [{}, {}, {}]
         """
-        yield self.auth.check_joined_room(room_id, user_id)
+        member_event = yield self.auth.check_user_was_in_room(room_id, user_id)
+
+        if member_event.membership == Membership.JOIN:
+            room_state = yield self.state_handler.get_current_state(room_id)
+        elif member_event.membership == Membership.LEAVE:
+            room_state = yield self.store.get_state_for_events(
+                room_id, [member_event.event_id], None
+            )
+            room_state = room_state[member_event.event_id]
 
-        # TODO: This is duplicating logic from snapshot_all_rooms
-        current_state = yield self.state_handler.get_current_state(room_id)
         now = self.clock.time_msec()
         defer.returnValue(
-            [serialize_event(c, now) for c in current_state.values()]
+            [serialize_event(c, now) for c in room_state.values()]
         )
 
     @defer.inlineCallbacks
-    def snapshot_all_rooms(self, user_id=None, pagin_config=None,
-                           feedback=False, as_client_event=True):
+    def snapshot_all_rooms(self, user_id=None, pagin_config=None, as_client_event=True):
         """Retrieve a snapshot of all rooms the user is invited or has joined.
 
         This snapshot may include messages for all rooms where the user is
@@ -306,7 +334,6 @@ class MessageHandler(BaseHandler):
             user_id (str): The ID of the user making the request.
             pagin_config (synapse.api.streams.PaginationConfig): The pagination
             config used to determine how many messages *PER ROOM* to return.
-            feedback (bool): True to get feedback along with these messages.
             as_client_event (bool): True to get events in client-server format.
         Returns:
             A list of dicts with "room_id" and "membership" keys for all rooms
@@ -316,7 +343,9 @@ class MessageHandler(BaseHandler):
         """
         room_list = yield self.store.get_rooms_for_user_where_membership_is(
             user_id=user_id,
-            membership_list=[Membership.INVITE, Membership.JOIN]
+            membership_list=[
+                Membership.INVITE, Membership.JOIN, Membership.LEAVE
+            ]
         )
 
         user = UserID.from_string(user_id)
@@ -358,19 +387,32 @@ class MessageHandler(BaseHandler):
 
             rooms_ret.append(d)
 
-            if event.membership != Membership.JOIN:
+            if event.membership not in (Membership.JOIN, Membership.LEAVE):
                 return
+
             try:
+                if event.membership == Membership.JOIN:
+                    room_end_token = now_token.room_key
+                    deferred_room_state = self.state_handler.get_current_state(
+                        event.room_id
+                    )
+                elif event.membership == Membership.LEAVE:
+                    room_end_token = "s%d" % (event.stream_ordering,)
+                    deferred_room_state = self.store.get_state_for_events(
+                        event.room_id, [event.event_id], None
+                    )
+                    deferred_room_state.addCallback(
+                        lambda states: states[event.event_id]
+                    )
+
                 (messages, token), current_state = yield defer.gatherResults(
                     [
                         self.store.get_recent_events_for_room(
                             event.room_id,
                             limit=limit,
-                            end_token=now_token.room_key,
-                        ),
-                        self.state_handler.get_current_state(
-                            event.room_id
+                            end_token=room_end_token,
                         ),
+                        deferred_room_state,
                     ]
                 ).addErrback(unwrapFirstError)
 
@@ -417,15 +459,85 @@ class MessageHandler(BaseHandler):
         defer.returnValue(ret)
 
     @defer.inlineCallbacks
-    def room_initial_sync(self, user_id, room_id, pagin_config=None,
-                          feedback=False):
-        current_state = yield self.state.get_current_state(
-            room_id=room_id,
+    def room_initial_sync(self, user_id, room_id, pagin_config=None):
+        """Capture the a snapshot of a room. If user is currently a member of
+        the room this will be what is currently in the room. If the user left
+        the room this will be what was in the room when they left.
+
+        Args:
+            user_id(str): The user to get a snapshot for.
+            room_id(str): The room to get a snapshot of.
+            pagin_config(synapse.streams.config.PaginationConfig):
+                The pagination config used to determine how many messages to
+                return.
+        Raises:
+            AuthError if the user wasn't in the room.
+        Returns:
+            A JSON serialisable dict with the snapshot of the room.
+        """
+
+        member_event = yield self.auth.check_user_was_in_room(room_id, user_id)
+
+        if member_event.membership == Membership.JOIN:
+            result = yield self._room_initial_sync_joined(
+                user_id, room_id, pagin_config, member_event
+            )
+        elif member_event.membership == Membership.LEAVE:
+            result = yield self._room_initial_sync_parted(
+                user_id, room_id, pagin_config, member_event
+            )
+        defer.returnValue(result)
+
+    @defer.inlineCallbacks
+    def _room_initial_sync_parted(self, user_id, room_id, pagin_config,
+                                  member_event):
+        room_state = yield self.store.get_state_for_events(
+            member_event.room_id, [member_event.event_id], None
+        )
+
+        room_state = room_state[member_event.event_id]
+
+        limit = pagin_config.limit if pagin_config else None
+        if limit is None:
+            limit = 10
+
+        stream_token = yield self.store.get_stream_token_for_event(
+            member_event.event_id
+        )
+
+        messages, token = yield self.store.get_recent_events_for_room(
+            room_id,
+            limit=limit,
+            end_token=stream_token
+        )
+
+        messages = yield self._filter_events_for_client(
+            user_id, room_id, messages
         )
 
-        yield self.auth.check_joined_room(
-            room_id, user_id,
-            current_state=current_state
+        start_token = StreamToken(token[0], 0, 0, 0)
+        end_token = StreamToken(token[1], 0, 0, 0)
+
+        time_now = self.clock.time_msec()
+
+        defer.returnValue({
+            "membership": member_event.membership,
+            "room_id": room_id,
+            "messages": {
+                "chunk": [serialize_event(m, time_now) for m in messages],
+                "start": start_token.to_string(),
+                "end": end_token.to_string(),
+            },
+            "state": [serialize_event(s, time_now) for s in room_state.values()],
+            "presence": [],
+            "receipts": [],
+        })
+
+    @defer.inlineCallbacks
+    def _room_initial_sync_joined(self, user_id, room_id, pagin_config,
+                                  member_event):
+        current_state = yield self.state.get_current_state(
+            room_id=room_id,
         )
 
         # TODO(paul): I wish I was called with user objects not user_id
@@ -439,8 +551,6 @@ class MessageHandler(BaseHandler):
             for x in current_state.values()
         ]
 
-        member_event = current_state.get((EventTypes.Member, user_id,))
-
         now_token = yield self.hs.get_event_sources().get_current_token()
 
         limit = pagin_config.limit if pagin_config else None
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index c5d1001b50..bb5eef6bbd 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -25,7 +25,6 @@ from synapse.api.constants import (
 from synapse.api.errors import StoreError, SynapseError
 from synapse.util import stringutils, unwrapFirstError
 from synapse.util.async import run_on_reactor
-from synapse.events.utils import serialize_event
 
 from collections import OrderedDict
 import logging
@@ -39,7 +38,7 @@ class RoomCreationHandler(BaseHandler):
     PRESETS_DICT = {
         RoomCreationPreset.PRIVATE_CHAT: {
             "join_rules": JoinRules.INVITE,
-            "history_visibility": "invited",
+            "history_visibility": "shared",
             "original_invitees_have_ops": False,
         },
         RoomCreationPreset.PUBLIC_CHAT: {
@@ -343,41 +342,6 @@ class RoomMemberHandler(BaseHandler):
                     remotedomains.add(member.domain)
 
     @defer.inlineCallbacks
-    def get_room_members_as_pagination_chunk(self, room_id=None, user_id=None,
-                                             limit=0, start_tok=None,
-                                             end_tok=None):
-        """Retrieve a list of room members in the room.
-
-        Args:
-            room_id (str): The room to get the member list for.
-            user_id (str): The ID of the user making the request.
-            limit (int): The max number of members to return.
-            start_tok (str): Optional. The start token if known.
-            end_tok (str): Optional. The end token if known.
-        Returns:
-            dict: A Pagination streamable dict.
-        Raises:
-            SynapseError if something goes wrong.
-        """
-        yield self.auth.check_joined_room(room_id, user_id)
-
-        member_list = yield self.store.get_room_members(room_id=room_id)
-        time_now = self.clock.time_msec()
-        event_list = [
-            serialize_event(entry, time_now)
-            for entry in member_list
-        ]
-        chunk_data = {
-            "start": "START",  # FIXME (erikj): START is no longer valid
-            "end": "END",
-            "chunk": event_list
-        }
-        # TODO honor Pagination stream params
-        # TODO snapshot this list to return on subsequent requests when
-        # paginating
-        defer.returnValue(chunk_data)
-
-    @defer.inlineCallbacks
     def change_membership(self, event, context, do_auth=True):
         """ Change the membership status of a user in a room.
 
@@ -646,7 +610,6 @@ class RoomEventSource(object):
             to_key=config.to_key,
             direction=config.direction,
             limit=config.limit,
-            with_feedback=True
         )
 
         defer.returnValue((events, next_key))
diff --git a/synapse/python_dependencies.py b/synapse/python_dependencies.py
index 795ef27182..e95316720e 100644
--- a/synapse/python_dependencies.py
+++ b/synapse/python_dependencies.py
@@ -18,18 +18,18 @@ from distutils.version import LooseVersion
 logger = logging.getLogger(__name__)
 
 REQUIREMENTS = {
+    "frozendict>=0.4": ["frozendict"],
     "unpaddedbase64>=1.0.1": ["unpaddedbase64>=1.0.1"],
     "canonicaljson>=1.0.0": ["canonicaljson>=1.0.0"],
     "signedjson>=1.0.0": ["signedjson>=1.0.0"],
-    "Twisted>=15.1.0": ["twisted>=15.1.0"],
+    "pynacl>=0.3.0": ["nacl>=0.3.0", "nacl.bindings"],
     "service_identity>=1.0.0": ["service_identity>=1.0.0"],
+    "Twisted>=15.1.0": ["twisted>=15.1.0"],
     "pyopenssl>=0.14": ["OpenSSL>=0.14"],
     "pyyaml": ["yaml"],
     "pyasn1": ["pyasn1"],
-    "pynacl>=0.3.0": ["nacl>=0.3.0"],
     "daemonize": ["daemonize"],
     "py-bcrypt": ["bcrypt"],
-    "frozendict>=0.4": ["frozendict"],
     "pillow": ["PIL"],
     "pydenticon": ["pydenticon"],
     "ujson": ["ujson"],
@@ -60,7 +60,10 @@ DEPENDENCY_LINKS = {
 
 
 class MissingRequirementError(Exception):
-    pass
+    def __init__(self, message, module_name, dependency):
+        super(MissingRequirementError, self).__init__(message)
+        self.module_name = module_name
+        self.dependency = dependency
 
 
 def check_requirements(config=None):
@@ -88,7 +91,7 @@ def check_requirements(config=None):
                 )
                 raise MissingRequirementError(
                     "Can't import %r which is part of %r"
-                    % (module_name, dependency)
+                    % (module_name, dependency), module_name, dependency
                 )
             version = getattr(module, "__version__", None)
             file_path = getattr(module, "__file__", None)
@@ -101,23 +104,25 @@ def check_requirements(config=None):
                 if version is None:
                     raise MissingRequirementError(
                         "Version of %r isn't set as __version__ of module %r"
-                        % (dependency, module_name)
+                        % (dependency, module_name), module_name, dependency
                     )
                 if LooseVersion(version) < LooseVersion(required_version):
                     raise MissingRequirementError(
                         "Version of %r in %r is too old. %r < %r"
-                        % (dependency, file_path, version, required_version)
+                        % (dependency, file_path, version, required_version),
+                        module_name, dependency
                     )
             elif version_test == "==":
                 if version is None:
                     raise MissingRequirementError(
                         "Version of %r isn't set as __version__ of module %r"
-                        % (dependency, module_name)
+                        % (dependency, module_name), module_name, dependency
                     )
                 if LooseVersion(version) != LooseVersion(required_version):
                     raise MissingRequirementError(
                         "Unexpected version of %r in %r. %r != %r"
-                        % (dependency, file_path, version, required_version)
+                        % (dependency, file_path, version, required_version),
+                        module_name, dependency
                     )
 
 
diff --git a/synapse/rest/client/v1/initial_sync.py b/synapse/rest/client/v1/initial_sync.py
index 4ea4da653c..bac68cc29f 100644
--- a/synapse/rest/client/v1/initial_sync.py
+++ b/synapse/rest/client/v1/initial_sync.py
@@ -26,14 +26,12 @@ class InitialSyncRestServlet(ClientV1RestServlet):
     @defer.inlineCallbacks
     def on_GET(self, request):
         user, _ = yield self.auth.get_user_by_req(request)
-        with_feedback = "feedback" in request.args
         as_client_event = "raw" not in request.args
         pagination_config = PaginationConfig.from_request(request)
         handler = self.handlers.message_handler
         content = yield handler.snapshot_all_rooms(
             user_id=user.to_string(),
             pagin_config=pagination_config,
-            feedback=with_feedback,
             as_client_event=as_client_event
         )
 
diff --git a/synapse/rest/client/v1/room.py b/synapse/rest/client/v1/room.py
index c9c27dd5a0..23871f161e 100644
--- a/synapse/rest/client/v1/room.py
+++ b/synapse/rest/client/v1/room.py
@@ -290,12 +290,18 @@ class RoomMemberListRestServlet(ClientV1RestServlet):
     def on_GET(self, request, room_id):
         # TODO support Pagination stream API (limit/tokens)
         user, _ = yield self.auth.get_user_by_req(request)
-        handler = self.handlers.room_member_handler
-        members = yield handler.get_room_members_as_pagination_chunk(
+        handler = self.handlers.message_handler
+        events = yield handler.get_state_events(
             room_id=room_id,
-            user_id=user.to_string())
+            user_id=user.to_string(),
+        )
+
+        chunk = []
 
-        for event in members["chunk"]:
+        for event in events:
+            if event["type"] != EventTypes.Member:
+                continue
+            chunk.append(event)
             # FIXME: should probably be state_key here, not user_id
             target_user = UserID.from_string(event["user_id"])
             # Presence is an optional cache; don't fail if we can't fetch it
@@ -308,7 +314,9 @@ class RoomMemberListRestServlet(ClientV1RestServlet):
             except:
                 pass
 
-        defer.returnValue((200, members))
+        defer.returnValue((200, {
+            "chunk": chunk
+        }))
 
 
 # TODO: Needs unit testing
@@ -321,14 +329,12 @@ class RoomMessageListRestServlet(ClientV1RestServlet):
         pagination_config = PaginationConfig.from_request(
             request, default_limit=10,
         )
-        with_feedback = "feedback" in request.args
         as_client_event = "raw" not in request.args
         handler = self.handlers.message_handler
         msgs = yield handler.get_messages(
             room_id=room_id,
             user_id=user.to_string(),
             pagin_config=pagination_config,
-            feedback=with_feedback,
             as_client_event=as_client_event
         )
 
diff --git a/synapse/state.py b/synapse/state.py
index 1fe4d066bd..ed36f844cb 100644
--- a/synapse/state.py
+++ b/synapse/state.py
@@ -17,7 +17,6 @@
 from twisted.internet import defer
 
 from synapse.util.logutils import log_function
-from synapse.util.async import run_on_reactor
 from synapse.util.caches.expiringcache import ExpiringCache
 from synapse.api.constants import EventTypes
 from synapse.api.errors import AuthError
@@ -119,8 +118,6 @@ class StateHandler(object):
         Returns:
             an EventContext
         """
-        yield run_on_reactor()
-
         context = EventContext()
 
         if outlier:
diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py
index 77cb1dbd81..340e59afcb 100644
--- a/synapse/storage/__init__.py
+++ b/synapse/storage/__init__.py
@@ -54,7 +54,7 @@ logger = logging.getLogger(__name__)
 
 # Remember to update this number every time a change is made to database
 # schema files, so the users will be informed on server restarts.
-SCHEMA_VERSION = 23
+SCHEMA_VERSION = 24
 
 dir_path = os.path.abspath(os.path.dirname(__file__))
 
@@ -126,6 +126,27 @@ class DataStore(RoomMemberStore, RoomStore,
             lock=False,
         )
 
+    @defer.inlineCallbacks
+    def count_daily_users(self):
+        """
+        Counts the number of users who used this homeserver in the last 24 hours.
+        """
+        def _count_users(txn):
+            txn.execute(
+                "SELECT COUNT(DISTINCT user_id) AS users"
+                " FROM user_ips"
+                " WHERE last_seen > ?",
+                # This is close enough to a day for our purposes.
+                (int(self._clock.time_msec()) - (1000 * 60 * 60 * 24),)
+            )
+            rows = self.cursor_to_dict(txn)
+            if rows:
+                return rows[0]["users"]
+            return 0
+
+        ret = yield self.runInteraction("count_users", _count_users)
+        defer.returnValue(ret)
+
     def get_user_ip_and_agents(self, user):
         return self._simple_select_list(
             table="user_ips",
diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index 0a477e3122..46df6b4d6d 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.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.
-
 from _base import SQLBaseStore, _RollbackButIsFineException
 
 from twisted.internet import defer, reactor
@@ -28,6 +27,7 @@ from canonicaljson import encode_canonical_json
 from contextlib import contextmanager
 
 import logging
+import math
 import ujson as json
 
 logger = logging.getLogger(__name__)
@@ -905,3 +905,65 @@ class EventsStore(SQLBaseStore):
         txn.execute(sql, (event.event_id,))
         result = txn.fetchone()
         return result[0] if result else None
+
+    @defer.inlineCallbacks
+    def count_daily_messages(self):
+        """
+        Returns an estimate of the number of messages sent in the last day.
+
+        If it has been significantly less or more than one day since the last
+        call to this function, it will return None.
+        """
+        def _count_messages(txn):
+            now = self.hs.get_clock().time()
+
+            txn.execute(
+                "SELECT reported_stream_token, reported_time FROM stats_reporting"
+            )
+            last_reported = self.cursor_to_dict(txn)
+
+            txn.execute(
+                "SELECT stream_ordering"
+                " FROM events"
+                " ORDER BY stream_ordering DESC"
+                " LIMIT 1"
+            )
+            now_reporting = self.cursor_to_dict(txn)
+            if not now_reporting:
+                return None
+            now_reporting = now_reporting[0]["stream_ordering"]
+
+            txn.execute("DELETE FROM stats_reporting")
+            txn.execute(
+                "INSERT INTO stats_reporting"
+                " (reported_stream_token, reported_time)"
+                " VALUES (?, ?)",
+                (now_reporting, now,)
+            )
+
+            if not last_reported:
+                return None
+
+            # Close enough to correct for our purposes.
+            yesterday = (now - 24 * 60 * 60)
+            if math.fabs(yesterday - last_reported[0]["reported_time"]) > 60 * 60:
+                return None
+
+            txn.execute(
+                "SELECT COUNT(*) as messages"
+                " FROM events NATURAL JOIN event_json"
+                " WHERE json like '%m.room.message%'"
+                " AND stream_ordering > ?"
+                " AND stream_ordering <= ?",
+                (
+                    last_reported[0]["reported_stream_token"],
+                    now_reporting,
+                )
+            )
+            rows = self.cursor_to_dict(txn)
+            if not rows:
+                return None
+            return rows[0]["messages"]
+
+        ret = yield self.runInteraction("count_messages", _count_messages)
+        defer.returnValue(ret)
diff --git a/synapse/storage/registration.py b/synapse/storage/registration.py
index c9ceb132ae..b454dd5b3a 100644
--- a/synapse/storage/registration.py
+++ b/synapse/storage/registration.py
@@ -289,3 +289,16 @@ class RegistrationStore(SQLBaseStore):
         if ret:
             defer.returnValue(ret['user_id'])
         defer.returnValue(None)
+
+    @defer.inlineCallbacks
+    def count_all_users(self):
+        """Counts all users registered on the homeserver."""
+        def _count_users(txn):
+            txn.execute("SELECT COUNT(*) AS users FROM users")
+            rows = self.cursor_to_dict(txn)
+            if rows:
+                return rows[0]["users"]
+            return 0
+
+        ret = yield self.runInteraction("count_users", _count_users)
+        defer.returnValue(ret)
diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py
index 8eee2dfbcc..cd9eefbd9f 100644
--- a/synapse/storage/roommember.py
+++ b/synapse/storage/roommember.py
@@ -30,7 +30,7 @@ logger = logging.getLogger(__name__)
 
 RoomsForUser = namedtuple(
     "RoomsForUser",
-    ("room_id", "sender", "membership")
+    ("room_id", "sender", "membership", "event_id", "stream_ordering")
 )
 
 
@@ -141,9 +141,11 @@ class RoomMemberStore(SQLBaseStore):
         args.extend(membership_list)
 
         sql = (
-            "SELECT m.room_id, m.sender, m.membership"
+            "SELECT m.room_id, m.sender, m.membership, m.event_id, e.stream_ordering"
             " FROM room_memberships as m"
             " INNER JOIN current_state_events as c"
+            " ON e.event_id = c.event_id "
+            " INNER JOIN events as e "
             " ON m.event_id = c.event_id "
             " AND m.room_id = c.room_id "
             " AND m.user_id = c.state_key"
diff --git a/synapse/storage/schema/delta/24/stats_reporting.sql b/synapse/storage/schema/delta/24/stats_reporting.sql
new file mode 100644
index 0000000000..e9165d2917
--- /dev/null
+++ b/synapse/storage/schema/delta/24/stats_reporting.sql
@@ -0,0 +1,22 @@
+/* Copyright 2015 OpenMarket Ltd
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * 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.
+ */
+
+-- Should only ever contain one row
+CREATE TABLE IF NOT EXISTS stats_reporting(
+  -- The stream ordering token which was most recently reported as stats
+  reported_stream_token INTEGER,
+  -- The time (seconds since epoch) stats were most recently reported
+  reported_time BIGINT
+);
diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py
index d7fe423f5a..3cab06fdef 100644
--- a/synapse/storage/stream.py
+++ b/synapse/storage/stream.py
@@ -159,9 +159,7 @@ class StreamStore(SQLBaseStore):
 
     @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
-
+                               limit=0):
         current_room_membership_sql = (
             "SELECT m.room_id FROM room_memberships as m "
             " INNER JOIN current_state_events as c"
@@ -227,10 +225,7 @@ class StreamStore(SQLBaseStore):
 
     @defer.inlineCallbacks
     def paginate_room_events(self, room_id, from_key, to_key=None,
-                             direction='b', limit=-1,
-                             with_feedback=False):
-        # TODO (erikj): Handle compressed feedback
-
+                             direction='b', limit=-1):
         # Tokens really represent positions between elements, but we use
         # the convention of pointing to the event before the gap. Hence
         # we have a bit of asymmetry when it comes to equalities.
@@ -302,7 +297,6 @@ class StreamStore(SQLBaseStore):
 
     @cachedInlineCallbacks(num_args=4)
     def get_recent_events_for_room(self, room_id, limit, end_token, from_token=None):
-        # TODO (erikj): Handle compressed feedback
 
         end_token = RoomStreamToken.parse_stream_token(end_token)
 
@@ -379,6 +373,38 @@ class StreamStore(SQLBaseStore):
             )
             defer.returnValue("t%d-%d" % (topo, token))
 
+    def get_stream_token_for_event(self, event_id):
+        """The stream token for an event
+        Args:
+            event_id(str): The id of the event to look up a stream token for.
+        Raises:
+            StoreError if the event wasn't in the database.
+        Returns:
+            A deferred "s%d" stream token.
+        """
+        return self._simple_select_one_onecol(
+            table="events",
+            keyvalues={"event_id": event_id},
+            retcol="stream_ordering",
+        ).addCallback(lambda row: "s%d" % (row,))
+
+    def get_topological_token_for_event(self, event_id):
+        """The stream token for an event
+        Args:
+            event_id(str): The id of the event to look up a stream token for.
+        Raises:
+            StoreError if the event wasn't in the database.
+        Returns:
+            A deferred "t%d-%d" topological token.
+        """
+        return self._simple_select_one(
+            table="events",
+            keyvalues={"event_id": event_id},
+            retcols=("stream_ordering", "topological_ordering"),
+        ).addCallback(lambda row: "t%d-%d" % (
+            row["topological_ordering"], row["stream_ordering"],)
+        )
+
     def _get_max_topological_txn(self, txn):
         txn.execute(
             "SELECT MAX(topological_ordering) FROM events"
diff --git a/synapse/streams/config.py b/synapse/streams/config.py
index 2ec7c5403b..167bfe0de3 100644
--- a/synapse/streams/config.py
+++ b/synapse/streams/config.py
@@ -34,6 +34,11 @@ class SourcePaginationConfig(object):
         self.direction = 'f' if direction == 'f' else 'b'
         self.limit = int(limit) if limit is not None else None
 
+    def __repr__(self):
+        return (
+            "StreamConfig(from_key=%r, to_key=%r, direction=%r, limit=%r)"
+        ) % (self.from_key, self.to_key, self.direction, self.limit)
+
 
 class PaginationConfig(object):
 
@@ -94,10 +99,10 @@ class PaginationConfig(object):
             logger.exception("Failed to create pagination config")
             raise SynapseError(400, "Invalid request.")
 
-    def __str__(self):
+    def __repr__(self):
         return (
-            "<PaginationConfig from_tok=%s, to_tok=%s, "
-            "direction=%s, limit=%s>"
+            "PaginationConfig(from_tok=%r, to_tok=%r,"
+            " direction=%r, limit=%r)"
         ) % (self.from_token, self.to_token, self.direction, self.limit)
 
     def get_source_config(self, source_name):