summary refs log tree commit diff
path: root/synapse
diff options
context:
space:
mode:
Diffstat (limited to 'synapse')
-rw-r--r--synapse/api/auth.py19
-rw-r--r--synapse/api/filtering.py8
-rw-r--r--synapse/app/pusher.py45
-rw-r--r--synapse/config/emailconfig.py98
-rw-r--r--synapse/config/homeserver.py3
-rw-r--r--synapse/config/key.py5
-rw-r--r--synapse/config/registration.py6
-rw-r--r--synapse/config/server.py8
-rw-r--r--synapse/federation/federation_server.py5
-rw-r--r--synapse/federation/transaction_queue.py3
-rw-r--r--synapse/federation/transport/server.py49
-rw-r--r--synapse/handlers/_base.py368
-rw-r--r--synapse/handlers/auth.py9
-rw-r--r--synapse/handlers/federation.py31
-rw-r--r--synapse/handlers/message.py215
-rw-r--r--synapse/handlers/presence.py6
-rw-r--r--synapse/handlers/register.py53
-rw-r--r--synapse/handlers/room.py7
-rw-r--r--synapse/handlers/room_member.py4
-rw-r--r--synapse/handlers/search.py17
-rw-r--r--synapse/handlers/sync.py29
-rw-r--r--synapse/notifier.py5
-rw-r--r--synapse/push/action_generator.py4
-rw-r--r--synapse/push/bulk_push_rule_evaluator.py7
-rw-r--r--synapse/push/emailpusher.py251
-rw-r--r--synapse/push/mailer.py456
-rw-r--r--synapse/push/pusher.py33
-rw-r--r--synapse/push/pusherpool.py6
-rw-r--r--synapse/python_dependencies.py4
-rw-r--r--synapse/replication/resource.py9
-rw-r--r--synapse/replication/slave/storage/account_data.py61
-rw-r--r--synapse/replication/slave/storage/events.py27
-rw-r--r--synapse/replication/slave/storage/pushers.py4
-rw-r--r--synapse/replication/slave/storage/receipts.py2
-rw-r--r--synapse/rest/__init__.py4
-rw-r--r--synapse/rest/client/v1/register.py71
-rw-r--r--synapse/rest/client/v2_alpha/openid.py96
-rw-r--r--synapse/rest/client/v2_alpha/register.py29
-rw-r--r--synapse/rest/client/v2_alpha/report_event.py59
-rw-r--r--synapse/storage/__init__.py5
-rw-r--r--synapse/storage/_base.py8
-rw-r--r--synapse/storage/account_data.py50
-rw-r--r--synapse/storage/event_push_actions.py61
-rw-r--r--synapse/storage/events.py162
-rw-r--r--synapse/storage/openid.py32
-rw-r--r--synapse/storage/prepare_database.py2
-rw-r--r--synapse/storage/pusher.py40
-rw-r--r--synapse/storage/receipts.py36
-rw-r--r--synapse/storage/registration.py3
-rw-r--r--synapse/storage/room.py18
-rw-r--r--synapse/storage/roommember.py7
-rw-r--r--synapse/storage/schema/delta/32/events.sql16
-rw-r--r--synapse/storage/schema/delta/32/openid.sql9
-rw-r--r--synapse/storage/schema/delta/32/pusher_throttle.sql23
-rw-r--r--synapse/storage/schema/delta/32/remove_indices.sql38
-rw-r--r--synapse/storage/schema/delta/32/reports.sql25
-rw-r--r--synapse/storage/transactions.py165
-rw-r--r--synapse/types.py4
-rw-r--r--synapse/util/presentable_names.py159
-rw-r--r--synapse/visibility.py210
60 files changed, 2651 insertions, 538 deletions
diff --git a/synapse/api/auth.py b/synapse/api/auth.py
index 9e912fdfbe..44e38b777a 100644
--- a/synapse/api/auth.py
+++ b/synapse/api/auth.py
@@ -22,7 +22,7 @@ from twisted.internet import defer
 
 from synapse.api.constants import EventTypes, Membership, JoinRules
 from synapse.api.errors import AuthError, Codes, SynapseError, EventSizeError
-from synapse.types import Requester, RoomID, UserID, EventID
+from synapse.types import Requester, UserID, get_domian_from_id
 from synapse.util.logutils import log_function
 from synapse.util.logcontext import preserve_context_over_fn
 from synapse.util.metrics import Measure
@@ -91,8 +91,8 @@ class Auth(object):
                     "Room %r does not exist" % (event.room_id,)
                 )
 
-            creating_domain = RoomID.from_string(event.room_id).domain
-            originating_domain = UserID.from_string(event.sender).domain
+            creating_domain = get_domian_from_id(event.room_id)
+            originating_domain = get_domian_from_id(event.sender)
             if creating_domain != originating_domain:
                 if not self.can_federate(event, auth_events):
                     raise AuthError(
@@ -219,7 +219,7 @@ class Auth(object):
         for event in curr_state.values():
             if event.type == EventTypes.Member:
                 try:
-                    if UserID.from_string(event.state_key).domain != host:
+                    if get_domian_from_id(event.state_key) != host:
                         continue
                 except:
                     logger.warn("state_key not user_id: %s", event.state_key)
@@ -266,8 +266,8 @@ class Auth(object):
 
         target_user_id = event.state_key
 
-        creating_domain = RoomID.from_string(event.room_id).domain
-        target_domain = UserID.from_string(target_user_id).domain
+        creating_domain = get_domian_from_id(event.room_id)
+        target_domain = get_domian_from_id(target_user_id)
         if creating_domain != target_domain:
             if not self.can_federate(event, auth_events):
                 raise AuthError(
@@ -612,7 +612,8 @@ class Auth(object):
     def get_user_from_macaroon(self, macaroon_str):
         try:
             macaroon = pymacaroons.Macaroon.deserialize(macaroon_str)
-            self.validate_macaroon(macaroon, "access", False)
+
+            self.validate_macaroon(macaroon, "access", self.hs.config.expire_access_token)
 
             user_prefix = "user_id = "
             user = None
@@ -889,8 +890,8 @@ class Auth(object):
         if user_level >= redact_level:
             return False
 
-        redacter_domain = EventID.from_string(event.event_id).domain
-        redactee_domain = EventID.from_string(event.redacts).domain
+        redacter_domain = get_domian_from_id(event.event_id)
+        redactee_domain = get_domian_from_id(event.redacts)
         if redacter_domain == redactee_domain:
             return True
 
diff --git a/synapse/api/filtering.py b/synapse/api/filtering.py
index cd699ef27f..4f5a4281fa 100644
--- a/synapse/api/filtering.py
+++ b/synapse/api/filtering.py
@@ -15,6 +15,8 @@
 from synapse.api.errors import SynapseError
 from synapse.types import UserID, RoomID
 
+from twisted.internet import defer
+
 import ujson as json
 
 
@@ -24,10 +26,10 @@ class Filtering(object):
         super(Filtering, self).__init__()
         self.store = hs.get_datastore()
 
+    @defer.inlineCallbacks
     def get_user_filter(self, user_localpart, filter_id):
-        result = self.store.get_user_filter(user_localpart, filter_id)
-        result.addCallback(FilterCollection)
-        return result
+        result = yield self.store.get_user_filter(user_localpart, filter_id)
+        defer.returnValue(FilterCollection(result))
 
     def add_user_filter(self, user_localpart, user_filter):
         self.check_valid_filter(user_filter)
diff --git a/synapse/app/pusher.py b/synapse/app/pusher.py
index b5339f030d..135dd58c15 100644
--- a/synapse/app/pusher.py
+++ b/synapse/app/pusher.py
@@ -20,11 +20,14 @@ from synapse.server import HomeServer
 from synapse.config._base import ConfigError
 from synapse.config.database import DatabaseConfig
 from synapse.config.logger import LoggingConfig
+from synapse.config.emailconfig import EmailConfig
 from synapse.http.site import SynapseSite
 from synapse.metrics.resource import MetricsResource, METRICS_PREFIX
+from synapse.storage.roommember import RoomMemberStore
 from synapse.replication.slave.storage.events import SlavedEventStore
 from synapse.replication.slave.storage.pushers import SlavedPusherStore
 from synapse.replication.slave.storage.receipts import SlavedReceiptsStore
+from synapse.replication.slave.storage.account_data import SlavedAccountDataStore
 from synapse.storage.engines import create_engine
 from synapse.storage import DataStore
 from synapse.util.async import sleep
@@ -58,6 +61,7 @@ class SlaveConfig(DatabaseConfig):
         self.soft_file_limit = config.get("soft_file_limit")
         self.daemonize = config.get("daemonize")
         self.pid_file = self.abspath(config.get("pid_file"))
+        self.public_baseurl = config["public_baseurl"]
 
     def default_config(self, server_name, **kwargs):
         pid_file = self.abspath("pusher.pid")
@@ -91,12 +95,13 @@ class SlaveConfig(DatabaseConfig):
         """ % locals()
 
 
-class PusherSlaveConfig(SlaveConfig, LoggingConfig):
+class PusherSlaveConfig(SlaveConfig, LoggingConfig, EmailConfig):
     pass
 
 
 class PusherSlaveStore(
-    SlavedEventStore, SlavedPusherStore, SlavedReceiptsStore
+    SlavedEventStore, SlavedPusherStore, SlavedReceiptsStore,
+    SlavedAccountDataStore
 ):
     update_pusher_last_stream_ordering_and_success = (
         DataStore.update_pusher_last_stream_ordering_and_success.__func__
@@ -110,6 +115,31 @@ class PusherSlaveStore(
         DataStore.update_pusher_last_stream_ordering.__func__
     )
 
+    get_throttle_params_by_room = (
+        DataStore.get_throttle_params_by_room.__func__
+    )
+
+    set_throttle_params = (
+        DataStore.set_throttle_params.__func__
+    )
+
+    get_time_of_last_push_action_before = (
+        DataStore.get_time_of_last_push_action_before.__func__
+    )
+
+    get_profile_displayname = (
+        DataStore.get_profile_displayname.__func__
+    )
+
+    # XXX: This is a bit broken because we don't persist forgotten rooms
+    # in a way that they can be streamed. This means that we don't have a
+    # way to invalidate the forgotten rooms cache correctly.
+    # For now we expire the cache every 10 minutes.
+    BROKEN_CACHE_EXPIRY_MS = 60 * 60 * 1000
+    who_forgot_in_room = (
+        RoomMemberStore.__dict__["who_forgot_in_room"]
+    )
+
 
 class PusherServer(HomeServer):
 
@@ -189,6 +219,7 @@ class PusherServer(HomeServer):
         store = self.get_datastore()
         replication_url = self.config.replication_url
         pusher_pool = self.get_pusherpool()
+        clock = self.get_clock()
 
         def stop_pusher(user_id, app_id, pushkey):
             key = "%s:%s" % (app_id, pushkey)
@@ -240,11 +271,21 @@ class PusherServer(HomeServer):
                     min_stream_id, max_stream_id, affected_room_ids
                 )
 
+        def expire_broken_caches():
+            store.who_forgot_in_room.invalidate_all()
+
+        next_expire_broken_caches_ms = 0
         while True:
             try:
                 args = store.stream_positions()
                 args["timeout"] = 30000
                 result = yield http_client.get_json(replication_url, args=args)
+                now_ms = clock.time_msec()
+                if now_ms > next_expire_broken_caches_ms:
+                    expire_broken_caches()
+                    next_expire_broken_caches_ms = (
+                        now_ms + store.BROKEN_CACHE_EXPIRY_MS
+                    )
                 yield store.process_replication(result)
                 poke_pushers(result)
             except:
diff --git a/synapse/config/emailconfig.py b/synapse/config/emailconfig.py
new file mode 100644
index 0000000000..90bdd08f00
--- /dev/null
+++ b/synapse/config/emailconfig.py
@@ -0,0 +1,98 @@
+# -*- coding: utf-8 -*-
+# Copyright 2015, 2016 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.
+
+# This file can't be called email.py because if it is, we cannot:
+import email.utils
+
+from ._base import Config
+
+
+class EmailConfig(Config):
+    def read_config(self, config):
+        self.email_enable_notifs = False
+
+        email_config = config.get("email", {})
+        self.email_enable_notifs = email_config.get("enable_notifs", False)
+
+        if self.email_enable_notifs:
+            # make sure we can import the required deps
+            import jinja2
+            import bleach
+            # prevent unused warnings
+            jinja2
+            bleach
+
+            required = [
+                "smtp_host",
+                "smtp_port",
+                "notif_from",
+                "template_dir",
+                "notif_template_html",
+                "notif_template_text",
+            ]
+
+            missing = []
+            for k in required:
+                if k not in email_config:
+                    missing.append(k)
+
+            if (len(missing) > 0):
+                raise RuntimeError(
+                    "email.enable_notifs is True but required keys are missing: %s" %
+                    (", ".join(["email." + k for k in missing]),)
+                )
+
+            if config.get("public_baseurl") is None:
+                raise RuntimeError(
+                    "email.enable_notifs is True but no public_baseurl is set"
+                )
+
+            self.email_smtp_host = email_config["smtp_host"]
+            self.email_smtp_port = email_config["smtp_port"]
+            self.email_notif_from = email_config["notif_from"]
+            self.email_template_dir = email_config["template_dir"]
+            self.email_notif_template_html = email_config["notif_template_html"]
+            self.email_notif_template_text = email_config["notif_template_text"]
+            self.email_notif_for_new_users = email_config.get(
+                "notif_for_new_users", True
+            )
+            if "app_name" in email_config:
+                self.email_app_name = email_config["app_name"]
+            else:
+                self.email_app_name = "Matrix"
+
+            # make sure it's valid
+            parsed = email.utils.parseaddr(self.email_notif_from)
+            if parsed[1] == '':
+                raise RuntimeError("Invalid notif_from address")
+        else:
+            self.email_enable_notifs = False
+            # Not much point setting defaults for the rest: it would be an
+            # error for them to be used.
+
+    def default_config(self, config_dir_path, server_name, **kwargs):
+        return """
+        # Enable sending emails for notification events
+        #email:
+        #   enable_notifs: false
+        #   smtp_host: "localhost"
+        #   smtp_port: 25
+        #   notif_from: Your Friendly Matrix Home Server <noreply@example.com>
+        #   app_name: Matrix
+        #   template_dir: res/templates
+        #   notif_template_html: notif_mail.html
+        #   notif_template_text: notif_mail.txt
+        #   notif_for_new_users: True
+        """
diff --git a/synapse/config/homeserver.py b/synapse/config/homeserver.py
index 9a80ac39ec..fc2445484c 100644
--- a/synapse/config/homeserver.py
+++ b/synapse/config/homeserver.py
@@ -31,13 +31,14 @@ from .cas import CasConfig
 from .password import PasswordConfig
 from .jwt import JWTConfig
 from .ldap import LDAPConfig
+from .emailconfig import EmailConfig
 
 
 class HomeServerConfig(TlsConfig, ServerConfig, DatabaseConfig, LoggingConfig,
                        RatelimitConfig, ContentRepositoryConfig, CaptchaConfig,
                        VoipConfig, RegistrationConfig, MetricsConfig, ApiConfig,
                        AppServiceConfig, KeyConfig, SAML2Config, CasConfig,
-                       JWTConfig, LDAPConfig, PasswordConfig,):
+                       JWTConfig, LDAPConfig, PasswordConfig, EmailConfig,):
     pass
 
 
diff --git a/synapse/config/key.py b/synapse/config/key.py
index a072aec714..6ee643793e 100644
--- a/synapse/config/key.py
+++ b/synapse/config/key.py
@@ -57,6 +57,8 @@ class KeyConfig(Config):
             seed = self.signing_key[0].seed
             self.macaroon_secret_key = hashlib.sha256(seed)
 
+        self.expire_access_token = config.get("expire_access_token", False)
+
     def default_config(self, config_dir_path, server_name, is_generating_file=False,
                        **kwargs):
         base_key_name = os.path.join(config_dir_path, server_name)
@@ -69,6 +71,9 @@ class KeyConfig(Config):
         return """\
         macaroon_secret_key: "%(macaroon_secret_key)s"
 
+        # Used to enable access token expiration.
+        expire_access_token: False
+
         ## Signing Keys ##
 
         # Path to the signing key to sign messages with
diff --git a/synapse/config/registration.py b/synapse/config/registration.py
index 87e500c97a..cc3f879857 100644
--- a/synapse/config/registration.py
+++ b/synapse/config/registration.py
@@ -32,6 +32,7 @@ class RegistrationConfig(Config):
             )
 
         self.registration_shared_secret = config.get("registration_shared_secret")
+        self.user_creation_max_duration = int(config["user_creation_max_duration"])
 
         self.bcrypt_rounds = config.get("bcrypt_rounds", 12)
         self.trusted_third_party_id_servers = config["trusted_third_party_id_servers"]
@@ -54,6 +55,11 @@ class RegistrationConfig(Config):
         # secret, even if registration is otherwise disabled.
         registration_shared_secret: "%(registration_shared_secret)s"
 
+        # Sets the expiry for the short term user creation in
+        # milliseconds. For instance the bellow duration is two weeks
+        # in milliseconds.
+        user_creation_max_duration: 1209600000
+
         # Set the number of bcrypt rounds used to generate password hash.
         # Larger numbers increase the work factor needed to generate the hash.
         # The default number of rounds is 12.
diff --git a/synapse/config/server.py b/synapse/config/server.py
index 46c633548a..0b5f462e44 100644
--- a/synapse/config/server.py
+++ b/synapse/config/server.py
@@ -28,6 +28,11 @@ class ServerConfig(Config):
         self.print_pidfile = config.get("print_pidfile")
         self.user_agent_suffix = config.get("user_agent_suffix")
         self.use_frozen_dicts = config.get("use_frozen_dicts", True)
+        self.public_baseurl = config.get("public_baseurl")
+
+        if self.public_baseurl is not None:
+            if self.public_baseurl[-1] != '/':
+                self.public_baseurl += '/'
         self.start_pushers = config.get("start_pushers", True)
 
         self.listeners = config.get("listeners", [])
@@ -143,6 +148,9 @@ class ServerConfig(Config):
         # Whether to serve a web client from the HTTP/HTTPS root resource.
         web_client: True
 
+        # The public-facing base URL for the client API (not including _matrix/...)
+        # public_baseurl: https://example.com:8448/
+
         # Set the soft limit on the number of file descriptors synapse can use
         # Zero is used to indicate synapse should set the soft limit to the
         # hard limit.
diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py
index 429ab6ddec..f1d231b9d8 100644
--- a/synapse/federation/federation_server.py
+++ b/synapse/federation/federation_server.py
@@ -388,6 +388,11 @@ class FederationServer(FederationBase):
         })
 
     @log_function
+    def on_openid_userinfo(self, token):
+        ts_now_ms = self._clock.time_msec()
+        return self.store.get_user_id_for_open_id_token(token, ts_now_ms)
+
+    @log_function
     def _get_persisted_pdu(self, origin, event_id, do_auth=True):
         """ Get a PDU from the database with given origin and id.
 
diff --git a/synapse/federation/transaction_queue.py b/synapse/federation/transaction_queue.py
index 1928da03b3..5787f854d4 100644
--- a/synapse/federation/transaction_queue.py
+++ b/synapse/federation/transaction_queue.py
@@ -20,6 +20,7 @@ from .persistence import TransactionActions
 from .units import Transaction
 
 from synapse.api.errors import HttpResponseException
+from synapse.util.async import run_on_reactor
 from synapse.util.logutils import log_function
 from synapse.util.logcontext import PreserveLoggingContext
 from synapse.util.retryutils import (
@@ -199,6 +200,8 @@ class TransactionQueue(object):
     @defer.inlineCallbacks
     @log_function
     def _attempt_new_transaction(self, destination):
+        yield run_on_reactor()
+
         # list of (pending_pdu, deferred, order)
         if destination in self.pending_transactions:
             # XXX: pending_transactions can get stuck on by a never-ending
diff --git a/synapse/federation/transport/server.py b/synapse/federation/transport/server.py
index d65a7893d8..5b6c7d11dd 100644
--- a/synapse/federation/transport/server.py
+++ b/synapse/federation/transport/server.py
@@ -18,7 +18,7 @@ from twisted.internet import defer
 from synapse.api.urls import FEDERATION_PREFIX as PREFIX
 from synapse.api.errors import Codes, SynapseError
 from synapse.http.server import JsonResource
-from synapse.http.servlet import parse_json_object_from_request
+from synapse.http.servlet import parse_json_object_from_request, parse_string
 from synapse.util.ratelimitutils import FederationRateLimiter
 
 import functools
@@ -323,7 +323,7 @@ class FederationSendLeaveServlet(BaseFederationServlet):
 
 
 class FederationEventAuthServlet(BaseFederationServlet):
-    PATH = "/event_auth(?P<context>[^/]*)/(?P<event_id>[^/]*)"
+    PATH = "/event_auth/(?P<context>[^/]*)/(?P<event_id>[^/]*)"
 
     def on_GET(self, origin, content, query, context, event_id):
         return self.handler.on_event_auth(origin, context, event_id)
@@ -448,6 +448,50 @@ class On3pidBindServlet(BaseFederationServlet):
         return code
 
 
+class OpenIdUserInfo(BaseFederationServlet):
+    """
+    Exchange a bearer token for information about a user.
+
+    The response format should be compatible with:
+        http://openid.net/specs/openid-connect-core-1_0.html#UserInfoResponse
+
+    GET /openid/userinfo?access_token=ABDEFGH HTTP/1.1
+
+    HTTP/1.1 200 OK
+    Content-Type: application/json
+
+    {
+        "sub": "@userpart:example.org",
+    }
+    """
+
+    PATH = "/openid/userinfo"
+
+    @defer.inlineCallbacks
+    def on_GET(self, request):
+        token = parse_string(request, "access_token")
+        if token is None:
+            defer.returnValue((401, {
+                "errcode": "M_MISSING_TOKEN", "error": "Access Token required"
+            }))
+            return
+
+        user_id = yield self.handler.on_openid_userinfo(token)
+
+        if user_id is None:
+            defer.returnValue((401, {
+                "errcode": "M_UNKNOWN_TOKEN",
+                "error": "Access Token unknown or expired"
+            }))
+
+        defer.returnValue((200, {"sub": user_id}))
+
+    # Avoid doing remote HS authorization checks which are done by default by
+    # BaseFederationServlet.
+    def _wrap(self, code):
+        return code
+
+
 SERVLET_CLASSES = (
     FederationSendServlet,
     FederationPullServlet,
@@ -468,6 +512,7 @@ SERVLET_CLASSES = (
     FederationClientKeysClaimServlet,
     FederationThirdPartyInviteExchangeServlet,
     On3pidBindServlet,
+    OpenIdUserInfo,
 )
 
 
diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py
index 13a675b208..c904c6c500 100644
--- a/synapse/handlers/_base.py
+++ b/synapse/handlers/_base.py
@@ -15,13 +15,10 @@
 
 from twisted.internet import defer
 
-from synapse.api.errors import LimitExceededError, SynapseError, AuthError
-from synapse.crypto.event_signing import add_hashes_and_signatures
+from synapse.api.errors import LimitExceededError
 from synapse.api.constants import Membership, EventTypes
-from synapse.types import UserID, RoomAlias, Requester
-from synapse.push.action_generator import ActionGenerator
+from synapse.types import UserID, Requester
 
-from synapse.util.logcontext import PreserveLoggingContext, preserve_fn
 
 import logging
 
@@ -29,23 +26,6 @@ import logging
 logger = logging.getLogger(__name__)
 
 
-VISIBILITY_PRIORITY = (
-    "world_readable",
-    "shared",
-    "invited",
-    "joined",
-)
-
-
-MEMBERSHIP_PRIORITY = (
-    Membership.JOIN,
-    Membership.INVITE,
-    Membership.KNOCK,
-    Membership.LEAVE,
-    Membership.BAN,
-)
-
-
 class BaseHandler(object):
     """
     Common base class for the event handlers.
@@ -65,161 +45,10 @@ class BaseHandler(object):
         self.clock = hs.get_clock()
         self.hs = hs
 
-        self.signing_key = hs.config.signing_key[0]
         self.server_name = hs.hostname
 
         self.event_builder_factory = hs.get_event_builder_factory()
 
-    @defer.inlineCallbacks
-    def filter_events_for_clients(self, user_tuples, events, event_id_to_state):
-        """ Returns dict of user_id -> list of events that user is allowed to
-        see.
-
-        Args:
-            user_tuples (str, bool): (user id, is_peeking) for each user to be
-                checked. is_peeking should be true if:
-                * the user is not currently a member of the room, and:
-                * the user has not been a member of the room since the
-                given events
-            events ([synapse.events.EventBase]): list of events to filter
-        """
-        forgotten = yield defer.gatherResults([
-            self.store.who_forgot_in_room(
-                room_id,
-            )
-            for room_id in frozenset(e.room_id for e in events)
-        ], consumeErrors=True)
-
-        # Set of membership event_ids that have been forgotten
-        event_id_forgotten = frozenset(
-            row["event_id"] for rows in forgotten for row in rows
-        )
-
-        def allowed(event, user_id, is_peeking):
-            """
-            Args:
-                event (synapse.events.EventBase): event to check
-                user_id (str)
-                is_peeking (bool)
-            """
-            state = event_id_to_state[event.event_id]
-
-            # get the room_visibility at the time of the event.
-            visibility_event = state.get((EventTypes.RoomHistoryVisibility, ""), None)
-            if visibility_event:
-                visibility = visibility_event.content.get("history_visibility", "shared")
-            else:
-                visibility = "shared"
-
-            if visibility not in VISIBILITY_PRIORITY:
-                visibility = "shared"
-
-            # if it was world_readable, it's easy: everyone can read it
-            if visibility == "world_readable":
-                return True
-
-            # Always allow history visibility events on boundaries. This is done
-            # by setting the effective visibility to the least restrictive
-            # of the old vs new.
-            if event.type == EventTypes.RoomHistoryVisibility:
-                prev_content = event.unsigned.get("prev_content", {})
-                prev_visibility = prev_content.get("history_visibility", None)
-
-                if prev_visibility not in VISIBILITY_PRIORITY:
-                    prev_visibility = "shared"
-
-                new_priority = VISIBILITY_PRIORITY.index(visibility)
-                old_priority = VISIBILITY_PRIORITY.index(prev_visibility)
-                if old_priority < new_priority:
-                    visibility = prev_visibility
-
-            # likewise, if the event is the user's own membership event, use
-            # the 'most joined' membership
-            membership = None
-            if event.type == EventTypes.Member and event.state_key == user_id:
-                membership = event.content.get("membership", None)
-                if membership not in MEMBERSHIP_PRIORITY:
-                    membership = "leave"
-
-                prev_content = event.unsigned.get("prev_content", {})
-                prev_membership = prev_content.get("membership", None)
-                if prev_membership not in MEMBERSHIP_PRIORITY:
-                    prev_membership = "leave"
-
-                new_priority = MEMBERSHIP_PRIORITY.index(membership)
-                old_priority = MEMBERSHIP_PRIORITY.index(prev_membership)
-                if old_priority < new_priority:
-                    membership = prev_membership
-
-            # otherwise, get the user's membership at the time of the event.
-            if membership is None:
-                membership_event = state.get((EventTypes.Member, user_id), None)
-                if membership_event:
-                    if membership_event.event_id not in event_id_forgotten:
-                        membership = membership_event.membership
-
-            # if the user was a member of the room at the time of the event,
-            # they can see it.
-            if membership == Membership.JOIN:
-                return True
-
-            if visibility == "joined":
-                # we weren't a member at the time of the event, so we can't
-                # see this event.
-                return False
-
-            elif visibility == "invited":
-                # user can also see the event if they were *invited* at the time
-                # of the event.
-                return membership == Membership.INVITE
-
-            else:
-                # visibility is shared: user can also see the event if they have
-                # become a member since the event
-                #
-                # XXX: if the user has subsequently joined and then left again,
-                # ideally we would share history up to the point they left. But
-                # we don't know when they left.
-                return not is_peeking
-
-        defer.returnValue({
-            user_id: [
-                event
-                for event in events
-                if allowed(event, user_id, is_peeking)
-            ]
-            for user_id, is_peeking in user_tuples
-        })
-
-    @defer.inlineCallbacks
-    def _filter_events_for_client(self, user_id, events, is_peeking=False):
-        """
-        Check which events a user is allowed to see
-
-        Args:
-            user_id(str): user id to be checked
-            events([synapse.events.EventBase]): list of events to be checked
-            is_peeking(bool): should be True if:
-              * the user is not currently a member of the room, and:
-              * the user has not been a member of the room since the given
-                events
-
-        Returns:
-            [synapse.events.EventBase]
-        """
-        types = (
-            (EventTypes.RoomHistoryVisibility, ""),
-            (EventTypes.Member, user_id),
-        )
-        event_id_to_state = yield self.store.get_state_for_events(
-            frozenset(e.event_id for e in events),
-            types=types
-        )
-        res = yield self.filter_events_for_clients(
-            [(user_id, is_peeking)], events, event_id_to_state
-        )
-        defer.returnValue(res.get(user_id, []))
-
     def ratelimit(self, requester):
         time_now = self.clock.time()
         allowed, time_allowed = self.ratelimiter.send_message(
@@ -232,56 +61,6 @@ class BaseHandler(object):
                 retry_after_ms=int(1000 * (time_allowed - time_now)),
             )
 
-    @defer.inlineCallbacks
-    def _create_new_client_event(self, builder, prev_event_ids=None):
-        if prev_event_ids:
-            prev_events = yield self.store.add_event_hashes(prev_event_ids)
-            prev_max_depth = yield self.store.get_max_depth_of_events(prev_event_ids)
-            depth = prev_max_depth + 1
-        else:
-            latest_ret = yield self.store.get_latest_event_ids_and_hashes_in_room(
-                builder.room_id,
-            )
-
-            if latest_ret:
-                depth = max([d for _, _, d in latest_ret]) + 1
-            else:
-                depth = 1
-
-            prev_events = [
-                (event_id, prev_hashes)
-                for event_id, prev_hashes, _ in latest_ret
-            ]
-
-        builder.prev_events = prev_events
-        builder.depth = depth
-
-        state_handler = self.state_handler
-
-        context = yield state_handler.compute_event_context(builder)
-
-        if builder.is_state():
-            builder.prev_state = yield self.store.add_event_hashes(
-                context.prev_state_events
-            )
-
-        yield self.auth.add_auth_events(builder, context)
-
-        add_hashes_and_signatures(
-            builder, self.server_name, self.signing_key
-        )
-
-        event = builder.build()
-
-        logger.debug(
-            "Created event %s with current state: %s",
-            event.event_id, context.current_state,
-        )
-
-        defer.returnValue(
-            (event, context,)
-        )
-
     def is_host_in_room(self, current_state):
         room_members = [
             (state_key, event.membership)
@@ -296,154 +75,13 @@ class BaseHandler(object):
                 return True
         for (state_key, membership) in room_members:
             if (
-                UserID.from_string(state_key).domain == self.hs.hostname
+                self.hs.is_mine_id(state_key)
                 and membership == Membership.JOIN
             ):
                 return True
         return False
 
     @defer.inlineCallbacks
-    def handle_new_client_event(
-        self,
-        requester,
-        event,
-        context,
-        ratelimit=True,
-        extra_users=[]
-    ):
-        # We now need to go and hit out to wherever we need to hit out to.
-
-        if ratelimit:
-            self.ratelimit(requester)
-
-        try:
-            self.auth.check(event, auth_events=context.current_state)
-        except AuthError as err:
-            logger.warn("Denying new event %r because %s", event, err)
-            raise err
-
-        yield self.maybe_kick_guest_users(event, context.current_state.values())
-
-        if event.type == EventTypes.CanonicalAlias:
-            # Check the alias is acually valid (at this time at least)
-            room_alias_str = event.content.get("alias", None)
-            if room_alias_str:
-                room_alias = RoomAlias.from_string(room_alias_str)
-                directory_handler = self.hs.get_handlers().directory_handler
-                mapping = yield directory_handler.get_association(room_alias)
-
-                if mapping["room_id"] != event.room_id:
-                    raise SynapseError(
-                        400,
-                        "Room alias %s does not point to the room" % (
-                            room_alias_str,
-                        )
-                    )
-
-        federation_handler = self.hs.get_handlers().federation_handler
-
-        if event.type == EventTypes.Member:
-            if event.content["membership"] == Membership.INVITE:
-                def is_inviter_member_event(e):
-                    return (
-                        e.type == EventTypes.Member and
-                        e.sender == event.sender
-                    )
-
-                event.unsigned["invite_room_state"] = [
-                    {
-                        "type": e.type,
-                        "state_key": e.state_key,
-                        "content": e.content,
-                        "sender": e.sender,
-                    }
-                    for k, e in context.current_state.items()
-                    if e.type in self.hs.config.room_invite_state_types
-                    or is_inviter_member_event(e)
-                ]
-
-                invitee = UserID.from_string(event.state_key)
-                if not self.hs.is_mine(invitee):
-                    # TODO: Can we add signature from remote server in a nicer
-                    # way? If we have been invited by a remote server, we need
-                    # to get them to sign the event.
-
-                    returned_invite = yield federation_handler.send_invite(
-                        invitee.domain,
-                        event,
-                    )
-
-                    event.unsigned.pop("room_state", None)
-
-                    # TODO: Make sure the signatures actually are correct.
-                    event.signatures.update(
-                        returned_invite.signatures
-                    )
-
-        if event.type == EventTypes.Redaction:
-            if self.auth.check_redaction(event, auth_events=context.current_state):
-                original_event = yield self.store.get_event(
-                    event.redacts,
-                    check_redacted=False,
-                    get_prev_content=False,
-                    allow_rejected=False,
-                    allow_none=False
-                )
-                if event.user_id != original_event.user_id:
-                    raise AuthError(
-                        403,
-                        "You don't have permission to redact events"
-                    )
-
-        if event.type == EventTypes.Create and context.current_state:
-            raise AuthError(
-                403,
-                "Changing the room create event is forbidden",
-            )
-
-        action_generator = ActionGenerator(self.hs)
-        yield action_generator.handle_push_actions_for_event(
-            event, context, self
-        )
-
-        (event_stream_id, max_stream_id) = yield self.store.persist_event(
-            event, context=context
-        )
-
-        # this intentionally does not yield: we don't care about the result
-        # and don't need to wait for it.
-        preserve_fn(self.hs.get_pusherpool().on_new_notifications)(
-            event_stream_id, max_stream_id
-        )
-
-        destinations = set()
-        for k, s in context.current_state.items():
-            try:
-                if k[0] == EventTypes.Member:
-                    if s.content["membership"] == Membership.JOIN:
-                        destinations.add(
-                            UserID.from_string(s.state_key).domain
-                        )
-            except SynapseError:
-                logger.warn(
-                    "Failed to get destination from event %s", s.event_id
-                )
-
-        with PreserveLoggingContext():
-            # Don't block waiting on waking up all the listeners.
-            self.notifier.on_new_room_event(
-                event, event_stream_id, max_stream_id,
-                extra_users=extra_users
-            )
-
-        # If invite, remove room_state from unsigned before sending.
-        event.unsigned.pop("invite_room_state", None)
-
-        federation_handler.handle_new_event(
-            event, destinations=destinations,
-        )
-
-    @defer.inlineCallbacks
     def maybe_kick_guest_users(self, event, current_state):
         # Technically this function invalidates current_state by changing it.
         # Hopefully this isn't that important to the caller.
diff --git a/synapse/handlers/auth.py b/synapse/handlers/auth.py
index 61fe56032a..68d0d78fc6 100644
--- a/synapse/handlers/auth.py
+++ b/synapse/handlers/auth.py
@@ -521,11 +521,11 @@ class AuthHandler(BaseHandler):
         ))
         return m.serialize()
 
-    def generate_short_term_login_token(self, user_id):
+    def generate_short_term_login_token(self, user_id, duration_in_ms=(2 * 60 * 1000)):
         macaroon = self._generate_base_macaroon(user_id)
         macaroon.add_first_party_caveat("type = login")
         now = self.hs.get_clock().time_msec()
-        expiry = now + (2 * 60 * 1000)
+        expiry = now + duration_in_ms
         macaroon.add_first_party_caveat("time < %d" % (expiry,))
         return macaroon.serialize()
 
@@ -615,4 +615,7 @@ class AuthHandler(BaseHandler):
         Returns:
             Whether self.hash(password) == stored_hash (bool).
         """
-        return bcrypt.hashpw(password, stored_hash) == stored_hash
+        if stored_hash:
+            return bcrypt.hashpw(password, stored_hash) == stored_hash
+        else:
+            return False
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index d95e0b23b1..c21d9d4d83 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -33,7 +33,7 @@ from synapse.util.frozenutils import unfreeze
 from synapse.crypto.event_signing import (
     compute_event_signature, add_hashes_and_signatures,
 )
-from synapse.types import UserID
+from synapse.types import UserID, get_domian_from_id
 
 from synapse.events.utils import prune_event
 
@@ -453,7 +453,7 @@ class FederationHandler(BaseHandler):
             joined_domains = {}
             for u, d in joined_users:
                 try:
-                    dom = UserID.from_string(u).domain
+                    dom = get_domian_from_id(u)
                     old_d = joined_domains.get(dom)
                     if old_d:
                         joined_domains[dom] = min(d, old_d)
@@ -682,7 +682,8 @@ class FederationHandler(BaseHandler):
         })
 
         try:
-            event, context = yield self._create_new_client_event(
+            message_handler = self.hs.get_handlers().message_handler
+            event, context = yield message_handler._create_new_client_event(
                 builder=builder,
             )
         except AuthError as e:
@@ -743,9 +744,7 @@ class FederationHandler(BaseHandler):
             try:
                 if k[0] == EventTypes.Member:
                     if s.content["membership"] == Membership.JOIN:
-                        destinations.add(
-                            UserID.from_string(s.state_key).domain
-                        )
+                        destinations.add(get_domian_from_id(s.state_key))
             except:
                 logger.warn(
                     "Failed to get destination from event %s", s.event_id
@@ -915,7 +914,8 @@ class FederationHandler(BaseHandler):
             "state_key": user_id,
         })
 
-        event, context = yield self._create_new_client_event(
+        message_handler = self.hs.get_handlers().message_handler
+        event, context = yield message_handler._create_new_client_event(
             builder=builder,
         )
 
@@ -970,9 +970,7 @@ class FederationHandler(BaseHandler):
             try:
                 if k[0] == EventTypes.Member:
                     if s.content["membership"] == Membership.LEAVE:
-                        destinations.add(
-                            UserID.from_string(s.state_key).domain
-                        )
+                        destinations.add(get_domian_from_id(s.state_key))
             except:
                 logger.warn(
                     "Failed to get destination from event %s", s.event_id
@@ -1115,7 +1113,7 @@ class FederationHandler(BaseHandler):
         if not event.internal_metadata.is_outlier():
             action_generator = ActionGenerator(self.hs)
             yield action_generator.handle_push_actions_for_event(
-                event, context, self
+                event, context
             )
 
         event_stream_id, max_stream_id = yield self.store.persist_event(
@@ -1692,7 +1690,10 @@ class FederationHandler(BaseHandler):
         if (yield self.auth.check_host_in_room(room_id, self.hs.hostname)):
             builder = self.event_builder_factory.new(event_dict)
             EventValidator().validate_new(builder)
-            event, context = yield self._create_new_client_event(builder=builder)
+            message_handler = self.hs.get_handlers().message_handler
+            event, context = yield message_handler._create_new_client_event(
+                builder=builder
+            )
 
             event, context = yield self.add_display_name_to_third_party_invite(
                 event_dict, event, context
@@ -1720,7 +1721,8 @@ class FederationHandler(BaseHandler):
     def on_exchange_third_party_invite_request(self, origin, room_id, event_dict):
         builder = self.event_builder_factory.new(event_dict)
 
-        event, context = yield self._create_new_client_event(
+        message_handler = self.hs.get_handlers().message_handler
+        event, context = yield message_handler._create_new_client_event(
             builder=builder,
         )
 
@@ -1759,7 +1761,8 @@ class FederationHandler(BaseHandler):
         event_dict["content"]["third_party_invite"]["display_name"] = display_name
         builder = self.event_builder_factory.new(event_dict)
         EventValidator().validate_new(builder)
-        event, context = yield self._create_new_client_event(builder=builder)
+        message_handler = self.hs.get_handlers().message_handler
+        event, context = yield message_handler._create_new_client_event(builder=builder)
         defer.returnValue((event, context))
 
     @defer.inlineCallbacks
diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index f51feda2f4..13154edb78 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -17,13 +17,19 @@ from twisted.internet import defer
 
 from synapse.api.constants import EventTypes, Membership
 from synapse.api.errors import AuthError, Codes, SynapseError
-from synapse.streams.config import PaginationConfig
+from synapse.crypto.event_signing import add_hashes_and_signatures
 from synapse.events.utils import serialize_event
 from synapse.events.validator import EventValidator
+from synapse.push.action_generator import ActionGenerator
+from synapse.streams.config import PaginationConfig
+from synapse.types import (
+    UserID, RoomAlias, RoomStreamToken, StreamToken, get_domian_from_id
+)
 from synapse.util import unwrapFirstError
 from synapse.util.async import concurrently_execute
 from synapse.util.caches.snapshot_cache import SnapshotCache
-from synapse.types import UserID, RoomStreamToken, StreamToken
+from synapse.util.logcontext import PreserveLoggingContext, preserve_fn
+from synapse.visibility import filter_events_for_client
 
 from ._base import BaseHandler
 
@@ -123,7 +129,8 @@ class MessageHandler(BaseHandler):
                 "end": next_token.to_string(),
             })
 
-        events = yield self._filter_events_for_client(
+        events = yield filter_events_for_client(
+            self.store,
             user_id,
             events,
             is_peeking=(member_event_id is None),
@@ -483,8 +490,8 @@ class MessageHandler(BaseHandler):
                     ]
                 ).addErrback(unwrapFirstError)
 
-                messages = yield self._filter_events_for_client(
-                    user_id, messages
+                messages = yield filter_events_for_client(
+                    self.store, user_id, messages
                 )
 
                 start_token = now_token.copy_and_replace("room_key", token[0])
@@ -619,8 +626,8 @@ class MessageHandler(BaseHandler):
             end_token=stream_token
         )
 
-        messages = yield self._filter_events_for_client(
-            user_id, messages, is_peeking=is_peeking
+        messages = yield filter_events_for_client(
+            self.store, user_id, messages, is_peeking=is_peeking
         )
 
         start_token = StreamToken.START.copy_and_replace("room_key", token[0])
@@ -700,8 +707,8 @@ class MessageHandler(BaseHandler):
             consumeErrors=True,
         ).addErrback(unwrapFirstError)
 
-        messages = yield self._filter_events_for_client(
-            user_id, messages, is_peeking=is_peeking,
+        messages = yield filter_events_for_client(
+            self.store, user_id, messages, is_peeking=is_peeking,
         )
 
         start_token = now_token.copy_and_replace("room_key", token[0])
@@ -724,3 +731,193 @@ class MessageHandler(BaseHandler):
             ret["membership"] = membership
 
         defer.returnValue(ret)
+
+    @defer.inlineCallbacks
+    def _create_new_client_event(self, builder, prev_event_ids=None):
+        if prev_event_ids:
+            prev_events = yield self.store.add_event_hashes(prev_event_ids)
+            prev_max_depth = yield self.store.get_max_depth_of_events(prev_event_ids)
+            depth = prev_max_depth + 1
+        else:
+            latest_ret = yield self.store.get_latest_event_ids_and_hashes_in_room(
+                builder.room_id,
+            )
+
+            if latest_ret:
+                depth = max([d for _, _, d in latest_ret]) + 1
+            else:
+                depth = 1
+
+            prev_events = [
+                (event_id, prev_hashes)
+                for event_id, prev_hashes, _ in latest_ret
+            ]
+
+        builder.prev_events = prev_events
+        builder.depth = depth
+
+        state_handler = self.state_handler
+
+        context = yield state_handler.compute_event_context(builder)
+
+        if builder.is_state():
+            builder.prev_state = yield self.store.add_event_hashes(
+                context.prev_state_events
+            )
+
+        yield self.auth.add_auth_events(builder, context)
+
+        signing_key = self.hs.config.signing_key[0]
+        add_hashes_and_signatures(
+            builder, self.server_name, signing_key
+        )
+
+        event = builder.build()
+
+        logger.debug(
+            "Created event %s with current state: %s",
+            event.event_id, context.current_state,
+        )
+
+        defer.returnValue(
+            (event, context,)
+        )
+
+    @defer.inlineCallbacks
+    def handle_new_client_event(
+        self,
+        requester,
+        event,
+        context,
+        ratelimit=True,
+        extra_users=[]
+    ):
+        # We now need to go and hit out to wherever we need to hit out to.
+
+        if ratelimit:
+            self.ratelimit(requester)
+
+        try:
+            self.auth.check(event, auth_events=context.current_state)
+        except AuthError as err:
+            logger.warn("Denying new event %r because %s", event, err)
+            raise err
+
+        yield self.maybe_kick_guest_users(event, context.current_state.values())
+
+        if event.type == EventTypes.CanonicalAlias:
+            # Check the alias is acually valid (at this time at least)
+            room_alias_str = event.content.get("alias", None)
+            if room_alias_str:
+                room_alias = RoomAlias.from_string(room_alias_str)
+                directory_handler = self.hs.get_handlers().directory_handler
+                mapping = yield directory_handler.get_association(room_alias)
+
+                if mapping["room_id"] != event.room_id:
+                    raise SynapseError(
+                        400,
+                        "Room alias %s does not point to the room" % (
+                            room_alias_str,
+                        )
+                    )
+
+        federation_handler = self.hs.get_handlers().federation_handler
+
+        if event.type == EventTypes.Member:
+            if event.content["membership"] == Membership.INVITE:
+                def is_inviter_member_event(e):
+                    return (
+                        e.type == EventTypes.Member and
+                        e.sender == event.sender
+                    )
+
+                event.unsigned["invite_room_state"] = [
+                    {
+                        "type": e.type,
+                        "state_key": e.state_key,
+                        "content": e.content,
+                        "sender": e.sender,
+                    }
+                    for k, e in context.current_state.items()
+                    if e.type in self.hs.config.room_invite_state_types
+                    or is_inviter_member_event(e)
+                ]
+
+                invitee = UserID.from_string(event.state_key)
+                if not self.hs.is_mine(invitee):
+                    # TODO: Can we add signature from remote server in a nicer
+                    # way? If we have been invited by a remote server, we need
+                    # to get them to sign the event.
+
+                    returned_invite = yield federation_handler.send_invite(
+                        invitee.domain,
+                        event,
+                    )
+
+                    event.unsigned.pop("room_state", None)
+
+                    # TODO: Make sure the signatures actually are correct.
+                    event.signatures.update(
+                        returned_invite.signatures
+                    )
+
+        if event.type == EventTypes.Redaction:
+            if self.auth.check_redaction(event, auth_events=context.current_state):
+                original_event = yield self.store.get_event(
+                    event.redacts,
+                    check_redacted=False,
+                    get_prev_content=False,
+                    allow_rejected=False,
+                    allow_none=False
+                )
+                if event.user_id != original_event.user_id:
+                    raise AuthError(
+                        403,
+                        "You don't have permission to redact events"
+                    )
+
+        if event.type == EventTypes.Create and context.current_state:
+            raise AuthError(
+                403,
+                "Changing the room create event is forbidden",
+            )
+
+        action_generator = ActionGenerator(self.hs)
+        yield action_generator.handle_push_actions_for_event(
+            event, context
+        )
+
+        (event_stream_id, max_stream_id) = yield self.store.persist_event(
+            event, context=context
+        )
+
+        # this intentionally does not yield: we don't care about the result
+        # and don't need to wait for it.
+        preserve_fn(self.hs.get_pusherpool().on_new_notifications)(
+            event_stream_id, max_stream_id
+        )
+
+        destinations = set()
+        for k, s in context.current_state.items():
+            try:
+                if k[0] == EventTypes.Member:
+                    if s.content["membership"] == Membership.JOIN:
+                        destinations.add(get_domian_from_id(s.state_key))
+            except SynapseError:
+                logger.warn(
+                    "Failed to get destination from event %s", s.event_id
+                )
+
+        with PreserveLoggingContext():
+            # Don't block waiting on waking up all the listeners.
+            self.notifier.on_new_room_event(
+                event, event_stream_id, max_stream_id,
+                extra_users=extra_users
+            )
+
+        # If invite, remove room_state from unsigned before sending.
+        event.unsigned.pop("invite_room_state", None)
+
+        federation_handler.handle_new_event(
+            event, destinations=destinations,
+        )
diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py
index d0c8f1328b..a8529cce42 100644
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -33,7 +33,7 @@ from synapse.util.logcontext import preserve_fn
 from synapse.util.logutils import log_function
 from synapse.util.metrics import Measure
 from synapse.util.wheel_timer import WheelTimer
-from synapse.types import UserID
+from synapse.types import UserID, get_domian_from_id
 import synapse.metrics
 
 from ._base import BaseHandler
@@ -168,7 +168,7 @@ class PresenceHandler(BaseHandler):
         # The initial delay is to allow disconnected clients a chance to
         # reconnect before we treat them as offline.
         self.clock.call_later(
-            0 * 1000,
+            30 * 1000,
             self.clock.looping_call,
             self._handle_timeouts,
             5000,
@@ -440,7 +440,7 @@ class PresenceHandler(BaseHandler):
             if not local_states:
                 continue
 
-            host = UserID.from_string(user_id).domain
+            host = get_domian_from_id(user_id)
             hosts_to_states.setdefault(host, []).extend(local_states)
 
         # TODO: de-dup hosts_to_states, as a single host might have multiple
diff --git a/synapse/handlers/register.py b/synapse/handlers/register.py
index b0862067e1..5883b9111e 100644
--- a/synapse/handlers/register.py
+++ b/synapse/handlers/register.py
@@ -358,6 +358,59 @@ class RegistrationHandler(BaseHandler):
         )
         defer.returnValue(data)
 
+    @defer.inlineCallbacks
+    def get_or_create_user(self, localpart, displayname, duration_seconds):
+        """Creates a new user or returns an access token for an existing one
+
+        Args:
+            localpart : The local part of the user ID to register. If None,
+              one will be randomly generated.
+        Returns:
+            A tuple of (user_id, access_token).
+        Raises:
+            RegistrationError if there was a problem registering.
+        """
+        yield run_on_reactor()
+
+        if localpart is None:
+            raise SynapseError(400, "Request must include user id")
+
+        need_register = True
+
+        try:
+            yield self.check_username(localpart)
+        except SynapseError as e:
+            if e.errcode == Codes.USER_IN_USE:
+                need_register = False
+            else:
+                raise
+
+        user = UserID(localpart, self.hs.hostname)
+        user_id = user.to_string()
+        auth_handler = self.hs.get_handlers().auth_handler
+        token = auth_handler.generate_short_term_login_token(user_id, duration_seconds)
+
+        if need_register:
+            yield self.store.register(
+                user_id=user_id,
+                token=token,
+                password_hash=None
+            )
+
+            yield registered_user(self.distributor, user)
+        else:
+            yield self.store.flush_user(user_id=user_id)
+            yield self.store.add_access_token_to_user(user_id=user_id, token=token)
+
+        if displayname is not None:
+            logger.info("setting user display name: %s -> %s", user_id, displayname)
+            profile_handler = self.hs.get_handlers().profile_handler
+            yield profile_handler.set_displayname(
+                user, user, displayname
+            )
+
+        defer.returnValue((user_id, token))
+
     def auth_handler(self):
         return self.hs.get_handlers().auth_handler
 
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index dd9c18df84..3d63b3c513 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -26,6 +26,7 @@ from synapse.api.errors import AuthError, StoreError, SynapseError
 from synapse.util import stringutils
 from synapse.util.async import concurrently_execute
 from synapse.util.caches.response_cache import ResponseCache
+from synapse.visibility import filter_events_for_client
 
 from collections import OrderedDict
 
@@ -449,10 +450,12 @@ class RoomContextHandler(BaseHandler):
         now_token = yield self.hs.get_event_sources().get_current_token()
 
         def filter_evts(events):
-            return self._filter_events_for_client(
+            return filter_events_for_client(
+                self.store,
                 user.to_string(),
                 events,
-                is_peeking=is_guest)
+                is_peeking=is_guest
+            )
 
         event = yield self.store.get_event(event_id, get_prev_content=True,
                                            allow_none=True)
diff --git a/synapse/handlers/room_member.py b/synapse/handlers/room_member.py
index ed2cda837f..b44e52a515 100644
--- a/synapse/handlers/room_member.py
+++ b/synapse/handlers/room_member.py
@@ -113,7 +113,7 @@ class RoomMemberHandler(BaseHandler):
             prev_event_ids=prev_event_ids,
         )
 
-        yield self.handle_new_client_event(
+        yield msg_handler.handle_new_client_event(
             requester,
             event,
             context,
@@ -357,7 +357,7 @@ class RoomMemberHandler(BaseHandler):
                 # so don't really fit into the general auth process.
                 raise AuthError(403, "Guest access not allowed")
 
-        yield self.handle_new_client_event(
+        yield message_handler.handle_new_client_event(
             requester,
             event,
             context,
diff --git a/synapse/handlers/search.py b/synapse/handlers/search.py
index 9937d8dd7f..df75d70fac 100644
--- a/synapse/handlers/search.py
+++ b/synapse/handlers/search.py
@@ -21,6 +21,7 @@ from synapse.api.constants import Membership, EventTypes
 from synapse.api.filtering import Filter
 from synapse.api.errors import SynapseError
 from synapse.events.utils import serialize_event
+from synapse.visibility import filter_events_for_client
 
 from unpaddedbase64 import decode_base64, encode_base64
 
@@ -172,8 +173,8 @@ class SearchHandler(BaseHandler):
 
             filtered_events = search_filter.filter([r["event"] for r in results])
 
-            events = yield self._filter_events_for_client(
-                user.to_string(), filtered_events
+            events = yield filter_events_for_client(
+                self.store, user.to_string(), filtered_events
             )
 
             events.sort(key=lambda e: -rank_map[e.event_id])
@@ -223,8 +224,8 @@ class SearchHandler(BaseHandler):
                     r["event"] for r in results
                 ])
 
-                events = yield self._filter_events_for_client(
-                    user.to_string(), filtered_events
+                events = yield filter_events_for_client(
+                    self.store, user.to_string(), filtered_events
                 )
 
                 room_events.extend(events)
@@ -281,12 +282,12 @@ class SearchHandler(BaseHandler):
                     event.room_id, event.event_id, before_limit, after_limit
                 )
 
-                res["events_before"] = yield self._filter_events_for_client(
-                    user.to_string(), res["events_before"]
+                res["events_before"] = yield filter_events_for_client(
+                    self.store, user.to_string(), res["events_before"]
                 )
 
-                res["events_after"] = yield self._filter_events_for_client(
-                    user.to_string(), res["events_after"]
+                res["events_after"] = yield filter_events_for_client(
+                    self.store, user.to_string(), res["events_after"]
                 )
 
                 res["start"] = now_token.copy_and_replace(
diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py
index 231140b655..921215469f 100644
--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -22,6 +22,7 @@ from synapse.util.logcontext import LoggingContext
 from synapse.util.metrics import Measure
 from synapse.util.caches.response_cache import ResponseCache
 from synapse.push.clientformat import format_push_rules_for_user
+from synapse.visibility import filter_events_for_client
 
 from twisted.internet import defer
 
@@ -247,6 +248,10 @@ class SyncHandler(BaseHandler):
             sync_config.user.to_string()
         )
 
+        ignored_users = account_data.get(
+            "m.ignored_user_list", {}
+        ).get("ignored_users", {}).keys()
+
         joined = []
         invited = []
         archived = []
@@ -267,6 +272,8 @@ class SyncHandler(BaseHandler):
                 )
                 joined.append(room_result)
             elif event.membership == Membership.INVITE:
+                if event.sender in ignored_users:
+                    return
                 invite = yield self.store.get_event(event.event_id)
                 invited.append(InvitedSyncResult(
                     room_id=event.room_id,
@@ -515,6 +522,15 @@ class SyncHandler(BaseHandler):
                 sync_config.user
             )
 
+        ignored_account_data = yield self.store.get_global_account_data_by_type_for_user(
+            "m.ignored_user_list", user_id=user_id,
+        )
+
+        if ignored_account_data:
+            ignored_users = ignored_account_data.get("ignored_users", {}).keys()
+        else:
+            ignored_users = frozenset()
+
         # Get a list of membership change events that have happened.
         rooms_changed = yield self.store.get_membership_changes_for_user(
             user_id, since_token.room_key, now_token.room_key
@@ -549,9 +565,10 @@ class SyncHandler(BaseHandler):
             # Only bother if we're still currently invited
             should_invite = non_joins[-1].membership == Membership.INVITE
             if should_invite:
-                room_sync = InvitedSyncResult(room_id, invite=non_joins[-1])
-                if room_sync:
-                    invited.append(room_sync)
+                if event.sender not in ignored_users:
+                    room_sync = InvitedSyncResult(room_id, invite=non_joins[-1])
+                    if room_sync:
+                        invited.append(room_sync)
 
             # Always include leave/ban events. Just take the last one.
             # TODO: How do we handle ban -> leave in same batch?
@@ -681,7 +698,8 @@ class SyncHandler(BaseHandler):
 
             if recents is not None:
                 recents = sync_config.filter_collection.filter_room_timeline(recents)
-                recents = yield self._filter_events_for_client(
+                recents = yield filter_events_for_client(
+                    self.store,
                     sync_config.user.to_string(),
                     recents,
                 )
@@ -702,7 +720,8 @@ class SyncHandler(BaseHandler):
                 loaded_recents = sync_config.filter_collection.filter_room_timeline(
                     events
                 )
-                loaded_recents = yield self._filter_events_for_client(
+                loaded_recents = yield filter_events_for_client(
+                    self.store,
                     sync_config.user.to_string(),
                     loaded_recents,
                 )
diff --git a/synapse/notifier.py b/synapse/notifier.py
index 6af7a8f424..33b79c0ec7 100644
--- a/synapse/notifier.py
+++ b/synapse/notifier.py
@@ -21,6 +21,7 @@ from synapse.util.logutils import log_function
 from synapse.util.async import ObservableDeferred
 from synapse.util.logcontext import PreserveLoggingContext
 from synapse.types import StreamToken
+from synapse.visibility import filter_events_for_client
 import synapse.metrics
 
 from collections import namedtuple
@@ -398,8 +399,8 @@ class Notifier(object):
                 )
 
                 if name == "room":
-                    room_member_handler = self.hs.get_handlers().room_member_handler
-                    new_events = yield room_member_handler._filter_events_for_client(
+                    new_events = yield filter_events_for_client(
+                        self.store,
                         user.to_string(),
                         new_events,
                         is_peeking=is_peeking,
diff --git a/synapse/push/action_generator.py b/synapse/push/action_generator.py
index a0160994b7..9b208668b6 100644
--- a/synapse/push/action_generator.py
+++ b/synapse/push/action_generator.py
@@ -37,14 +37,14 @@ class ActionGenerator:
         # tag (ie. we just need all the users).
 
     @defer.inlineCallbacks
-    def handle_push_actions_for_event(self, event, context, handler):
+    def handle_push_actions_for_event(self, event, context):
         with Measure(self.clock, "handle_push_actions_for_event"):
             bulk_evaluator = yield evaluator_for_event(
                 event, self.hs, self.store
             )
 
             actions_by_user = yield bulk_evaluator.action_for_event_by_user(
-                event, handler, context.current_state
+                event, context.current_state
             )
 
             context.push_actions = [
diff --git a/synapse/push/bulk_push_rule_evaluator.py b/synapse/push/bulk_push_rule_evaluator.py
index f97df36d80..25e13b3423 100644
--- a/synapse/push/bulk_push_rule_evaluator.py
+++ b/synapse/push/bulk_push_rule_evaluator.py
@@ -22,6 +22,7 @@ from .baserules import list_with_base_rules
 from .push_rule_evaluator import PushRuleEvaluatorForEvent
 
 from synapse.api.constants import EventTypes
+from synapse.visibility import filter_events_for_clients
 
 
 logger = logging.getLogger(__name__)
@@ -126,7 +127,7 @@ class BulkPushRuleEvaluator:
         self.store = store
 
     @defer.inlineCallbacks
-    def action_for_event_by_user(self, event, handler, current_state):
+    def action_for_event_by_user(self, event, current_state):
         actions_by_user = {}
 
         # None of these users can be peeking since this list of users comes
@@ -136,8 +137,8 @@ class BulkPushRuleEvaluator:
             (u, False) for u in self.rules_by_user.keys()
         ]
 
-        filtered_by_user = yield handler.filter_events_for_clients(
-            user_tuples, [event], {event.event_id: current_state}
+        filtered_by_user = yield filter_events_for_clients(
+            self.store, user_tuples, [event], {event.event_id: current_state}
         )
 
         room_members = yield self.store.get_users_in_room(self.room_id)
diff --git a/synapse/push/emailpusher.py b/synapse/push/emailpusher.py
new file mode 100644
index 0000000000..3a13c7485a
--- /dev/null
+++ b/synapse/push/emailpusher.py
@@ -0,0 +1,251 @@
+# -*- coding: utf-8 -*-
+# Copyright 2016 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.
+
+from twisted.internet import defer, reactor
+
+import logging
+
+from synapse.util.metrics import Measure
+from synapse.util.logcontext import LoggingContext
+
+from mailer import Mailer
+
+logger = logging.getLogger(__name__)
+
+# The amount of time we always wait before ever emailing about a notification
+# (to give the user a chance to respond to other push or notice the window)
+DELAY_BEFORE_MAIL_MS = 2 * 60 * 1000
+
+THROTTLE_START_MS = 2 * 60 * 1000
+THROTTLE_MAX_MS = (2 * 60 * 1000) * (2 ** 11)  # ~3 days
+THROTTLE_MULTIPLIER = 2
+
+# If no event triggers a notification for this long after the previous,
+# the throttle is released.
+THROTTLE_RESET_AFTER_MS = (2 * 60 * 1000) * (2 ** 11)  # ~3 days
+
+
+class EmailPusher(object):
+    """
+    A pusher that sends email notifications about events (approximately)
+    when they happen.
+    This shares quite a bit of code with httpusher: it would be good to
+    factor out the common parts
+    """
+    def __init__(self, hs, pusherdict):
+        self.hs = hs
+        self.store = self.hs.get_datastore()
+        self.clock = self.hs.get_clock()
+        self.pusher_id = pusherdict['id']
+        self.user_id = pusherdict['user_name']
+        self.app_id = pusherdict['app_id']
+        self.email = pusherdict['pushkey']
+        self.last_stream_ordering = pusherdict['last_stream_ordering']
+        self.timed_call = None
+        self.throttle_params = None
+
+        # See httppusher
+        self.max_stream_ordering = None
+
+        self.processing = False
+
+        if self.hs.config.email_enable_notifs:
+            self.mailer = Mailer(self.hs)
+        else:
+            self.mailer = None
+
+    @defer.inlineCallbacks
+    def on_started(self):
+        if self.mailer is not None:
+            self.throttle_params = yield self.store.get_throttle_params_by_room(
+                self.pusher_id
+            )
+            yield self._process()
+
+    def on_stop(self):
+        if self.timed_call:
+            self.timed_call.cancel()
+
+    @defer.inlineCallbacks
+    def on_new_notifications(self, min_stream_ordering, max_stream_ordering):
+        self.max_stream_ordering = max(max_stream_ordering, self.max_stream_ordering)
+        yield self._process()
+
+    def on_new_receipts(self, min_stream_id, max_stream_id):
+        # We could wake up and cancel the timer but there tend to be quite a
+        # lot of read receipts so it's probably less work to just let the
+        # timer fire
+        return defer.succeed(None)
+
+    @defer.inlineCallbacks
+    def on_timer(self):
+        self.timed_call = None
+        yield self._process()
+
+    @defer.inlineCallbacks
+    def _process(self):
+        if self.processing:
+            return
+
+        with LoggingContext("emailpush._process"):
+            with Measure(self.clock, "emailpush._process"):
+                try:
+                    self.processing = True
+                    # if the max ordering changes while we're running _unsafe_process,
+                    # call it again, and so on until we've caught up.
+                    while True:
+                        starting_max_ordering = self.max_stream_ordering
+                        try:
+                            yield self._unsafe_process()
+                        except:
+                            logger.exception("Exception processing notifs")
+                        if self.max_stream_ordering == starting_max_ordering:
+                            break
+                finally:
+                    self.processing = False
+
+    @defer.inlineCallbacks
+    def _unsafe_process(self):
+        """
+        Main logic of the push loop without the wrapper function that sets
+        up logging, measures and guards against multiple instances of it
+        being run.
+        """
+        unprocessed = yield self.store.get_unread_push_actions_for_user_in_range(
+            self.user_id, self.last_stream_ordering, self.max_stream_ordering
+        )
+
+        soonest_due_at = None
+
+        for push_action in unprocessed:
+            received_at = push_action['received_ts']
+            if received_at is None:
+                received_at = 0
+            notif_ready_at = received_at + DELAY_BEFORE_MAIL_MS
+
+            room_ready_at = self.room_ready_to_notify_at(
+                push_action['room_id']
+            )
+
+            should_notify_at = max(notif_ready_at, room_ready_at)
+
+            if should_notify_at < self.clock.time_msec():
+                # one of our notifications is ready for sending, so we send
+                # *one* email updating the user on their notifications,
+                # we then consider all previously outstanding notifications
+                # to be delivered.
+                yield self.send_notification(unprocessed)
+
+                yield self.save_last_stream_ordering_and_success(max([
+                    ea['stream_ordering'] for ea in unprocessed
+                ]))
+                yield self.sent_notif_update_throttle(
+                    push_action['room_id'], push_action
+                )
+                break
+            else:
+                if soonest_due_at is None or should_notify_at < soonest_due_at:
+                    soonest_due_at = should_notify_at
+
+                if self.timed_call is not None:
+                    self.timed_call.cancel()
+                    self.timed_call = None
+
+        if soonest_due_at is not None:
+            self.timed_call = reactor.callLater(
+                self.seconds_until(soonest_due_at), self.on_timer
+            )
+
+    @defer.inlineCallbacks
+    def save_last_stream_ordering_and_success(self, last_stream_ordering):
+        self.last_stream_ordering = last_stream_ordering
+        yield self.store.update_pusher_last_stream_ordering_and_success(
+            self.app_id, self.email, self.user_id,
+            last_stream_ordering, self.clock.time_msec()
+        )
+
+    def seconds_until(self, ts_msec):
+        return (ts_msec - self.clock.time_msec()) / 1000
+
+    def get_room_throttle_ms(self, room_id):
+        if room_id in self.throttle_params:
+            return self.throttle_params[room_id]["throttle_ms"]
+        else:
+            return 0
+
+    def get_room_last_sent_ts(self, room_id):
+        if room_id in self.throttle_params:
+            return self.throttle_params[room_id]["last_sent_ts"]
+        else:
+            return 0
+
+    def room_ready_to_notify_at(self, room_id):
+        """
+        Determines whether throttling should prevent us from sending an email
+        for the given room
+        Returns: True if we should send, False if we should not
+        """
+        last_sent_ts = self.get_room_last_sent_ts(room_id)
+        throttle_ms = self.get_room_throttle_ms(room_id)
+
+        may_send_at = last_sent_ts + throttle_ms
+        return may_send_at
+
+    @defer.inlineCallbacks
+    def sent_notif_update_throttle(self, room_id, notified_push_action):
+        # We have sent a notification, so update the throttle accordingly.
+        # If the event that triggered the notif happened more than
+        # THROTTLE_RESET_AFTER_MS after the previous one that triggered a
+        # notif, we release the throttle. Otherwise, the throttle is increased.
+        time_of_previous_notifs = yield self.store.get_time_of_last_push_action_before(
+            notified_push_action['stream_ordering']
+        )
+
+        time_of_this_notifs = notified_push_action['received_ts']
+
+        if time_of_previous_notifs is not None and time_of_this_notifs is not None:
+            gap = time_of_this_notifs - time_of_previous_notifs
+        else:
+            # if we don't know the arrival time of one of the notifs (it was not
+            # stored prior to email notification code) then assume a gap of
+            # zero which will just not reset the throttle
+            gap = 0
+
+        current_throttle_ms = self.get_room_throttle_ms(room_id)
+
+        if gap > THROTTLE_RESET_AFTER_MS:
+            new_throttle_ms = THROTTLE_START_MS
+        else:
+            if current_throttle_ms == 0:
+                new_throttle_ms = THROTTLE_START_MS
+            else:
+                new_throttle_ms = min(
+                    current_throttle_ms * THROTTLE_MULTIPLIER,
+                    THROTTLE_MAX_MS
+                )
+        self.throttle_params[room_id] = {
+            "last_sent_ts": self.clock.time_msec(),
+            "throttle_ms": new_throttle_ms
+        }
+        yield self.store.set_throttle_params(
+            self.pusher_id, room_id, self.throttle_params[room_id]
+        )
+
+    @defer.inlineCallbacks
+    def send_notification(self, push_actions):
+        logger.info("Sending notif email for user %r", self.user_id)
+        yield self.mailer.send_notification_mail(
+            self.user_id, self.email, push_actions
+        )
diff --git a/synapse/push/mailer.py b/synapse/push/mailer.py
new file mode 100644
index 0000000000..2fd38a036a
--- /dev/null
+++ b/synapse/push/mailer.py
@@ -0,0 +1,456 @@
+# -*- coding: utf-8 -*-
+# Copyright 2016 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.
+
+from twisted.internet import defer
+from twisted.mail.smtp import sendmail
+
+import email.utils
+import email.mime.multipart
+from email.mime.text import MIMEText
+from email.mime.multipart import MIMEMultipart
+
+from synapse.util.async import concurrently_execute
+from synapse.util.presentable_names import (
+    calculate_room_name, name_from_member_event, descriptor_from_member_events
+)
+from synapse.types import UserID
+from synapse.api.errors import StoreError
+from synapse.api.constants import EventTypes
+from synapse.visibility import filter_events_for_client
+
+import jinja2
+import bleach
+
+import time
+import urllib
+
+import logging
+logger = logging.getLogger(__name__)
+
+
+MESSAGE_FROM_PERSON_IN_ROOM = "You have a message on %(app)s from %(person)s " \
+                              "in the %s room..."
+MESSAGE_FROM_PERSON = "You have a message on %(app)s from %(person)s..."
+MESSAGES_FROM_PERSON = "You have messages on %(app)s from %(person)s..."
+MESSAGES_IN_ROOM = "There are some messages on %(app)s for you in the %(room)s room..."
+MESSAGES_IN_ROOMS = "Here are some messages on %(app)s you may have missed..."
+INVITE_FROM_PERSON_TO_ROOM = "%(person)s has invited you to join the " \
+                             "%(room)s room on %(app)s..."
+INVITE_FROM_PERSON = "%(person)s has invited you to chat on %(app)s..."
+
+CONTEXT_BEFORE = 1
+CONTEXT_AFTER = 1
+
+# From https://github.com/matrix-org/matrix-react-sdk/blob/master/src/HtmlUtils.js
+ALLOWED_TAGS = [
+    'font',  # custom to matrix for IRC-style font coloring
+    'del',  # for markdown
+    # deliberately no h1/h2 to stop people shouting.
+    'h3', 'h4', 'h5', 'h6', 'blockquote', 'p', 'a', 'ul', 'ol',
+    'nl', 'li', 'b', 'i', 'u', 'strong', 'em', 'strike', 'code', 'hr', 'br', 'div',
+    'table', 'thead', 'caption', 'tbody', 'tr', 'th', 'td', 'pre'
+]
+ALLOWED_ATTRS = {
+    # custom ones first:
+    "font": ["color"],  # custom to matrix
+    "a": ["href", "name", "target"],  # remote target: custom to matrix
+    # We don't currently allow img itself by default, but this
+    # would make sense if we did
+    "img": ["src"],
+}
+# When bleach release a version with this option, we can specify schemes
+# ALLOWED_SCHEMES = ["http", "https", "ftp", "mailto"]
+
+
+class Mailer(object):
+    def __init__(self, hs):
+        self.hs = hs
+        self.store = self.hs.get_datastore()
+        self.state_handler = self.hs.get_state_handler()
+        loader = jinja2.FileSystemLoader(self.hs.config.email_template_dir)
+        self.app_name = self.hs.config.email_app_name
+        env = jinja2.Environment(loader=loader)
+        env.filters["format_ts"] = format_ts_filter
+        env.filters["mxc_to_http"] = self.mxc_to_http_filter
+        self.notif_template_html = env.get_template(
+            self.hs.config.email_notif_template_html
+        )
+        self.notif_template_text = env.get_template(
+            self.hs.config.email_notif_template_text
+        )
+
+    @defer.inlineCallbacks
+    def send_notification_mail(self, user_id, email_address, push_actions):
+        raw_from = email.utils.parseaddr(self.hs.config.email_notif_from)[1]
+        raw_to = email.utils.parseaddr(email_address)[1]
+
+        if raw_to == '':
+            raise RuntimeError("Invalid 'to' address")
+
+        rooms_in_order = deduped_ordered_list(
+            [pa['room_id'] for pa in push_actions]
+        )
+
+        notif_events = yield self.store.get_events(
+            [pa['event_id'] for pa in push_actions]
+        )
+
+        notifs_by_room = {}
+        for pa in push_actions:
+            notifs_by_room.setdefault(pa["room_id"], []).append(pa)
+
+        # collect the current state for all the rooms in which we have
+        # notifications
+        state_by_room = {}
+
+        try:
+            user_display_name = yield self.store.get_profile_displayname(
+                UserID.from_string(user_id).localpart
+            )
+        except StoreError:
+            user_display_name = user_id
+
+        @defer.inlineCallbacks
+        def _fetch_room_state(room_id):
+            room_state = yield self.state_handler.get_current_state(room_id)
+            state_by_room[room_id] = room_state
+
+        # Run at most 3 of these at once: sync does 10 at a time but email
+        # notifs are much realtime than sync so we can afford to wait a bit.
+        yield concurrently_execute(_fetch_room_state, rooms_in_order, 3)
+
+        rooms = []
+
+        for r in rooms_in_order:
+            roomvars = yield self.get_room_vars(
+                r, user_id, notifs_by_room[r], notif_events, state_by_room[r]
+            )
+            rooms.append(roomvars)
+
+        summary_text = self.make_summary_text(
+            notifs_by_room, state_by_room, notif_events, user_id
+        )
+
+        template_vars = {
+            "user_display_name": user_display_name,
+            "unsubscribe_link": self.make_unsubscribe_link(),
+            "summary_text": summary_text,
+            "app_name": self.app_name,
+            "rooms": rooms,
+        }
+
+        html_text = self.notif_template_html.render(**template_vars)
+        html_part = MIMEText(html_text, "html", "utf8")
+
+        plain_text = self.notif_template_text.render(**template_vars)
+        text_part = MIMEText(plain_text, "plain", "utf8")
+
+        multipart_msg = MIMEMultipart('alternative')
+        multipart_msg['Subject'] = "[%s] %s" % (self.app_name, summary_text)
+        multipart_msg['From'] = self.hs.config.email_notif_from
+        multipart_msg['To'] = email_address
+        multipart_msg['Date'] = email.utils.formatdate()
+        multipart_msg['Message-ID'] = email.utils.make_msgid()
+        multipart_msg.attach(text_part)
+        multipart_msg.attach(html_part)
+
+        logger.info("Sending email push notification to %s" % email_address)
+        # logger.debug(html_text)
+
+        yield sendmail(
+            self.hs.config.email_smtp_host,
+            raw_from, raw_to, multipart_msg.as_string(),
+            port=self.hs.config.email_smtp_port
+        )
+
+    @defer.inlineCallbacks
+    def get_room_vars(self, room_id, user_id, notifs, notif_events, room_state):
+        my_member_event = room_state[("m.room.member", user_id)]
+        is_invite = my_member_event.content["membership"] == "invite"
+
+        room_vars = {
+            "title": calculate_room_name(room_state, user_id),
+            "hash": string_ordinal_total(room_id),  # See sender avatar hash
+            "notifs": [],
+            "invite": is_invite,
+            "link": self.make_room_link(room_id),
+        }
+
+        if not is_invite:
+            for n in notifs:
+                notifvars = yield self.get_notif_vars(
+                    n, user_id, notif_events[n['event_id']], room_state
+                )
+
+                # merge overlapping notifs together.
+                # relies on the notifs being in chronological order.
+                merge = False
+                if room_vars['notifs'] and 'messages' in room_vars['notifs'][-1]:
+                    prev_messages = room_vars['notifs'][-1]['messages']
+                    for message in notifvars['messages']:
+                        pm = filter(lambda pm: pm['id'] == message['id'], prev_messages)
+                        if pm:
+                            if not message["is_historical"]:
+                                pm[0]["is_historical"] = False
+                            merge = True
+                        elif merge:
+                            # we're merging, so append any remaining messages
+                            # in this notif to the previous one
+                            prev_messages.append(message)
+
+                if not merge:
+                    room_vars['notifs'].append(notifvars)
+
+        defer.returnValue(room_vars)
+
+    @defer.inlineCallbacks
+    def get_notif_vars(self, notif, user_id, notif_event, room_state):
+        results = yield self.store.get_events_around(
+            notif['room_id'], notif['event_id'],
+            before_limit=CONTEXT_BEFORE, after_limit=CONTEXT_AFTER
+        )
+
+        ret = {
+            "link": self.make_notif_link(notif),
+            "ts": notif['received_ts'],
+            "messages": [],
+        }
+
+        the_events = yield filter_events_for_client(
+            self.store, user_id, results["events_before"]
+        )
+        the_events.append(notif_event)
+
+        for event in the_events:
+            messagevars = self.get_message_vars(notif, event, room_state)
+            if messagevars is not None:
+                ret['messages'].append(messagevars)
+
+        defer.returnValue(ret)
+
+    def get_message_vars(self, notif, event, room_state):
+        if event.type != EventTypes.Message:
+            return None
+
+        sender_state_event = room_state[("m.room.member", event.sender)]
+        sender_name = name_from_member_event(sender_state_event)
+        sender_avatar_url = sender_state_event.content["avatar_url"]
+
+        # 'hash' for deterministically picking default images: use
+        # sender_hash % the number of default images to choose from
+        sender_hash = string_ordinal_total(event.sender)
+
+        ret = {
+            "msgtype": event.content["msgtype"],
+            "is_historical": event.event_id != notif['event_id'],
+            "id": event.event_id,
+            "ts": event.origin_server_ts,
+            "sender_name": sender_name,
+            "sender_avatar_url": sender_avatar_url,
+            "sender_hash": sender_hash,
+        }
+
+        if event.content["msgtype"] == "m.text":
+            self.add_text_message_vars(ret, event)
+        elif event.content["msgtype"] == "m.image":
+            self.add_image_message_vars(ret, event)
+
+        if "body" in event.content:
+            ret["body_text_plain"] = event.content["body"]
+
+        return ret
+
+    def add_text_message_vars(self, messagevars, event):
+        if "format" in event.content:
+            msgformat = event.content["format"]
+        else:
+            msgformat = None
+        messagevars["format"] = msgformat
+
+        if msgformat == "org.matrix.custom.html":
+            messagevars["body_text_html"] = safe_markup(event.content["formatted_body"])
+        else:
+            messagevars["body_text_html"] = safe_text(event.content["body"])
+
+        return messagevars
+
+    def add_image_message_vars(self, messagevars, event):
+        messagevars["image_url"] = event.content["url"]
+
+        return messagevars
+
+    def make_summary_text(self, notifs_by_room, state_by_room, notif_events, user_id):
+        if len(notifs_by_room) == 1:
+            # Only one room has new stuff
+            room_id = notifs_by_room.keys()[0]
+
+            # If the room has some kind of name, use it, but we don't
+            # want the generated-from-names one here otherwise we'll
+            # end up with, "new message from Bob in the Bob room"
+            room_name = calculate_room_name(
+                state_by_room[room_id], user_id, fallback_to_members=False
+            )
+
+            my_member_event = state_by_room[room_id][("m.room.member", user_id)]
+            if my_member_event.content["membership"] == "invite":
+                inviter_member_event = state_by_room[room_id][
+                    ("m.room.member", my_member_event.sender)
+                ]
+                inviter_name = name_from_member_event(inviter_member_event)
+
+                if room_name is None:
+                    return INVITE_FROM_PERSON % {
+                        "person": inviter_name,
+                        "app": self.app_name
+                    }
+                else:
+                    return INVITE_FROM_PERSON_TO_ROOM % {
+                        "person": inviter_name,
+                        "room": room_name,
+                        "app": self.app_name,
+                    }
+
+            sender_name = None
+            if len(notifs_by_room[room_id]) == 1:
+                # There is just the one notification, so give some detail
+                event = notif_events[notifs_by_room[room_id][0]["event_id"]]
+                if ("m.room.member", event.sender) in state_by_room[room_id]:
+                    state_event = state_by_room[room_id][("m.room.member", event.sender)]
+                    sender_name = name_from_member_event(state_event)
+
+                if sender_name is not None and room_name is not None:
+                    return MESSAGE_FROM_PERSON_IN_ROOM % {
+                        "person": sender_name,
+                        "room": room_name,
+                        "app": self.app_name,
+                    }
+                elif sender_name is not None:
+                    return MESSAGE_FROM_PERSON % {
+                        "person": sender_name,
+                        "app": self.app_name,
+                    }
+            else:
+                # There's more than one notification for this room, so just
+                # say there are several
+                if room_name is not None:
+                    return MESSAGES_IN_ROOM % {
+                        "room": room_name,
+                        "app": self.app_name,
+                    }
+                else:
+                    # If the room doesn't have a name, say who the messages
+                    # are from explicitly to avoid, "messages in the Bob room"
+                    sender_ids = list(set([
+                        notif_events[n['event_id']].sender
+                        for n in notifs_by_room[room_id]
+                    ]))
+
+                    return MESSAGES_FROM_PERSON % {
+                        "person": descriptor_from_member_events([
+                            state_by_room[room_id][("m.room.member", s)]
+                            for s in sender_ids
+                        ]),
+                        "app": self.app_name,
+                    }
+        else:
+            # Stuff's happened in multiple different rooms
+            return MESSAGES_IN_ROOMS % {
+                "app": self.app_name,
+            }
+
+    def make_room_link(self, room_id):
+        # need /beta for Universal Links to work on iOS
+        if self.app_name == "Vector":
+            return "https://vector.im/beta/#/room/%s" % (room_id,)
+        else:
+            return "https://matrix.to/#/%s" % (room_id,)
+
+    def make_notif_link(self, notif):
+        # need /beta for Universal Links to work on iOS
+        if self.app_name == "Vector":
+            return "https://vector.im/beta/#/room/%s/%s" % (
+                notif['room_id'], notif['event_id']
+            )
+        else:
+            return "https://matrix.to/#/%s/%s" % (
+                notif['room_id'], notif['event_id']
+            )
+
+    def make_unsubscribe_link(self):
+        # XXX: matrix.to
+        return "https://vector.im/#/settings"
+
+    def mxc_to_http_filter(self, value, width, height, resize_method="crop"):
+        if value[0:6] != "mxc://":
+            return ""
+
+        serverAndMediaId = value[6:]
+        fragment = None
+        if '#' in serverAndMediaId:
+            (serverAndMediaId, fragment) = serverAndMediaId.split('#', 1)
+            fragment = "#" + fragment
+
+        params = {
+            "width": width,
+            "height": height,
+            "method": resize_method,
+        }
+        return "%s_matrix/media/v1/thumbnail/%s?%s%s" % (
+            self.hs.config.public_baseurl,
+            serverAndMediaId,
+            urllib.urlencode(params),
+            fragment or "",
+        )
+
+
+def safe_markup(raw_html):
+    return jinja2.Markup(bleach.linkify(bleach.clean(
+        raw_html, tags=ALLOWED_TAGS, attributes=ALLOWED_ATTRS,
+        # bleach master has this, but it isn't released yet
+        # protocols=ALLOWED_SCHEMES,
+        strip=True
+    )))
+
+
+def safe_text(raw_text):
+    """
+    Process text: treat it as HTML but escape any tags (ie. just escape the
+    HTML) then linkify it.
+    """
+    return jinja2.Markup(bleach.linkify(bleach.clean(
+        raw_text, tags=[], attributes={},
+        strip=False
+    )))
+
+
+def deduped_ordered_list(l):
+    seen = set()
+    ret = []
+    for item in l:
+        if item not in seen:
+            seen.add(item)
+            ret.append(item)
+    return ret
+
+
+def string_ordinal_total(s):
+    tot = 0
+    for c in s:
+        tot += ord(c)
+    return tot
+
+
+def format_ts_filter(value, format):
+    return time.strftime(format, time.localtime(value / 1000))
diff --git a/synapse/push/pusher.py b/synapse/push/pusher.py
index 4960837504..e6c0806415 100644
--- a/synapse/push/pusher.py
+++ b/synapse/push/pusher.py
@@ -1,10 +1,37 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014-2016 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.
+
 from httppusher import HttpPusher
 
-PUSHER_TYPES = {
-    'http': HttpPusher
-}
+import logging
+logger = logging.getLogger(__name__)
 
 
 def create_pusher(hs, pusherdict):
+    logger.info("trying to create_pusher for %r", pusherdict)
+
+    PUSHER_TYPES = {
+        "http": HttpPusher,
+    }
+
+    logger.info("email enable notifs: %r", hs.config.email_enable_notifs)
+    if hs.config.email_enable_notifs:
+        from synapse.push.emailpusher import EmailPusher
+        PUSHER_TYPES["email"] = EmailPusher
+        logger.info("defined email pusher type")
+
     if pusherdict['kind'] in PUSHER_TYPES:
+        logger.info("found pusher")
         return PUSHER_TYPES[pusherdict['kind']](hs, pusherdict)
diff --git a/synapse/push/pusherpool.py b/synapse/push/pusherpool.py
index 6ef48d63f7..5853ec36a9 100644
--- a/synapse/push/pusherpool.py
+++ b/synapse/push/pusherpool.py
@@ -17,7 +17,6 @@
 from twisted.internet import defer
 
 import pusher
-from synapse.push import PusherConfigException
 from synapse.util.logcontext import preserve_fn
 from synapse.util.async import run_on_reactor
 
@@ -50,6 +49,7 @@ class PusherPool:
         # recreated, added and started: this means we have only one
         # code path adding pushers.
         pusher.create_pusher(self.hs, {
+            "id": None,
             "user_name": user_id,
             "kind": kind,
             "app_id": app_id,
@@ -185,8 +185,8 @@ class PusherPool:
         for pusherdict in pushers:
             try:
                 p = pusher.create_pusher(self.hs, pusherdict)
-            except PusherConfigException:
-                logger.exception("Couldn't start a pusher: caught PusherConfigException")
+            except:
+                logger.exception("Couldn't start a pusher: caught Exception")
                 continue
             if p:
                 appid_pushkey = "%s:%s" % (
diff --git a/synapse/python_dependencies.py b/synapse/python_dependencies.py
index 0eb3d6c1de..e0a7a19777 100644
--- a/synapse/python_dependencies.py
+++ b/synapse/python_dependencies.py
@@ -44,6 +44,10 @@ CONDITIONAL_REQUIREMENTS = {
     "preview_url": {
         "netaddr>=0.7.18": ["netaddr"],
     },
+    "email.enable_notifs": {
+        "Jinja2>=2.8": ["Jinja2>=2.8"],
+        "bleach>=1.4.2": ["bleach>=1.4.2"],
+    },
 }
 
 
diff --git a/synapse/replication/resource.py b/synapse/replication/resource.py
index ff78c60f13..0e983ae7fa 100644
--- a/synapse/replication/resource.py
+++ b/synapse/replication/resource.py
@@ -159,6 +159,15 @@ class ReplicationResource(Resource):
 
         result = yield self.notifier.wait_for_replication(replicate, timeout)
 
+        for stream_name, stream_content in result.items():
+            logger.info(
+                "Replicating %d rows of %s from %s -> %s",
+                len(stream_content["rows"]),
+                stream_name,
+                request_streams.get(stream_name),
+                stream_content["position"],
+            )
+
         request.write(json.dumps(result, ensure_ascii=False))
         finish_request(request)
 
diff --git a/synapse/replication/slave/storage/account_data.py b/synapse/replication/slave/storage/account_data.py
new file mode 100644
index 0000000000..f59b0eabbc
--- /dev/null
+++ b/synapse/replication/slave/storage/account_data.py
@@ -0,0 +1,61 @@
+# -*- coding: utf-8 -*-
+# Copyright 2016 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.
+
+from ._base import BaseSlavedStore
+from ._slaved_id_tracker import SlavedIdTracker
+from synapse.storage.account_data import AccountDataStore
+
+
+class SlavedAccountDataStore(BaseSlavedStore):
+
+    def __init__(self, db_conn, hs):
+        super(SlavedAccountDataStore, self).__init__(db_conn, hs)
+        self._account_data_id_gen = SlavedIdTracker(
+            db_conn, "account_data_max_stream_id", "stream_id",
+        )
+
+    get_global_account_data_by_type_for_users = (
+        AccountDataStore.__dict__["get_global_account_data_by_type_for_users"]
+    )
+
+    get_global_account_data_by_type_for_user = (
+        AccountDataStore.__dict__["get_global_account_data_by_type_for_user"]
+    )
+
+    def stream_positions(self):
+        result = super(SlavedAccountDataStore, self).stream_positions()
+        position = self._account_data_id_gen.get_current_token()
+        result["user_account_data"] = position
+        result["room_account_data"] = position
+        result["tag_account_data"] = position
+        return result
+
+    def process_replication(self, result):
+        stream = result.get("user_account_data")
+        if stream:
+            self._account_data_id_gen.advance(int(stream["position"]))
+            for row in stream["rows"]:
+                user_id, data_type = row[1:3]
+                self.get_global_account_data_by_type_for_user.invalidate(
+                    (data_type, user_id,)
+                )
+
+        stream = result.get("room_account_data")
+        if stream:
+            self._account_data_id_gen.advance(int(stream["position"]))
+
+        stream = result.get("tag_account_data")
+        if stream:
+            self._account_data_id_gen.advance(int(stream["position"]))
diff --git a/synapse/replication/slave/storage/events.py b/synapse/replication/slave/storage/events.py
index 86f00b6ff5..c0d741452d 100644
--- a/synapse/replication/slave/storage/events.py
+++ b/synapse/replication/slave/storage/events.py
@@ -75,6 +75,18 @@ class SlavedEventStore(BaseSlavedStore):
     get_unread_event_push_actions_by_room_for_user = (
         EventPushActionsStore.__dict__["get_unread_event_push_actions_by_room_for_user"]
     )
+    _get_state_group_for_events = (
+        StateStore.__dict__["_get_state_group_for_events"]
+    )
+    _get_state_group_for_event = (
+        StateStore.__dict__["_get_state_group_for_event"]
+    )
+    _get_state_groups_from_groups = (
+        StateStore.__dict__["_get_state_groups_from_groups"]
+    )
+    _get_state_group_from_group = (
+        StateStore.__dict__["_get_state_group_from_group"]
+    )
 
     get_unread_push_actions_for_user_in_range = (
         DataStore.get_unread_push_actions_for_user_in_range.__func__
@@ -83,6 +95,7 @@ class SlavedEventStore(BaseSlavedStore):
         DataStore.get_push_action_users_in_range.__func__
     )
     get_event = DataStore.get_event.__func__
+    get_events = DataStore.get_events.__func__
     get_current_state = DataStore.get_current_state.__func__
     get_current_state_for_key = DataStore.get_current_state_for_key.__func__
     get_rooms_for_user_where_membership_is = (
@@ -95,6 +108,9 @@ class SlavedEventStore(BaseSlavedStore):
     get_room_events_stream_for_room = (
         DataStore.get_room_events_stream_for_room.__func__
     )
+    get_events_around = DataStore.get_events_around.__func__
+    get_state_for_events = DataStore.get_state_for_events.__func__
+    get_state_groups = DataStore.get_state_groups.__func__
 
     _set_before_and_after = DataStore._set_before_and_after
 
@@ -104,6 +120,7 @@ class SlavedEventStore(BaseSlavedStore):
     _invalidate_get_event_cache = DataStore._invalidate_get_event_cache.__func__
     _parse_events_txn = DataStore._parse_events_txn.__func__
     _get_events_txn = DataStore._get_events_txn.__func__
+    _get_event_txn = DataStore._get_event_txn.__func__
     _enqueue_events = DataStore._enqueue_events.__func__
     _do_fetch = DataStore._do_fetch.__func__
     _fetch_events_txn = DataStore._fetch_events_txn.__func__
@@ -114,6 +131,10 @@ class SlavedEventStore(BaseSlavedStore):
         DataStore._get_rooms_for_user_where_membership_is_txn.__func__
     )
     _get_members_rows_txn = DataStore._get_members_rows_txn.__func__
+    _get_state_for_groups = DataStore._get_state_for_groups.__func__
+    _get_all_state_from_cache = DataStore._get_all_state_from_cache.__func__
+    _get_events_around_txn = DataStore._get_events_around_txn.__func__
+    _get_some_state_from_cache = DataStore._get_some_state_from_cache.__func__
 
     def stream_positions(self):
         result = super(SlavedEventStore, self).stream_positions()
@@ -128,7 +149,7 @@ class SlavedEventStore(BaseSlavedStore):
 
         stream = result.get("events")
         if stream:
-            self._stream_id_gen.advance(stream["position"])
+            self._stream_id_gen.advance(int(stream["position"]))
             for row in stream["rows"]:
                 self._process_replication_row(
                     row, backfilled=False, state_resets=state_resets
@@ -136,7 +157,7 @@ class SlavedEventStore(BaseSlavedStore):
 
         stream = result.get("backfill")
         if stream:
-            self._backfill_id_gen.advance(-stream["position"])
+            self._backfill_id_gen.advance(-int(stream["position"]))
             for row in stream["rows"]:
                 self._process_replication_row(
                     row, backfilled=True, state_resets=state_resets
@@ -144,12 +165,14 @@ class SlavedEventStore(BaseSlavedStore):
 
         stream = result.get("forward_ex_outliers")
         if stream:
+            self._stream_id_gen.advance(int(stream["position"]))
             for row in stream["rows"]:
                 event_id = row[1]
                 self._invalidate_get_event_cache(event_id)
 
         stream = result.get("backward_ex_outliers")
         if stream:
+            self._backfill_id_gen.advance(-int(stream["position"]))
             for row in stream["rows"]:
                 event_id = row[1]
                 self._invalidate_get_event_cache(event_id)
diff --git a/synapse/replication/slave/storage/pushers.py b/synapse/replication/slave/storage/pushers.py
index 8faddb2595..d88206b3bb 100644
--- a/synapse/replication/slave/storage/pushers.py
+++ b/synapse/replication/slave/storage/pushers.py
@@ -43,10 +43,10 @@ class SlavedPusherStore(BaseSlavedStore):
     def process_replication(self, result):
         stream = result.get("pushers")
         if stream:
-            self._pushers_id_gen.advance(stream["position"])
+            self._pushers_id_gen.advance(int(stream["position"]))
 
         stream = result.get("deleted_pushers")
         if stream:
-            self._pushers_id_gen.advance(stream["position"])
+            self._pushers_id_gen.advance(int(stream["position"]))
 
         return super(SlavedPusherStore, self).process_replication(result)
diff --git a/synapse/replication/slave/storage/receipts.py b/synapse/replication/slave/storage/receipts.py
index b55d5dfd08..ec007516d0 100644
--- a/synapse/replication/slave/storage/receipts.py
+++ b/synapse/replication/slave/storage/receipts.py
@@ -50,7 +50,7 @@ class SlavedReceiptsStore(BaseSlavedStore):
     def process_replication(self, result):
         stream = result.get("receipts")
         if stream:
-            self._receipts_id_gen.advance(stream["position"])
+            self._receipts_id_gen.advance(int(stream["position"]))
             for row in stream["rows"]:
                 room_id, receipt_type, user_id = row[1:4]
                 self.invalidate_caches_for_receipt(room_id, receipt_type, user_id)
diff --git a/synapse/rest/__init__.py b/synapse/rest/__init__.py
index 6688fa8fa0..8b223e032b 100644
--- a/synapse/rest/__init__.py
+++ b/synapse/rest/__init__.py
@@ -44,6 +44,8 @@ from synapse.rest.client.v2_alpha import (
     tokenrefresh,
     tags,
     account_data,
+    report_event,
+    openid,
 )
 
 from synapse.http.server import JsonResource
@@ -86,3 +88,5 @@ class ClientRestResource(JsonResource):
         tokenrefresh.register_servlets(hs, client_resource)
         tags.register_servlets(hs, client_resource)
         account_data.register_servlets(hs, client_resource)
+        report_event.register_servlets(hs, client_resource)
+        openid.register_servlets(hs, client_resource)
diff --git a/synapse/rest/client/v1/register.py b/synapse/rest/client/v1/register.py
index c6a2ef2ccc..e3f4fbb0bb 100644
--- a/synapse/rest/client/v1/register.py
+++ b/synapse/rest/client/v1/register.py
@@ -355,5 +355,76 @@ class RegisterRestServlet(ClientV1RestServlet):
             )
 
 
+class CreateUserRestServlet(ClientV1RestServlet):
+    """Handles user creation via a server-to-server interface
+    """
+
+    PATTERNS = client_path_patterns("/createUser$", releases=())
+
+    def __init__(self, hs):
+        super(CreateUserRestServlet, self).__init__(hs)
+        self.store = hs.get_datastore()
+        self.direct_user_creation_max_duration = hs.config.user_creation_max_duration
+
+    @defer.inlineCallbacks
+    def on_POST(self, request):
+        user_json = parse_json_object_from_request(request)
+
+        if "access_token" not in request.args:
+            raise SynapseError(400, "Expected application service token.")
+
+        app_service = yield self.store.get_app_service_by_token(
+            request.args["access_token"][0]
+        )
+        if not app_service:
+            raise SynapseError(403, "Invalid application service token.")
+
+        logger.debug("creating user: %s", user_json)
+
+        response = yield self._do_create(user_json)
+
+        defer.returnValue((200, response))
+
+    def on_OPTIONS(self, request):
+        return 403, {}
+
+    @defer.inlineCallbacks
+    def _do_create(self, user_json):
+        yield run_on_reactor()
+
+        if "localpart" not in user_json:
+            raise SynapseError(400, "Expected 'localpart' key.")
+
+        if "displayname" not in user_json:
+            raise SynapseError(400, "Expected 'displayname' key.")
+
+        if "duration_seconds" not in user_json:
+            raise SynapseError(400, "Expected 'duration_seconds' key.")
+
+        localpart = user_json["localpart"].encode("utf-8")
+        displayname = user_json["displayname"].encode("utf-8")
+        duration_seconds = 0
+        try:
+            duration_seconds = int(user_json["duration_seconds"])
+        except ValueError:
+            raise SynapseError(400, "Failed to parse 'duration_seconds'")
+        if duration_seconds > self.direct_user_creation_max_duration:
+            duration_seconds = self.direct_user_creation_max_duration
+
+        handler = self.handlers.registration_handler
+        user_id, token = yield handler.get_or_create_user(
+            localpart=localpart,
+            displayname=displayname,
+            duration_seconds=duration_seconds
+        )
+
+        defer.returnValue({
+            "user_id": user_id,
+            "access_token": token,
+            "home_server": self.hs.hostname,
+        })
+
+
 def register_servlets(hs, http_server):
     RegisterRestServlet(hs).register(http_server)
+    CreateUserRestServlet(hs).register(http_server)
diff --git a/synapse/rest/client/v2_alpha/openid.py b/synapse/rest/client/v2_alpha/openid.py
new file mode 100644
index 0000000000..aa1cae8e1e
--- /dev/null
+++ b/synapse/rest/client/v2_alpha/openid.py
@@ -0,0 +1,96 @@
+# -*- coding: utf-8 -*-
+# Copyright 2015, 2016 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.
+
+
+from ._base import client_v2_patterns
+
+from synapse.http.servlet import RestServlet, parse_json_object_from_request
+from synapse.api.errors import AuthError
+from synapse.util.stringutils import random_string
+
+from twisted.internet import defer
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class IdTokenServlet(RestServlet):
+    """
+    Get a bearer token that may be passed to a third party to confirm ownership
+    of a matrix user id.
+
+    The format of the response could be made compatible with the format given
+    in http://openid.net/specs/openid-connect-core-1_0.html#TokenResponse
+
+    But instead of returning a signed "id_token" the response contains the
+    name of the issuing matrix homeserver. This means that for now the third
+    party will need to check the validity of the "id_token" against the
+    federation /openid/userinfo endpoint of the homeserver.
+
+    Request:
+
+    POST /user/{user_id}/openid/request_token?access_token=... HTTP/1.1
+
+    {}
+
+    Response:
+
+    HTTP/1.1 200 OK
+    {
+        "access_token": "ABDEFGH",
+        "token_type": "Bearer",
+        "matrix_server_name": "example.com",
+        "expires_in": 3600,
+    }
+    """
+    PATTERNS = client_v2_patterns(
+        "/user/(?P<user_id>[^/]*)/openid/request_token"
+    )
+
+    EXPIRES_MS = 3600 * 1000
+
+    def __init__(self, hs):
+        super(IdTokenServlet, self).__init__()
+        self.auth = hs.get_auth()
+        self.store = hs.get_datastore()
+        self.clock = hs.get_clock()
+        self.server_name = hs.config.server_name
+
+    @defer.inlineCallbacks
+    def on_POST(self, request, user_id):
+        requester = yield self.auth.get_user_by_req(request)
+        if user_id != requester.user.to_string():
+            raise AuthError(403, "Cannot request tokens for other users.")
+
+        # Parse the request body to make sure it's JSON, but ignore the contents
+        # for now.
+        parse_json_object_from_request(request)
+
+        token = random_string(24)
+        ts_valid_until_ms = self.clock.time_msec() + self.EXPIRES_MS
+
+        yield self.store.insert_open_id_token(token, ts_valid_until_ms, user_id)
+
+        defer.returnValue((200, {
+            "access_token": token,
+            "token_type": "Bearer",
+            "matrix_server_name": self.server_name,
+            "expires_in": self.EXPIRES_MS / 1000,
+        }))
+
+
+def register_servlets(hs, http_server):
+    IdTokenServlet(hs).register(http_server)
diff --git a/synapse/rest/client/v2_alpha/register.py b/synapse/rest/client/v2_alpha/register.py
index ff8f69ddbf..1ecc02d94d 100644
--- a/synapse/rest/client/v2_alpha/register.py
+++ b/synapse/rest/client/v2_alpha/register.py
@@ -48,6 +48,7 @@ class RegisterRestServlet(RestServlet):
         super(RegisterRestServlet, self).__init__()
         self.hs = hs
         self.auth = hs.get_auth()
+        self.store = hs.get_datastore()
         self.auth_handler = hs.get_handlers().auth_handler
         self.registration_handler = hs.get_handlers().registration_handler
         self.identity_handler = hs.get_handlers().identity_handler
@@ -214,6 +215,34 @@ class RegisterRestServlet(RestServlet):
                         threepid['validated_at'],
                     )
 
+                    # And we add an email pusher for them by default, but only
+                    # if email notifications are enabled (so people don't start
+                    # getting mail spam where they weren't before if email
+                    # notifs are set up on a home server)
+                    if (
+                        self.hs.config.email_enable_notifs and
+                        self.hs.config.email_notif_for_new_users
+                    ):
+                        # Pull the ID of the access token back out of the db
+                        # It would really make more sense for this to be passed
+                        # up when the access token is saved, but that's quite an
+                        # invasive change I'd rather do separately.
+                        user_tuple = yield self.store.get_user_by_access_token(
+                            token
+                        )
+
+                        yield self.hs.get_pusherpool().add_pusher(
+                            user_id=user_id,
+                            access_token=user_tuple["token_id"],
+                            kind="email",
+                            app_id="m.email",
+                            app_display_name="Email Notifications",
+                            device_display_name=threepid["address"],
+                            pushkey=threepid["address"],
+                            lang=None,  # We don't know a user's language here
+                            data={},
+                        )
+
             if 'bind_email' in params and params['bind_email']:
                 logger.info("bind_email specified: binding")
 
diff --git a/synapse/rest/client/v2_alpha/report_event.py b/synapse/rest/client/v2_alpha/report_event.py
new file mode 100644
index 0000000000..8903e12405
--- /dev/null
+++ b/synapse/rest/client/v2_alpha/report_event.py
@@ -0,0 +1,59 @@
+# -*- coding: utf-8 -*-
+# Copyright 2016 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.
+
+from twisted.internet import defer
+
+from synapse.http.servlet import RestServlet, parse_json_object_from_request
+from ._base import client_v2_patterns
+
+import logging
+
+
+logger = logging.getLogger(__name__)
+
+
+class ReportEventRestServlet(RestServlet):
+    PATTERNS = client_v2_patterns(
+        "/rooms/(?P<room_id>[^/]*)/report/(?P<event_id>[^/]*)$"
+    )
+
+    def __init__(self, hs):
+        super(ReportEventRestServlet, self).__init__()
+        self.hs = hs
+        self.auth = hs.get_auth()
+        self.clock = hs.get_clock()
+        self.store = hs.get_datastore()
+
+    @defer.inlineCallbacks
+    def on_POST(self, request, room_id, event_id):
+        requester = yield self.auth.get_user_by_req(request)
+        user_id = requester.user.to_string()
+
+        body = parse_json_object_from_request(request)
+
+        yield self.store.add_event_report(
+            room_id=room_id,
+            event_id=event_id,
+            user_id=user_id,
+            reason=body.get("reason"),
+            content=body,
+            received_ts=self.clock.time_msec(),
+        )
+
+        defer.returnValue((200, {}))
+
+
+def register_servlets(hs, http_server):
+    ReportEventRestServlet(hs).register(http_server)
diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py
index 045ae6c03f..d970fde9e8 100644
--- a/synapse/storage/__init__.py
+++ b/synapse/storage/__init__.py
@@ -44,6 +44,7 @@ from .receipts import ReceiptsStore
 from .search import SearchStore
 from .tags import TagsStore
 from .account_data import AccountDataStore
+from .openid import OpenIdStore
 
 from .util.id_generators import IdGenerator, StreamIdGenerator, ChainedIdGenerator
 
@@ -81,7 +82,8 @@ class DataStore(RoomMemberStore, RoomStore,
                 SearchStore,
                 TagsStore,
                 AccountDataStore,
-                EventPushActionsStore
+                EventPushActionsStore,
+                OpenIdStore,
                 ):
 
     def __init__(self, db_conn, hs):
@@ -114,6 +116,7 @@ class DataStore(RoomMemberStore, RoomStore,
         self._state_groups_id_gen = StreamIdGenerator(db_conn, "state_groups", "id")
         self._access_tokens_id_gen = IdGenerator(db_conn, "access_tokens", "id")
         self._refresh_tokens_id_gen = IdGenerator(db_conn, "refresh_tokens", "id")
+        self._event_reports_id_gen = IdGenerator(db_conn, "event_reports", "id")
         self._push_rule_id_gen = IdGenerator(db_conn, "push_rules", "id")
         self._push_rules_enable_id_gen = IdGenerator(db_conn, "push_rules_enable", "id")
         self._push_rules_stream_id_gen = ChainedIdGenerator(
diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index 1e27c2c0ce..e0d7098692 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -453,7 +453,9 @@ class SQLBaseStore(object):
             keyvalues (dict): The unique key tables and their new values
             values (dict): The nonunique columns and their new values
             insertion_values (dict): key/values to use when inserting
-        Returns: A deferred
+        Returns:
+            Deferred(bool): True if a new entry was created, False if an
+                existing one was updated.
         """
         return self.runInteraction(
             desc,
@@ -498,6 +500,10 @@ class SQLBaseStore(object):
             )
             txn.execute(sql, allvalues.values())
 
+            return True
+        else:
+            return False
+
     def _simple_select_one(self, table, keyvalues, retcols,
                            allow_none=False, desc="_simple_select_one"):
         """Executes a SELECT query on the named table, which is expected to
diff --git a/synapse/storage/account_data.py b/synapse/storage/account_data.py
index 7a7fbf1e52..ec7e8d40d2 100644
--- a/synapse/storage/account_data.py
+++ b/synapse/storage/account_data.py
@@ -16,6 +16,8 @@
 from ._base import SQLBaseStore
 from twisted.internet import defer
 
+from synapse.util.caches.descriptors import cached, cachedList, cachedInlineCallbacks
+
 import ujson as json
 import logging
 
@@ -24,6 +26,7 @@ logger = logging.getLogger(__name__)
 
 class AccountDataStore(SQLBaseStore):
 
+    @cached()
     def get_account_data_for_user(self, user_id):
         """Get all the client account_data for a user.
 
@@ -60,6 +63,47 @@ class AccountDataStore(SQLBaseStore):
             "get_account_data_for_user", get_account_data_for_user_txn
         )
 
+    @cachedInlineCallbacks(num_args=2)
+    def get_global_account_data_by_type_for_user(self, data_type, user_id):
+        """
+        Returns:
+            Deferred: A dict
+        """
+        result = yield self._simple_select_one_onecol(
+            table="account_data",
+            keyvalues={
+                "user_id": user_id,
+                "account_data_type": data_type,
+            },
+            retcol="content",
+            desc="get_global_account_data_by_type_for_user",
+            allow_none=True,
+        )
+
+        if result:
+            defer.returnValue(json.loads(result))
+        else:
+            defer.returnValue(None)
+
+    @cachedList(cached_method_name="get_global_account_data_by_type_for_user",
+                num_args=2, list_name="user_ids", inlineCallbacks=True)
+    def get_global_account_data_by_type_for_users(self, data_type, user_ids):
+        rows = yield self._simple_select_many_batch(
+            table="account_data",
+            column="user_id",
+            iterable=user_ids,
+            keyvalues={
+                "account_data_type": data_type,
+            },
+            retcols=("user_id", "content",),
+            desc="get_global_account_data_by_type_for_users",
+        )
+
+        defer.returnValue({
+            row["user_id"]: json.loads(row["content"]) if row["content"] else None
+            for row in rows
+        })
+
     def get_account_data_for_room(self, user_id, room_id):
         """Get all the client account_data for a user for a room.
 
@@ -193,6 +237,7 @@ class AccountDataStore(SQLBaseStore):
                 self._account_data_stream_cache.entity_has_changed,
                 user_id, next_id,
             )
+            txn.call_after(self.get_account_data_for_user.invalidate, (user_id,))
             self._update_max_stream_id(txn, next_id)
 
         with self._account_data_id_gen.get_next() as next_id:
@@ -232,6 +277,11 @@ class AccountDataStore(SQLBaseStore):
                 self._account_data_stream_cache.entity_has_changed,
                 user_id, next_id,
             )
+            txn.call_after(self.get_account_data_for_user.invalidate, (user_id,))
+            txn.call_after(
+                self.get_global_account_data_by_type_for_user.invalidate,
+                (account_data_type, user_id,)
+            )
             self._update_max_stream_id(txn, next_id)
 
         with self._account_data_id_gen.get_next() as next_id:
diff --git a/synapse/storage/event_push_actions.py b/synapse/storage/event_push_actions.py
index 86a98b6f11..9705db5c47 100644
--- a/synapse/storage/event_push_actions.py
+++ b/synapse/storage/event_push_actions.py
@@ -118,16 +118,19 @@ class EventPushActionsStore(SQLBaseStore):
                                                   max_stream_ordering=None):
         def get_after_receipt(txn):
             sql = (
-                "SELECT ep.event_id, ep.stream_ordering, ep.actions "
-                "FROM event_push_actions AS ep, ("
-                "   SELECT room_id, user_id,"
-                "       max(topological_ordering) as topological_ordering,"
-                "       max(stream_ordering) as stream_ordering"
+                "SELECT ep.event_id, ep.room_id, ep.stream_ordering, ep.actions, "
+                "e.received_ts "
+                "FROM ("
+                "   SELECT room_id, user_id, "
+                "       max(topological_ordering) as topological_ordering, "
+                "       max(stream_ordering) as stream_ordering "
                 "       FROM events"
                 "   NATURAL JOIN receipts_linearized WHERE receipt_type = 'm.read'"
                 "   GROUP BY room_id, user_id"
-                ") AS rl "
-                "WHERE"
+                ") AS rl,"
+                " event_push_actions AS ep"
+                " INNER JOIN events AS e USING (room_id, event_id)"
+                " WHERE"
                 "   ep.room_id = rl.room_id"
                 "   AND ("
                 "       ep.topological_ordering > rl.topological_ordering"
@@ -153,11 +156,13 @@ class EventPushActionsStore(SQLBaseStore):
 
         def get_no_receipt(txn):
             sql = (
-                "SELECT ep.event_id, ep.stream_ordering, ep.actions "
-                "FROM event_push_actions AS ep "
-                "WHERE ep.room_id not in ("
+                "SELECT ep.event_id, ep.room_id, ep.stream_ordering, ep.actions,"
+                " e.received_ts"
+                " FROM event_push_actions AS ep"
+                " JOIN events e ON ep.room_id = e.room_id AND ep.event_id = e.event_id"
+                " WHERE ep.room_id not in ("
                 "   SELECT room_id FROM events NATURAL JOIN receipts_linearized"
-                "   WHERE receipt_type = 'm.read' AND user_id = ? "
+                "   WHERE receipt_type = 'm.read' AND user_id = ?"
                 "   GROUP BY room_id"
                 ") AND ep.user_id = ? AND ep.stream_ordering > ?"
             )
@@ -175,12 +180,30 @@ class EventPushActionsStore(SQLBaseStore):
         defer.returnValue([
             {
                 "event_id": row[0],
-                "stream_ordering": row[1],
-                "actions": json.loads(row[2]),
+                "room_id": row[1],
+                "stream_ordering": row[2],
+                "actions": json.loads(row[3]),
+                "received_ts": row[4],
             } for row in after_read_receipt + no_read_receipt
         ])
 
     @defer.inlineCallbacks
+    def get_time_of_last_push_action_before(self, stream_ordering):
+        def f(txn):
+            sql = (
+                "SELECT e.received_ts"
+                " FROM event_push_actions AS ep"
+                " JOIN events e ON ep.room_id = e.room_id AND ep.event_id = e.event_id"
+                " WHERE ep.stream_ordering > ?"
+                " ORDER BY ep.stream_ordering ASC"
+                " LIMIT 1"
+            )
+            txn.execute(sql, (stream_ordering,))
+            return txn.fetchone()
+        result = yield self.runInteraction("get_time_of_last_push_action_before", f)
+        defer.returnValue(result[0] if result else None)
+
+    @defer.inlineCallbacks
     def get_latest_push_action_stream_ordering(self):
         def f(txn):
             txn.execute("SELECT MAX(stream_ordering) FROM event_push_actions")
@@ -201,6 +224,18 @@ class EventPushActionsStore(SQLBaseStore):
             (room_id, event_id)
         )
 
+    def _remove_push_actions_before_txn(self, txn, room_id, user_id,
+                                        topological_ordering):
+        txn.call_after(
+            self.get_unread_event_push_actions_by_room_for_user.invalidate_many,
+            (room_id, user_id, )
+        )
+        txn.execute(
+            "DELETE FROM event_push_actions"
+            " WHERE room_id = ? AND user_id = ? AND topological_ordering < ?",
+            (room_id, user_id, topological_ordering,)
+        )
+
 
 def _action_has_highlight(actions):
     for action in actions:
diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index 0307b2af3c..4655669ba0 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -19,12 +19,14 @@ from twisted.internet import defer, reactor
 from synapse.events import FrozenEvent, USE_FROZEN_DICTS
 from synapse.events.utils import prune_event
 
+from synapse.util.async import ObservableDeferred
 from synapse.util.logcontext import preserve_fn, PreserveLoggingContext
 from synapse.util.logutils import log_function
 from synapse.api.constants import EventTypes
 
 from canonicaljson import encode_canonical_json
-from collections import namedtuple
+from collections import deque, namedtuple
+
 
 import logging
 import math
@@ -50,28 +52,169 @@ EVENT_QUEUE_ITERATIONS = 3  # No. times we block waiting for requests for events
 EVENT_QUEUE_TIMEOUT_S = 0.1  # Timeout when waiting for requests for events
 
 
+class _EventPeristenceQueue(object):
+    """Queues up events so that they can be persisted in bulk with only one
+    concurrent transaction per room.
+    """
+
+    _EventPersistQueueItem = namedtuple("_EventPersistQueueItem", (
+        "events_and_contexts", "current_state", "backfilled", "deferred",
+    ))
+
+    def __init__(self):
+        self._event_persist_queues = {}
+        self._currently_persisting_rooms = set()
+
+    def add_to_queue(self, room_id, events_and_contexts, backfilled, current_state):
+        """Add events to the queue, with the given persist_event options.
+        """
+        queue = self._event_persist_queues.setdefault(room_id, deque())
+        if queue:
+            end_item = queue[-1]
+            if end_item.current_state or current_state:
+                # We perist events with current_state set to True one at a time
+                pass
+            if end_item.backfilled == backfilled:
+                end_item.events_and_contexts.extend(events_and_contexts)
+                return end_item.deferred.observe()
+
+        deferred = ObservableDeferred(defer.Deferred())
+
+        queue.append(self._EventPersistQueueItem(
+            events_and_contexts=events_and_contexts,
+            backfilled=backfilled,
+            current_state=current_state,
+            deferred=deferred,
+        ))
+
+        return deferred.observe()
+
+    def handle_queue(self, room_id, per_item_callback):
+        """Attempts to handle the queue for a room if not already being handled.
+
+        The given callback will be invoked with for each item in the queue,1
+        of type _EventPersistQueueItem. The per_item_callback will continuously
+        be called with new items, unless the queue becomnes empty. The return
+        value of the function will be given to the deferreds waiting on the item,
+        exceptions will be passed to the deferres as well.
+
+        This function should therefore be called whenever anything is added
+        to the queue.
+
+        If another callback is currently handling the queue then it will not be
+        invoked.
+        """
+
+        if room_id in self._currently_persisting_rooms:
+            return
+
+        self._currently_persisting_rooms.add(room_id)
+
+        @defer.inlineCallbacks
+        def handle_queue_loop():
+            try:
+                queue = self._get_drainining_queue(room_id)
+                for item in queue:
+                    try:
+                        ret = yield per_item_callback(item)
+                        item.deferred.callback(ret)
+                    except Exception as e:
+                        item.deferred.errback(e)
+            finally:
+                queue = self._event_persist_queues.pop(room_id, None)
+                if queue:
+                    self._event_persist_queues[room_id] = queue
+                self._currently_persisting_rooms.discard(room_id)
+
+        preserve_fn(handle_queue_loop)()
+
+    def _get_drainining_queue(self, room_id):
+        queue = self._event_persist_queues.setdefault(room_id, deque())
+
+        try:
+            while True:
+                yield queue.popleft()
+        except IndexError:
+            # Queue has been drained.
+            pass
+
+
 class EventsStore(SQLBaseStore):
     EVENT_ORIGIN_SERVER_TS_NAME = "event_origin_server_ts"
 
     def __init__(self, hs):
         super(EventsStore, self).__init__(hs)
+        self._clock = hs.get_clock()
         self.register_background_update_handler(
             self.EVENT_ORIGIN_SERVER_TS_NAME, self._background_reindex_origin_server_ts
         )
 
-    @defer.inlineCallbacks
+        self._event_persist_queue = _EventPeristenceQueue()
+
     def persist_events(self, events_and_contexts, backfilled=False):
         """
         Write events to the database
         Args:
             events_and_contexts: list of tuples of (event, context)
             backfilled: ?
+        """
+        partitioned = {}
+        for event, ctx in events_and_contexts:
+            partitioned.setdefault(event.room_id, []).append((event, ctx))
+
+        deferreds = []
+        for room_id, evs_ctxs in partitioned.items():
+            d = self._event_persist_queue.add_to_queue(
+                room_id, evs_ctxs,
+                backfilled=backfilled,
+                current_state=None,
+            )
+            deferreds.append(d)
 
-        Returns: Tuple of stream_orderings where the first is the minimum and
-            last is the maximum stream ordering assigned to the events when
-            persisting.
+        for room_id in partitioned.keys():
+            self._maybe_start_persisting(room_id)
 
-        """
+        return defer.gatherResults(deferreds, consumeErrors=True)
+
+    @defer.inlineCallbacks
+    @log_function
+    def persist_event(self, event, context, current_state=None, backfilled=False):
+        deferred = self._event_persist_queue.add_to_queue(
+            event.room_id, [(event, context)],
+            backfilled=backfilled,
+            current_state=current_state,
+        )
+
+        self._maybe_start_persisting(event.room_id)
+
+        yield deferred
+
+        max_persisted_id = yield self._stream_id_gen.get_current_token()
+        defer.returnValue((event.internal_metadata.stream_ordering, max_persisted_id))
+
+    def _maybe_start_persisting(self, room_id):
+        @defer.inlineCallbacks
+        def persisting_queue(item):
+            if item.current_state:
+                for event, context in item.events_and_contexts:
+                    # There should only ever be one item in
+                    # events_and_contexts when current_state is
+                    # not None
+                    yield self._persist_event(
+                        event, context,
+                        current_state=item.current_state,
+                        backfilled=item.backfilled,
+                    )
+            else:
+                yield self._persist_events(
+                    item.events_and_contexts,
+                    backfilled=item.backfilled,
+                )
+
+        self._event_persist_queue.handle_queue(room_id, persisting_queue)
+
+    @defer.inlineCallbacks
+    def _persist_events(self, events_and_contexts, backfilled=False):
         if not events_and_contexts:
             return
 
@@ -118,8 +261,7 @@ class EventsStore(SQLBaseStore):
 
     @defer.inlineCallbacks
     @log_function
-    def persist_event(self, event, context, current_state=None, backfilled=False):
-
+    def _persist_event(self, event, context, current_state=None, backfilled=False):
         try:
             with self._stream_id_gen.get_next() as stream_ordering:
                 with self._state_groups_id_gen.get_next() as state_group_id:
@@ -136,9 +278,6 @@ class EventsStore(SQLBaseStore):
         except _RollbackButIsFineException:
             pass
 
-        max_persisted_id = yield self._stream_id_gen.get_current_token()
-        defer.returnValue((stream_ordering, max_persisted_id))
-
     @defer.inlineCallbacks
     def get_event(self, event_id, check_redacted=True,
                   get_prev_content=False, allow_rejected=False,
@@ -427,6 +566,7 @@ class EventsStore(SQLBaseStore):
                     "outlier": event.internal_metadata.is_outlier(),
                     "content": encode_json(event.content).decode("UTF-8"),
                     "origin_server_ts": int(event.origin_server_ts),
+                    "received_ts": self._clock.time_msec(),
                 }
                 for event, _ in events_and_contexts
             ],
diff --git a/synapse/storage/openid.py b/synapse/storage/openid.py
new file mode 100644
index 0000000000..5dabb607bd
--- /dev/null
+++ b/synapse/storage/openid.py
@@ -0,0 +1,32 @@
+from ._base import SQLBaseStore
+
+
+class OpenIdStore(SQLBaseStore):
+    def insert_open_id_token(self, token, ts_valid_until_ms, user_id):
+        return self._simple_insert(
+            table="open_id_tokens",
+            values={
+                "token": token,
+                "ts_valid_until_ms": ts_valid_until_ms,
+                "user_id": user_id,
+            },
+            desc="insert_open_id_token"
+        )
+
+    def get_user_id_for_open_id_token(self, token, ts_now_ms):
+        def get_user_id_for_token_txn(txn):
+            sql = (
+                "SELECT user_id FROM open_id_tokens"
+                " WHERE token = ? AND ? <= ts_valid_until_ms"
+            )
+
+            txn.execute(sql, (token, ts_now_ms))
+
+            rows = txn.fetchall()
+            if not rows:
+                return None
+            else:
+                return rows[0][0]
+        return self.runInteraction(
+            "get_user_id_for_token", get_user_id_for_token_txn
+        )
diff --git a/synapse/storage/prepare_database.py b/synapse/storage/prepare_database.py
index 57f14fd12b..c8487c8838 100644
--- a/synapse/storage/prepare_database.py
+++ b/synapse/storage/prepare_database.py
@@ -25,7 +25,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 = 31
+SCHEMA_VERSION = 32
 
 dir_path = os.path.abspath(os.path.dirname(__file__))
 
diff --git a/synapse/storage/pusher.py b/synapse/storage/pusher.py
index 11feb3eb11..9e8e2e2964 100644
--- a/synapse/storage/pusher.py
+++ b/synapse/storage/pusher.py
@@ -156,8 +156,7 @@ class PusherStore(SQLBaseStore):
                    profile_tag=""):
         with self._pushers_id_gen.get_next() as stream_id:
             def f(txn):
-                txn.call_after(self.get_users_with_pushers_in_room.invalidate_all)
-                return self._simple_upsert_txn(
+                newly_inserted = self._simple_upsert_txn(
                     txn,
                     "pushers",
                     {
@@ -178,11 +177,18 @@ class PusherStore(SQLBaseStore):
                         "id": stream_id,
                     },
                 )
-        defer.returnValue((yield self.runInteraction("add_pusher", f)))
+                if newly_inserted:
+                    # get_users_with_pushers_in_room only cares if the user has
+                    # at least *one* pusher.
+                    txn.call_after(self.get_users_with_pushers_in_room.invalidate_all)
+
+            yield self.runInteraction("add_pusher", f)
 
     @defer.inlineCallbacks
     def delete_pusher_by_app_id_pushkey_user_id(self, app_id, pushkey, user_id):
         def delete_pusher_txn(txn, stream_id):
+            txn.call_after(self.get_users_with_pushers_in_room.invalidate_all)
+
             self._simple_delete_one_txn(
                 txn,
                 "pushers",
@@ -194,6 +200,7 @@ class PusherStore(SQLBaseStore):
                 {"app_id": app_id, "pushkey": pushkey, "user_id": user_id},
                 {"stream_id": stream_id},
             )
+
         with self._pushers_id_gen.get_next() as stream_id:
             yield self.runInteraction(
                 "delete_pusher", delete_pusher_txn, stream_id
@@ -233,3 +240,30 @@ class PusherStore(SQLBaseStore):
             {'failing_since': failing_since},
             desc="update_pusher_failing_since",
         )
+
+    @defer.inlineCallbacks
+    def get_throttle_params_by_room(self, pusher_id):
+        res = yield self._simple_select_list(
+            "pusher_throttle",
+            {"pusher": pusher_id},
+            ["room_id", "last_sent_ts", "throttle_ms"],
+            desc="get_throttle_params_by_room"
+        )
+
+        params_by_room = {}
+        for row in res:
+            params_by_room[row["room_id"]] = {
+                "last_sent_ts": row["last_sent_ts"],
+                "throttle_ms": row["throttle_ms"]
+            }
+
+        defer.returnValue(params_by_room)
+
+    @defer.inlineCallbacks
+    def set_throttle_params(self, pusher_id, room_id, params):
+        yield self._simple_upsert(
+            "pusher_throttle",
+            {"pusher": pusher_id, "room_id": room_id},
+            params,
+            desc="set_throttle_params"
+        )
diff --git a/synapse/storage/receipts.py b/synapse/storage/receipts.py
index 935fc503d9..fdcf28f3e1 100644
--- a/synapse/storage/receipts.py
+++ b/synapse/storage/receipts.py
@@ -100,7 +100,7 @@ class ReceiptsStore(SQLBaseStore):
 
         defer.returnValue([ev for res in results.values() for ev in res])
 
-    @cachedInlineCallbacks(num_args=3, max_entries=5000)
+    @cachedInlineCallbacks(num_args=3, max_entries=5000, lru=True, tree=True)
     def get_linearized_receipts_for_room(self, room_id, to_key, from_key=None):
         """Get receipts for a single room for sending to clients.
 
@@ -232,7 +232,7 @@ class ReceiptsStore(SQLBaseStore):
             self.get_receipts_for_user.invalidate, (user_id, receipt_type)
         )
         # FIXME: This shouldn't invalidate the whole cache
-        txn.call_after(self.get_linearized_receipts_for_room.invalidate_all)
+        txn.call_after(self.get_linearized_receipts_for_room.invalidate_many, (room_id,))
 
         txn.call_after(
             self._receipts_stream_cache.entity_has_changed,
@@ -244,6 +244,17 @@ class ReceiptsStore(SQLBaseStore):
             (user_id, room_id, receipt_type)
         )
 
+        res = self._simple_select_one_txn(
+            txn,
+            table="events",
+            retcols=["topological_ordering", "stream_ordering"],
+            keyvalues={"event_id": event_id},
+            allow_none=True
+        )
+
+        topological_ordering = int(res["topological_ordering"]) if res else None
+        stream_ordering = int(res["stream_ordering"]) if res else None
+
         # We don't want to clobber receipts for more recent events, so we
         # have to compare orderings of existing receipts
         sql = (
@@ -255,16 +266,7 @@ class ReceiptsStore(SQLBaseStore):
         txn.execute(sql, (room_id, receipt_type, user_id))
         results = txn.fetchall()
 
-        if results:
-            res = self._simple_select_one_txn(
-                txn,
-                table="events",
-                retcols=["topological_ordering", "stream_ordering"],
-                keyvalues={"event_id": event_id},
-            )
-            topological_ordering = int(res["topological_ordering"])
-            stream_ordering = int(res["stream_ordering"])
-
+        if results and topological_ordering:
             for to, so, _ in results:
                 if int(to) > topological_ordering:
                     return False
@@ -294,6 +296,14 @@ class ReceiptsStore(SQLBaseStore):
             }
         )
 
+        if receipt_type == "m.read" and topological_ordering:
+            self._remove_push_actions_before_txn(
+                txn,
+                room_id=room_id,
+                user_id=user_id,
+                topological_ordering=topological_ordering,
+            )
+
         return True
 
     @defer.inlineCallbacks
@@ -367,7 +377,7 @@ class ReceiptsStore(SQLBaseStore):
             self.get_receipts_for_user.invalidate, (user_id, receipt_type)
         )
         # FIXME: This shouldn't invalidate the whole cache
-        txn.call_after(self.get_linearized_receipts_for_room.invalidate_all)
+        txn.call_after(self.get_linearized_receipts_for_room.invalidate_many, (room_id,))
 
         self._simple_delete_txn(
             txn,
diff --git a/synapse/storage/registration.py b/synapse/storage/registration.py
index 7af0cae6a5..bda84a744a 100644
--- a/synapse/storage/registration.py
+++ b/synapse/storage/registration.py
@@ -101,6 +101,7 @@ class RegistrationStore(SQLBaseStore):
             make_guest,
             appservice_id
         )
+        self.get_user_by_id.invalidate((user_id,))
         self.is_guest.invalidate((user_id,))
 
     def _register(
@@ -156,6 +157,7 @@ class RegistrationStore(SQLBaseStore):
                 (next_id, user_id, token,)
             )
 
+    @cached()
     def get_user_by_id(self, user_id):
         return self._simple_select_one(
             table="users",
@@ -193,6 +195,7 @@ class RegistrationStore(SQLBaseStore):
         }, {
             'password_hash': password_hash
         })
+        self.get_user_by_id.invalidate((user_id,))
 
     @defer.inlineCallbacks
     def user_delete_access_tokens(self, user_id, except_token_ids=[]):
diff --git a/synapse/storage/room.py b/synapse/storage/room.py
index 70aa64fb31..26933e593a 100644
--- a/synapse/storage/room.py
+++ b/synapse/storage/room.py
@@ -23,6 +23,7 @@ from .engines import PostgresEngine, Sqlite3Engine
 
 import collections
 import logging
+import ujson as json
 
 logger = logging.getLogger(__name__)
 
@@ -221,3 +222,20 @@ class RoomStore(SQLBaseStore):
                     aliases.extend(e.content['aliases'])
 
         defer.returnValue((name, aliases))
+
+    def add_event_report(self, room_id, event_id, user_id, reason, content,
+                         received_ts):
+        next_id = self._event_reports_id_gen.get_next()
+        return self._simple_insert(
+            table="event_reports",
+            values={
+                "id": next_id,
+                "received_ts": received_ts,
+                "room_id": room_id,
+                "event_id": event_id,
+                "user_id": user_id,
+                "reason": reason,
+                "content": json.dumps(content),
+            },
+            desc="add_event_report"
+        )
diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py
index 08a54cbdd1..9d6bfd5245 100644
--- a/synapse/storage/roommember.py
+++ b/synapse/storage/roommember.py
@@ -21,7 +21,7 @@ from ._base import SQLBaseStore
 from synapse.util.caches.descriptors import cached, cachedInlineCallbacks
 
 from synapse.api.constants import Membership
-from synapse.types import UserID
+from synapse.types import get_domian_from_id
 
 import logging
 
@@ -273,10 +273,7 @@ class RoomMemberStore(SQLBaseStore):
             room_id, membership=Membership.JOIN
         )
 
-        joined_domains = set(
-            UserID.from_string(r["user_id"]).domain
-            for r in rows
-        )
+        joined_domains = set(get_domian_from_id(r["user_id"]) for r in rows)
 
         return joined_domains
 
diff --git a/synapse/storage/schema/delta/32/events.sql b/synapse/storage/schema/delta/32/events.sql
new file mode 100644
index 0000000000..1dd0f9e170
--- /dev/null
+++ b/synapse/storage/schema/delta/32/events.sql
@@ -0,0 +1,16 @@
+/* Copyright 2016 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.
+ */
+
+ALTER TABLE events ADD COLUMN received_ts BIGINT;
diff --git a/synapse/storage/schema/delta/32/openid.sql b/synapse/storage/schema/delta/32/openid.sql
new file mode 100644
index 0000000000..36f37b11c8
--- /dev/null
+++ b/synapse/storage/schema/delta/32/openid.sql
@@ -0,0 +1,9 @@
+
+CREATE TABLE open_id_tokens (
+    token TEXT NOT NULL PRIMARY KEY,
+    ts_valid_until_ms bigint NOT NULL,
+    user_id TEXT NOT NULL,
+    UNIQUE (token)
+);
+
+CREATE index open_id_tokens_ts_valid_until_ms ON open_id_tokens(ts_valid_until_ms);
diff --git a/synapse/storage/schema/delta/32/pusher_throttle.sql b/synapse/storage/schema/delta/32/pusher_throttle.sql
new file mode 100644
index 0000000000..d86d30c13c
--- /dev/null
+++ b/synapse/storage/schema/delta/32/pusher_throttle.sql
@@ -0,0 +1,23 @@
+/* Copyright 2016 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.
+ */
+
+
+CREATE TABLE pusher_throttle(
+    pusher BIGINT NOT NULL,
+    room_id TEXT NOT NULL,
+    last_sent_ts BIGINT,
+    throttle_ms BIGINT,
+    PRIMARY KEY (pusher, room_id)
+);
diff --git a/synapse/storage/schema/delta/32/remove_indices.sql b/synapse/storage/schema/delta/32/remove_indices.sql
new file mode 100644
index 0000000000..f859be46a6
--- /dev/null
+++ b/synapse/storage/schema/delta/32/remove_indices.sql
@@ -0,0 +1,38 @@
+/* Copyright 2016 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.
+ */
+
+
+-- The following indices are redundant, other indices are equivalent or
+-- supersets
+DROP INDEX IF EXISTS events_room_id; -- Prefix of events_room_stream
+DROP INDEX IF EXISTS events_order; -- Prefix of events_order_topo_stream_room
+DROP INDEX IF EXISTS events_topological_ordering; -- Prefix of events_order_topo_stream_room
+DROP INDEX IF EXISTS events_stream_ordering; -- Duplicate of PRIMARY KEY
+DROP INDEX IF EXISTS state_groups_id; -- Duplicate of PRIMARY KEY
+DROP INDEX IF EXISTS event_to_state_groups_id; -- Duplicate of PRIMARY KEY
+DROP INDEX IF EXISTS event_push_actions_room_id_event_id_user_id_profile_tag; -- Duplicate of UNIQUE CONSTRAINT
+
+DROP INDEX IF EXISTS event_destinations_id; -- Prefix of UNIQUE CONSTRAINT
+DROP INDEX IF EXISTS st_extrem_id; -- Prefix of UNIQUE CONSTRAINT
+DROP INDEX IF EXISTS event_content_hashes_id; -- Prefix of UNIQUE CONSTRAINT
+DROP INDEX IF EXISTS event_signatures_id; -- Prefix of UNIQUE CONSTRAINT
+DROP INDEX IF EXISTS event_edge_hashes_id; -- Prefix of UNIQUE CONSTRAINT
+DROP INDEX IF EXISTS redactions_event_id; -- Duplicate of UNIQUE CONSTRAINT
+DROP INDEX IF EXISTS room_hosts_room_id; -- Prefix of UNIQUE CONSTRAINT
+
+-- The following indices were unused
+DROP INDEX IF EXISTS remote_media_cache_thumbnails_media_id;
+DROP INDEX IF EXISTS evauth_edges_auth_id;
+DROP INDEX IF EXISTS presence_stream_state;
diff --git a/synapse/storage/schema/delta/32/reports.sql b/synapse/storage/schema/delta/32/reports.sql
new file mode 100644
index 0000000000..d13609776f
--- /dev/null
+++ b/synapse/storage/schema/delta/32/reports.sql
@@ -0,0 +1,25 @@
+/* Copyright 2016 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.
+ */
+
+
+CREATE TABLE event_reports(
+    id BIGINT NOT NULL PRIMARY KEY,
+    received_ts BIGINT NOT NULL,
+    room_id TEXT NOT NULL,
+    event_id TEXT NOT NULL,
+    user_id TEXT NOT NULL,
+    reason TEXT,
+    content TEXT
+);
diff --git a/synapse/storage/transactions.py b/synapse/storage/transactions.py
index d338dfcf0a..6c7481a728 100644
--- a/synapse/storage/transactions.py
+++ b/synapse/storage/transactions.py
@@ -16,16 +16,56 @@
 from ._base import SQLBaseStore
 from synapse.util.caches.descriptors import cached
 
+from twisted.internet import defer, reactor
+
 from canonicaljson import encode_canonical_json
+
+from collections import namedtuple
+
+import itertools
 import logging
 
 logger = logging.getLogger(__name__)
 
 
+_TransactionRow = namedtuple(
+    "_TransactionRow", (
+        "id", "transaction_id", "destination", "ts", "response_code",
+        "response_json",
+    )
+)
+
+_UpdateTransactionRow = namedtuple(
+    "_TransactionRow", (
+        "response_code", "response_json",
+    )
+)
+
+
 class TransactionStore(SQLBaseStore):
     """A collection of queries for handling PDUs.
     """
 
+    def __init__(self, hs):
+        super(TransactionStore, self).__init__(hs)
+
+        # New transactions that are currently in flights
+        self.inflight_transactions = {}
+
+        # Newly delievered transactions that *weren't* persisted while in flight
+        self.new_delivered_transactions = {}
+
+        # Newly delivered transactions that *were* persisted while in flight
+        self.update_delivered_transactions = {}
+
+        self.last_transaction = {}
+
+        reactor.addSystemEventTrigger("before", "shutdown", self._persist_in_mem_txns)
+        hs.get_clock().looping_call(
+            self._persist_in_mem_txns,
+            1000,
+        )
+
     def get_received_txn_response(self, transaction_id, origin):
         """For an incoming transaction from a given origin, check if we have
         already responded to it. If so, return the response code and response
@@ -108,17 +148,30 @@ class TransactionStore(SQLBaseStore):
             list: A list of previous transaction ids.
         """
 
-        return self.runInteraction(
-            "prep_send_transaction",
-            self._prep_send_transaction,
-            transaction_id, destination, origin_server_ts
+        auto_id = self._transaction_id_gen.get_next()
+
+        txn_row = _TransactionRow(
+            id=auto_id,
+            transaction_id=transaction_id,
+            destination=destination,
+            ts=origin_server_ts,
+            response_code=0,
+            response_json=None,
         )
 
-    def _prep_send_transaction(self, txn, transaction_id, destination,
-                               origin_server_ts):
+        self.inflight_transactions.setdefault(destination, {})[transaction_id] = txn_row
 
-        next_id = self._transaction_id_gen.get_next()
+        prev_txn = self.last_transaction.get(destination)
+        if prev_txn:
+            return defer.succeed(prev_txn)
+        else:
+            return self.runInteraction(
+                "_get_prevs_txn",
+                self._get_prevs_txn,
+                destination,
+            )
 
+    def _get_prevs_txn(self, txn, destination):
         # First we find out what the prev_txns should be.
         # Since we know that we are only sending one transaction at a time,
         # we can simply take the last one.
@@ -133,23 +186,6 @@ class TransactionStore(SQLBaseStore):
 
         prev_txns = [r["transaction_id"] for r in results]
 
-        # Actually add the new transaction to the sent_transactions table.
-
-        self._simple_insert_txn(
-            txn,
-            table="sent_transactions",
-            values={
-                "id": next_id,
-                "transaction_id": transaction_id,
-                "destination": destination,
-                "ts": origin_server_ts,
-                "response_code": 0,
-                "response_json": None,
-            }
-        )
-
-        # TODO Update the tx id -> pdu id mapping
-
         return prev_txns
 
     def delivered_txn(self, transaction_id, destination, code, response_dict):
@@ -161,27 +197,23 @@ class TransactionStore(SQLBaseStore):
             code (int)
             response_json (str)
         """
-        return self.runInteraction(
-            "delivered_txn",
-            self._delivered_txn,
-            transaction_id, destination, code,
-            buffer(encode_canonical_json(response_dict)),
-        )
 
-    def _delivered_txn(self, txn, transaction_id, destination,
-                       code, response_json):
-        self._simple_update_one_txn(
-            txn,
-            table="sent_transactions",
-            keyvalues={
-                "transaction_id": transaction_id,
-                "destination": destination,
-            },
-            updatevalues={
-                "response_code": code,
-                "response_json": None,  # For now, don't persist response_json
-            }
-        )
+        txn_row = self.inflight_transactions.get(
+            destination, {}
+        ).pop(transaction_id, None)
+
+        self.last_transaction[destination] = transaction_id
+
+        if txn_row:
+            d = self.new_delivered_transactions.setdefault(destination, {})
+            d[transaction_id] = txn_row._replace(
+                response_code=code,
+                response_json=None,  # For now, don't persist response
+            )
+        else:
+            d = self.update_delivered_transactions.setdefault(destination, {})
+            # For now, don't persist response
+            d[transaction_id] = _UpdateTransactionRow(code, None)
 
     def get_transactions_after(self, transaction_id, destination):
         """Get all transactions after a given local transaction_id.
@@ -305,3 +337,48 @@ class TransactionStore(SQLBaseStore):
 
         txn.execute(query, (self._clock.time_msec(),))
         return self.cursor_to_dict(txn)
+
+    @defer.inlineCallbacks
+    def _persist_in_mem_txns(self):
+        try:
+            inflight = self.inflight_transactions
+            new_delivered = self.new_delivered_transactions
+            update_delivered = self.update_delivered_transactions
+
+            self.inflight_transactions = {}
+            self.new_delivered_transactions = {}
+            self.update_delivered_transactions = {}
+
+            full_rows = [
+                row._asdict()
+                for txn_map in itertools.chain(inflight.values(), new_delivered.values())
+                for row in txn_map.values()
+            ]
+
+            def f(txn):
+                if full_rows:
+                    self._simple_insert_many_txn(
+                        txn=txn,
+                        table="sent_transactions",
+                        values=full_rows
+                    )
+
+                for dest, txn_map in update_delivered.items():
+                    for txn_id, update_row in txn_map.items():
+                        self._simple_update_one_txn(
+                            txn,
+                            table="sent_transactions",
+                            keyvalues={
+                                "transaction_id": txn_id,
+                                "destination": dest,
+                            },
+                            updatevalues={
+                                "response_code": update_row.response_code,
+                                "response_json": None,  # For now, don't persist response
+                            }
+                        )
+
+            if full_rows or update_delivered:
+                yield self.runInteraction("_persist_in_mem_txns", f)
+        except:
+            logger.exception("Failed to persist transactions!")
diff --git a/synapse/types.py b/synapse/types.py
index 5b166835bd..42fd9c7204 100644
--- a/synapse/types.py
+++ b/synapse/types.py
@@ -21,6 +21,10 @@ from collections import namedtuple
 Requester = namedtuple("Requester", ["user", "access_token_id", "is_guest"])
 
 
+def get_domian_from_id(string):
+    return string.split(":", 1)[1]
+
+
 class DomainSpecificString(
         namedtuple("DomainSpecificString", ("localpart", "domain"))
 ):
diff --git a/synapse/util/presentable_names.py b/synapse/util/presentable_names.py
new file mode 100644
index 0000000000..3efa8a8206
--- /dev/null
+++ b/synapse/util/presentable_names.py
@@ -0,0 +1,159 @@
+# -*- coding: utf-8 -*-
+# Copyright 2016 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.
+
+import re
+
+# intentionally looser than what aliases we allow to be registered since
+# other HSes may allow aliases that we would not
+ALIAS_RE = re.compile(r"^#.*:.+$")
+
+ALL_ALONE = "Empty Room"
+
+
+def calculate_room_name(room_state, user_id, fallback_to_members=True):
+    """
+    Works out a user-facing name for the given room as per Matrix
+    spec recommendations.
+    Does not yet support internationalisation.
+    Args:
+        room_state: Dictionary of the room's state
+        user_id: The ID of the user to whom the room name is being presented
+        fallback_to_members: If False, return None instead of generating a name
+                             based on the room's members if the room has no
+                             title or aliases.
+
+    Returns:
+        (string or None) A human readable name for the room.
+    """
+    # does it have a name?
+    if ("m.room.name", "") in room_state:
+        m_room_name = room_state[("m.room.name", "")]
+        if m_room_name.content and m_room_name.content["name"]:
+            return m_room_name.content["name"]
+
+    # does it have a canonical alias?
+    if ("m.room.canonical_alias", "") in room_state:
+        canon_alias = room_state[("m.room.canonical_alias", "")]
+        if (
+            canon_alias.content and canon_alias.content["alias"] and
+            _looks_like_an_alias(canon_alias.content["alias"])
+        ):
+            return canon_alias.content["alias"]
+
+    # at this point we're going to need to search the state by all state keys
+    # for an event type, so rearrange the data structure
+    room_state_bytype = _state_as_two_level_dict(room_state)
+
+    # right then, any aliases at all?
+    if "m.room.aliases" in room_state_bytype:
+        m_room_aliases = room_state_bytype["m.room.aliases"]
+        if len(m_room_aliases.values()) > 0:
+            first_alias_event = m_room_aliases.values()[0]
+            if first_alias_event.content and first_alias_event.content["aliases"]:
+                the_aliases = first_alias_event.content["aliases"]
+                if len(the_aliases) > 0 and _looks_like_an_alias(the_aliases[0]):
+                    return the_aliases[0]
+
+    if not fallback_to_members:
+        return None
+
+    my_member_event = None
+    if ("m.room.member", user_id) in room_state:
+        my_member_event = room_state[("m.room.member", user_id)]
+
+    if (
+        my_member_event is not None and
+        my_member_event.content['membership'] == "invite"
+    ):
+        if ("m.room.member", my_member_event.sender) in room_state:
+            inviter_member_event = room_state[("m.room.member", my_member_event.sender)]
+            return "Invite from %s" % (name_from_member_event(inviter_member_event),)
+        else:
+            return "Room Invite"
+
+    # we're going to have to generate a name based on who's in the room,
+    # so find out who is in the room that isn't the user.
+    if "m.room.member" in room_state_bytype:
+        all_members = [
+            ev for ev in room_state_bytype["m.room.member"].values()
+            if ev.content['membership'] == "join" or ev.content['membership'] == "invite"
+        ]
+        # Sort the member events oldest-first so the we name people in the
+        # order the joined (it should at least be deterministic rather than
+        # dictionary iteration order)
+        all_members.sort(key=lambda e: e.origin_server_ts)
+        other_members = [m for m in all_members if m.state_key != user_id]
+    else:
+        other_members = []
+        all_members = []
+
+    if len(other_members) == 0:
+        if len(all_members) == 1:
+            # self-chat, peeked room with 1 participant,
+            # or inbound invite, or outbound 3PID invite.
+            if all_members[0].sender == user_id:
+                if "m.room.third_party_invite" in room_state_bytype:
+                    third_party_invites = room_state_bytype["m.room.third_party_invite"]
+                    if len(third_party_invites) > 0:
+                        # technically third party invite events are not member
+                        # events, but they are close enough
+                        return "Inviting %s" (
+                            descriptor_from_member_events(third_party_invites)
+                        )
+                    else:
+                        return ALL_ALONE
+            else:
+                return name_from_member_event(all_members[0])
+        else:
+            return ALL_ALONE
+    else:
+        return descriptor_from_member_events(other_members)
+
+
+def descriptor_from_member_events(member_events):
+    if len(member_events) == 0:
+        return "nobody"
+    elif len(member_events) == 1:
+        return name_from_member_event(member_events[0])
+    elif len(member_events) == 2:
+        return "%s and %s" % (
+            name_from_member_event(member_events[0]),
+            name_from_member_event(member_events[1]),
+        )
+    else:
+        return "%s and %d others" % (
+            name_from_member_event(member_events[0]),
+            len(member_events) - 1,
+        )
+
+
+def name_from_member_event(member_event):
+    if (
+        member_event.content and "displayname" in member_event.content and
+        member_event.content["displayname"]
+    ):
+        return member_event.content["displayname"]
+    return member_event.state_key
+
+
+def _state_as_two_level_dict(state):
+    ret = {}
+    for k, v in state.items():
+        ret.setdefault(k[0], {})[k[1]] = v
+    return ret
+
+
+def _looks_like_an_alias(string):
+    return ALIAS_RE.match(string) is not None
diff --git a/synapse/visibility.py b/synapse/visibility.py
new file mode 100644
index 0000000000..948ad51772
--- /dev/null
+++ b/synapse/visibility.py
@@ -0,0 +1,210 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014 - 2016 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.
+
+from twisted.internet import defer
+
+from synapse.api.constants import Membership, EventTypes
+
+from synapse.util.logcontext import preserve_fn
+
+import logging
+
+
+logger = logging.getLogger(__name__)
+
+
+VISIBILITY_PRIORITY = (
+    "world_readable",
+    "shared",
+    "invited",
+    "joined",
+)
+
+
+MEMBERSHIP_PRIORITY = (
+    Membership.JOIN,
+    Membership.INVITE,
+    Membership.KNOCK,
+    Membership.LEAVE,
+    Membership.BAN,
+)
+
+
+@defer.inlineCallbacks
+def filter_events_for_clients(store, user_tuples, events, event_id_to_state):
+    """ Returns dict of user_id -> list of events that user is allowed to
+    see.
+
+    Args:
+        user_tuples (str, bool): (user id, is_peeking) for each user to be
+            checked. is_peeking should be true if:
+            * the user is not currently a member of the room, and:
+            * the user has not been a member of the room since the
+            given events
+        events ([synapse.events.EventBase]): list of events to filter
+    """
+    forgotten = yield defer.gatherResults([
+        preserve_fn(store.who_forgot_in_room)(
+            room_id,
+        )
+        for room_id in frozenset(e.room_id for e in events)
+    ], consumeErrors=True)
+
+    # Set of membership event_ids that have been forgotten
+    event_id_forgotten = frozenset(
+        row["event_id"] for rows in forgotten for row in rows
+    )
+
+    ignore_dict_content = yield store.get_global_account_data_by_type_for_users(
+        "m.ignored_user_list", user_ids=[user_id for user_id, _ in user_tuples]
+    )
+
+    # FIXME: This will explode if people upload something incorrect.
+    ignore_dict = {
+        user_id: frozenset(
+            content.get("ignored_users", {}).keys() if content else []
+        )
+        for user_id, content in ignore_dict_content.items()
+    }
+
+    def allowed(event, user_id, is_peeking, ignore_list):
+        """
+        Args:
+            event (synapse.events.EventBase): event to check
+            user_id (str)
+            is_peeking (bool)
+            ignore_list (list): list of users to ignore
+        """
+        if not event.is_state() and event.sender in ignore_list:
+            return False
+
+        state = event_id_to_state[event.event_id]
+
+        # get the room_visibility at the time of the event.
+        visibility_event = state.get((EventTypes.RoomHistoryVisibility, ""), None)
+        if visibility_event:
+            visibility = visibility_event.content.get("history_visibility", "shared")
+        else:
+            visibility = "shared"
+
+        if visibility not in VISIBILITY_PRIORITY:
+            visibility = "shared"
+
+        # if it was world_readable, it's easy: everyone can read it
+        if visibility == "world_readable":
+            return True
+
+        # Always allow history visibility events on boundaries. This is done
+        # by setting the effective visibility to the least restrictive
+        # of the old vs new.
+        if event.type == EventTypes.RoomHistoryVisibility:
+            prev_content = event.unsigned.get("prev_content", {})
+            prev_visibility = prev_content.get("history_visibility", None)
+
+            if prev_visibility not in VISIBILITY_PRIORITY:
+                prev_visibility = "shared"
+
+            new_priority = VISIBILITY_PRIORITY.index(visibility)
+            old_priority = VISIBILITY_PRIORITY.index(prev_visibility)
+            if old_priority < new_priority:
+                visibility = prev_visibility
+
+        # likewise, if the event is the user's own membership event, use
+        # the 'most joined' membership
+        membership = None
+        if event.type == EventTypes.Member and event.state_key == user_id:
+            membership = event.content.get("membership", None)
+            if membership not in MEMBERSHIP_PRIORITY:
+                membership = "leave"
+
+            prev_content = event.unsigned.get("prev_content", {})
+            prev_membership = prev_content.get("membership", None)
+            if prev_membership not in MEMBERSHIP_PRIORITY:
+                prev_membership = "leave"
+
+            new_priority = MEMBERSHIP_PRIORITY.index(membership)
+            old_priority = MEMBERSHIP_PRIORITY.index(prev_membership)
+            if old_priority < new_priority:
+                membership = prev_membership
+
+        # otherwise, get the user's membership at the time of the event.
+        if membership is None:
+            membership_event = state.get((EventTypes.Member, user_id), None)
+            if membership_event:
+                if membership_event.event_id not in event_id_forgotten:
+                    membership = membership_event.membership
+
+        # if the user was a member of the room at the time of the event,
+        # they can see it.
+        if membership == Membership.JOIN:
+            return True
+
+        if visibility == "joined":
+            # we weren't a member at the time of the event, so we can't
+            # see this event.
+            return False
+
+        elif visibility == "invited":
+            # user can also see the event if they were *invited* at the time
+            # of the event.
+            return membership == Membership.INVITE
+
+        else:
+            # visibility is shared: user can also see the event if they have
+            # become a member since the event
+            #
+            # XXX: if the user has subsequently joined and then left again,
+            # ideally we would share history up to the point they left. But
+            # we don't know when they left.
+            return not is_peeking
+
+    defer.returnValue({
+        user_id: [
+            event
+            for event in events
+            if allowed(event, user_id, is_peeking, ignore_dict.get(user_id, []))
+        ]
+        for user_id, is_peeking in user_tuples
+    })
+
+
+@defer.inlineCallbacks
+def filter_events_for_client(store, user_id, events, is_peeking=False):
+    """
+    Check which events a user is allowed to see
+
+    Args:
+        user_id(str): user id to be checked
+        events([synapse.events.EventBase]): list of events to be checked
+        is_peeking(bool): should be True if:
+          * the user is not currently a member of the room, and:
+          * the user has not been a member of the room since the given
+            events
+
+    Returns:
+        [synapse.events.EventBase]
+    """
+    types = (
+        (EventTypes.RoomHistoryVisibility, ""),
+        (EventTypes.Member, user_id),
+    )
+    event_id_to_state = yield store.get_state_for_events(
+        frozenset(e.event_id for e in events),
+        types=types
+    )
+    res = yield filter_events_for_clients(
+        store, [(user_id, is_peeking)], events, event_id_to_state
+    )
+    defer.returnValue(res.get(user_id, []))