Merge branch 'release-v0.16.0' of github.com:matrix-org/synapse v0.16.0

12 files changed, 1355 insertions, 662 deletions

diff --git a/synapse/push/__init__.py b/synapse/push/__init__.py
index 296c4447ec..edf45dc599 100644
--- a/synapse/push/__init__.py
+++ b/synapse/push/__init__.py
@@ -13,333 +13,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from twisted.internet import defer
-
-from synapse.streams.config import PaginationConfig
-from synapse.types import StreamToken
-from synapse.util.logcontext import LoggingContext
-from synapse.util.metrics import Measure
-
-import synapse.util.async
-from .push_rule_evaluator import evaluator_for_user_id
-
-import logging
-import random
-
-logger = logging.getLogger(__name__)
-
-
-_NEXT_ID = 1
-
-
-def _get_next_id():
-    global _NEXT_ID
-    _id = _NEXT_ID
-    _NEXT_ID += 1
-    return _id
-
-
-# Pushers could now be moved to pull out of the event_push_actions table instead
-# of listening on the event stream: this would avoid them having to run the
-# rules again.
-class Pusher(object):
-    INITIAL_BACKOFF = 1000
-    MAX_BACKOFF = 60 * 60 * 1000
-    GIVE_UP_AFTER = 24 * 60 * 60 * 1000
-
-    def __init__(self, _hs, user_id, app_id,
-                 app_display_name, device_display_name, pushkey, pushkey_ts,
-                 data, last_token, last_success, failing_since):
-        self.hs = _hs
-        self.evStreamHandler = self.hs.get_handlers().event_stream_handler
-        self.store = self.hs.get_datastore()
-        self.clock = self.hs.get_clock()
-        self.user_id = user_id
-        self.app_id = app_id
-        self.app_display_name = app_display_name
-        self.device_display_name = device_display_name
-        self.pushkey = pushkey
-        self.pushkey_ts = pushkey_ts
-        self.data = data
-        self.last_token = last_token
-        self.last_success = last_success  # not actually used
-        self.backoff_delay = Pusher.INITIAL_BACKOFF
-        self.failing_since = failing_since
-        self.alive = True
-        self.badge = None
-
-        self.name = "Pusher-%d" % (_get_next_id(),)
-
-        # The last value of last_active_time that we saw
-        self.last_last_active_time = 0
-        self.has_unread = True
-
-    @defer.inlineCallbacks
-    def get_context_for_event(self, ev):
-        name_aliases = yield self.store.get_room_name_and_aliases(
-            ev['room_id']
-        )
-
-        ctx = {'aliases': name_aliases[1]}
-        if name_aliases[0] is not None:
-            ctx['name'] = name_aliases[0]
-
-        their_member_events_for_room = yield self.store.get_current_state(
-            room_id=ev['room_id'],
-            event_type='m.room.member',
-            state_key=ev['user_id']
-        )
-        for mev in their_member_events_for_room:
-            if mev.content['membership'] == 'join' and 'displayname' in mev.content:
-                dn = mev.content['displayname']
-                if dn is not None:
-                    ctx['sender_display_name'] = dn
-
-        defer.returnValue(ctx)
-
-    @defer.inlineCallbacks
-    def start(self):
-        with LoggingContext(self.name):
-            if not self.last_token:
-                # First-time setup: get a token to start from (we can't
-                # just start from no token, ie. 'now'
-                # because we need the result to be reproduceable in case
-                # we fail to dispatch the push)
-                config = PaginationConfig(from_token=None, limit='1')
-                chunk = yield self.evStreamHandler.get_stream(
-                    self.user_id, config, timeout=0, affect_presence=False
-                )
-                self.last_token = chunk['end']
-                yield self.store.update_pusher_last_token(
-                    self.app_id, self.pushkey, self.user_id, self.last_token
-                )
-                logger.info("New pusher %s for user %s starting from token %s",
-                            self.pushkey, self.user_id, self.last_token)
-
-            else:
-                logger.info(
-                    "Old pusher %s for user %s starting",
-                    self.pushkey, self.user_id,
-                )
-
-            wait = 0
-            while self.alive:
-                try:
-                    if wait > 0:
-                        yield synapse.util.async.sleep(wait)
-                    with Measure(self.clock, "push"):
-                        yield self.get_and_dispatch()
-                    wait = 0
-                except:
-                    if wait == 0:
-                        wait = 1
-                    else:
-                        wait = min(wait * 2, 1800)
-                    logger.exception(
-                        "Exception in pusher loop for pushkey %s. Pausing for %ds",
-                        self.pushkey, wait
-                    )
-
-    @defer.inlineCallbacks
-    def get_and_dispatch(self):
-        from_tok = StreamToken.from_string(self.last_token)
-        config = PaginationConfig(from_token=from_tok, limit='1')
-        timeout = (300 + random.randint(-60, 60)) * 1000
-        chunk = yield self.evStreamHandler.get_stream(
-            self.user_id, config, timeout=timeout, affect_presence=False,
-            only_keys=("room", "receipt",),
-        )
-
-        # limiting to 1 may get 1 event plus 1 presence event, so
-        # pick out the actual event
-        single_event = None
-        read_receipt = None
-        for c in chunk['chunk']:
-            if 'event_id' in c:  # Hmmm...
-                single_event = c
-            elif c['type'] == 'm.receipt':
-                read_receipt = c
-
-        have_updated_badge = False
-        if read_receipt:
-            for receipt_part in read_receipt['content'].values():
-                if 'm.read' in receipt_part:
-                    if self.user_id in receipt_part['m.read'].keys():
-                        have_updated_badge = True
-
-        if not single_event:
-            if have_updated_badge:
-                yield self.update_badge()
-            self.last_token = chunk['end']
-            yield self.store.update_pusher_last_token(
-                self.app_id,
-                self.pushkey,
-                self.user_id,
-                self.last_token
-            )
-            return
-
-        if not self.alive:
-            return
-
-        processed = False
-
-        rule_evaluator = yield \
-            evaluator_for_user_id(
-                self.user_id, single_event['room_id'], self.store
-            )
-
-        actions = yield rule_evaluator.actions_for_event(single_event)
-        tweaks = rule_evaluator.tweaks_for_actions(actions)
-
-        if 'notify' in actions:
-            self.badge = yield self._get_badge_count()
-            rejected = yield self.dispatch_push(single_event, tweaks, self.badge)
-            self.has_unread = True
-            if isinstance(rejected, list) or isinstance(rejected, tuple):
-                processed = True
-                for pk in rejected:
-                    if pk != self.pushkey:
-                        # for sanity, we only remove the pushkey if it
-                        # was the one we actually sent...
-                        logger.warn(
-                            ("Ignoring rejected pushkey %s because we"
-                             " didn't send it"), pk
-                        )
-                    else:
-                        logger.info(
-                            "Pushkey %s was rejected: removing",
-                            pk
-                        )
-                        yield self.hs.get_pusherpool().remove_pusher(
-                            self.app_id, pk, self.user_id
-                        )
-        else:
-            if have_updated_badge:
-                yield self.update_badge()
-            processed = True
-
-        if not self.alive:
-            return
-
-        if processed:
-            self.backoff_delay = Pusher.INITIAL_BACKOFF
-            self.last_token = chunk['end']
-            yield self.store.update_pusher_last_token_and_success(
-                self.app_id,
-                self.pushkey,
-                self.user_id,
-                self.last_token,
-                self.clock.time_msec()
-            )
-            if self.failing_since:
-                self.failing_since = None
-                yield self.store.update_pusher_failing_since(
-                    self.app_id,
-                    self.pushkey,
-                    self.user_id,
-                    self.failing_since)
-        else:
-            if not self.failing_since:
-                self.failing_since = self.clock.time_msec()
-                yield self.store.update_pusher_failing_since(
-                    self.app_id,
-                    self.pushkey,
-                    self.user_id,
-                    self.failing_since
-                )
-
-            if (self.failing_since and
-               self.failing_since <
-               self.clock.time_msec() - Pusher.GIVE_UP_AFTER):
-                # we really only give up so that if the URL gets
-                # fixed, we don't suddenly deliver a load
-                # of old notifications.
-                logger.warn("Giving up on a notification to user %s, "
-                            "pushkey %s",
-                            self.user_id, self.pushkey)
-                self.backoff_delay = Pusher.INITIAL_BACKOFF
-                self.last_token = chunk['end']
-                yield self.store.update_pusher_last_token(
-                    self.app_id,
-                    self.pushkey,
-                    self.user_id,
-                    self.last_token
-                )
-
-                self.failing_since = None
-                yield self.store.update_pusher_failing_since(
-                    self.app_id,
-                    self.pushkey,
-                    self.user_id,
-                    self.failing_since
-                )
-            else:
-                logger.warn("Failed to dispatch push for user %s "
-                            "(failing for %dms)."
-                            "Trying again in %dms",
-                            self.user_id,
-                            self.clock.time_msec() - self.failing_since,
-                            self.backoff_delay)
-                yield synapse.util.async.sleep(self.backoff_delay / 1000.0)
-                self.backoff_delay *= 2
-                if self.backoff_delay > Pusher.MAX_BACKOFF:
-                    self.backoff_delay = Pusher.MAX_BACKOFF
-
-    def stop(self):
-        self.alive = False
-
-    def dispatch_push(self, p, tweaks, badge):
-        """
-        Overridden by implementing classes to actually deliver the notification
-        Args:
-            p: The event to notify for as a single event from the event stream
-        Returns: If the notification was delivered, an array containing any
-                 pushkeys that were rejected by the push gateway.
-                 False if the notification could not be delivered (ie.
-                 should be retried).
-        """
-        pass
-
-    @defer.inlineCallbacks
-    def update_badge(self):
-        new_badge = yield self._get_badge_count()
-        if self.badge != new_badge:
-            self.badge = new_badge
-            yield self.send_badge(self.badge)
-
-    def send_badge(self, badge):
-        """
-        Overridden by implementing classes to send an updated badge count
-        """
-        pass
-
-    @defer.inlineCallbacks
-    def _get_badge_count(self):
-        invites, joins = yield defer.gatherResults([
-            self.store.get_invited_rooms_for_user(self.user_id),
-            self.store.get_rooms_for_user(self.user_id),
-        ], consumeErrors=True)
-
-        my_receipts_by_room = yield self.store.get_receipts_for_user(
-            self.user_id,
-            "m.read",
-        )
-
-        badge = len(invites)
-
-        for r in joins:
-            if r.room_id in my_receipts_by_room:
-                last_unread_event_id = my_receipts_by_room[r.room_id]
-
-                notifs = yield (
-                    self.store.get_unread_event_push_actions_by_room_for_user(
-                        r.room_id, self.user_id, last_unread_event_id
-                    )
-                )
-                badge += notifs["notify_count"]
-        defer.returnValue(badge)
-
 
 class PusherConfigException(Exception):
     def __init__(self, msg):
diff --git a/synapse/push/action_generator.py b/synapse/push/action_generator.py
index 84efcdd184..46e768e35c 100644
--- a/synapse/push/action_generator.py
+++ b/synapse/push/action_generator.py
@@ -15,7 +15,9 @@
 
 from twisted.internet import defer
 
-from .bulk_push_rule_evaluator import evaluator_for_room_id
+from .bulk_push_rule_evaluator import evaluator_for_event
+
+from synapse.util.metrics import Measure
 
 import logging
 
@@ -25,6 +27,7 @@ logger = logging.getLogger(__name__)
 class ActionGenerator:
     def __init__(self, hs):
         self.hs = hs
+        self.clock = hs.get_clock()
         self.store = hs.get_datastore()
         # really we want to get all user ids and all profile tags too,
         # since we want the actions for each profile tag for every user and
@@ -34,15 +37,16 @@ class ActionGenerator:
         # tag (ie. we just need all the users).
 
     @defer.inlineCallbacks
-    def handle_push_actions_for_event(self, event, context, handler):
-        bulk_evaluator = yield evaluator_for_room_id(
-            event.room_id, self.hs, self.store
-        )
-
-        actions_by_user = yield bulk_evaluator.action_for_event_by_user(
-            event, handler, context.current_state
-        )
-
-        context.push_actions = [
-            (uid, actions) for uid, actions in actions_by_user.items()
-        ]
+    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, context.current_state
+            )
+
+            actions_by_user = yield bulk_evaluator.action_for_event_by_user(
+                event, context.current_state
+            )
+
+            context.push_actions = [
+                (uid, actions) for uid, actions in actions_by_user.items()
+            ]
diff --git a/synapse/push/baserules.py b/synapse/push/baserules.py
index 86a2998bcc..024c14904f 100644
--- a/synapse/push/baserules.py
+++ b/synapse/push/baserules.py
@@ -19,9 +19,11 @@ import copy
 def list_with_base_rules(rawrules):
     """Combine the list of rules set by the user with the default push rules
 
-    :param list rawrules: The rules the user has modified or set.
-    :returns: A new list with the rules set by the user combined with the
-        defaults.
+    Args:
+        rawrules(list): The rules the user has modified or set.
+
+    Returns:
+        A new list with the rules set by the user combined with the defaults.
     """
     ruleslist = []
 
@@ -77,7 +79,7 @@ def make_base_append_rules(kind, modified_base_rules):
     rules = []
 
     if kind == 'override':
-        rules = BASE_APPEND_OVRRIDE_RULES
+        rules = BASE_APPEND_OVERRIDE_RULES
     elif kind == 'underride':
         rules = BASE_APPEND_UNDERRIDE_RULES
     elif kind == 'content':
@@ -146,7 +148,7 @@ BASE_PREPEND_OVERRIDE_RULES = [
 ]
 
 
-BASE_APPEND_OVRRIDE_RULES = [
+BASE_APPEND_OVERRIDE_RULES = [
     {
         'rule_id': 'global/override/.m.rule.suppress_notices',
         'conditions': [
@@ -160,7 +162,61 @@ BASE_APPEND_OVRRIDE_RULES = [
         'actions': [
             'dont_notify',
         ]
-    }
+    },
+    # NB. .m.rule.invite_for_me must be higher prio than .m.rule.member_event
+    # otherwise invites will be matched by .m.rule.member_event
+    {
+        'rule_id': 'global/override/.m.rule.invite_for_me',
+        'conditions': [
+            {
+                'kind': 'event_match',
+                'key': 'type',
+                'pattern': 'm.room.member',
+                '_id': '_member',
+            },
+            {
+                'kind': 'event_match',
+                'key': 'content.membership',
+                'pattern': 'invite',
+                '_id': '_invite_member',
+            },
+            {
+                'kind': 'event_match',
+                'key': 'state_key',
+                'pattern_type': 'user_id'
+            },
+        ],
+        'actions': [
+            'notify',
+            {
+                'set_tweak': 'sound',
+                'value': 'default'
+            }, {
+                'set_tweak': 'highlight',
+                'value': False
+            }
+        ]
+    },
+    # Will we sometimes want to know about people joining and leaving?
+    # Perhaps: if so, this could be expanded upon. Seems the most usual case
+    # is that we don't though. We add this override rule so that even if
+    # the room rule is set to notify, we don't get notifications about
+    # join/leave/avatar/displayname events.
+    # See also: https://matrix.org/jira/browse/SYN-607
+    {
+        'rule_id': 'global/override/.m.rule.member_event',
+        'conditions': [
+            {
+                'kind': 'event_match',
+                'key': 'type',
+                'pattern': 'm.room.member',
+                '_id': '_member',
+            }
+        ],
+        'actions': [
+            'dont_notify'
+        ]
+    },
 ]
 
 
@@ -230,57 +286,6 @@ BASE_APPEND_UNDERRIDE_RULES = [
         ]
     },
     {
-        'rule_id': 'global/underride/.m.rule.invite_for_me',
-        'conditions': [
-            {
-                'kind': 'event_match',
-                'key': 'type',
-                'pattern': 'm.room.member',
-                '_id': '_member',
-            },
-            {
-                'kind': 'event_match',
-                'key': 'content.membership',
-                'pattern': 'invite',
-                '_id': '_invite_member',
-            },
-            {
-                'kind': 'event_match',
-                'key': 'state_key',
-                'pattern_type': 'user_id'
-            },
-        ],
-        'actions': [
-            'notify',
-            {
-                'set_tweak': 'sound',
-                'value': 'default'
-            }, {
-                'set_tweak': 'highlight',
-                'value': False
-            }
-        ]
-    },
-    # This is too simple: https://matrix.org/jira/browse/SYN-607
-    # Removing for now
-    # {
-    #     'rule_id': 'global/underride/.m.rule.member_event',
-    #     'conditions': [
-    #         {
-    #             'kind': 'event_match',
-    #             'key': 'type',
-    #             'pattern': 'm.room.member',
-    #             '_id': '_member',
-    #         }
-    #     ],
-    #     'actions': [
-    #         'notify', {
-    #             'set_tweak': 'highlight',
-    #             'value': False
-    #         }
-    #     ]
-    # },
-    {
         'rule_id': 'global/underride/.m.rule.message',
         'conditions': [
             {
@@ -312,7 +317,7 @@ for r in BASE_PREPEND_OVERRIDE_RULES:
     r['default'] = True
     BASE_RULE_IDS.add(r['rule_id'])
 
-for r in BASE_APPEND_OVRRIDE_RULES:
+for r in BASE_APPEND_OVERRIDE_RULES:
     r['priority_class'] = PRIORITY_CLASS_MAP['override']
     r['default'] = True
     BASE_RULE_IDS.add(r['rule_id'])
diff --git a/synapse/push/bulk_push_rule_evaluator.py b/synapse/push/bulk_push_rule_evaluator.py
index 76d7eb7ce0..756e5da513 100644
--- a/synapse/push/bulk_push_rule_evaluator.py
+++ b/synapse/push/bulk_push_rule_evaluator.py
@@ -14,67 +14,66 @@
 # limitations under the License.
 
 import logging
-import ujson as json
 
 from twisted.internet import defer
 
-from .baserules import list_with_base_rules
 from .push_rule_evaluator import PushRuleEvaluatorForEvent
 
-from synapse.api.constants import EventTypes
+from synapse.api.constants import EventTypes, Membership
+from synapse.visibility import filter_events_for_clients
 
 
 logger = logging.getLogger(__name__)
 
 
-def decode_rule_json(rule):
-    rule['conditions'] = json.loads(rule['conditions'])
-    rule['actions'] = json.loads(rule['actions'])
-    return rule
-
-
 @defer.inlineCallbacks
 def _get_rules(room_id, user_ids, store):
     rules_by_user = yield store.bulk_get_push_rules(user_ids)
-    rules_enabled_by_user = yield store.bulk_get_push_rules_enabled(user_ids)
-
-    rules_by_user = {
-        uid: list_with_base_rules([
-            decode_rule_json(rule_list)
-            for rule_list in rules_by_user.get(uid, [])
-        ])
-        for uid in user_ids
-    }
-
-    # We apply the rules-enabled map here: bulk_get_push_rules doesn't
-    # fetch disabled rules, but this won't account for any server default
-    # rules the user has disabled, so we need to do this too.
-    for uid in user_ids:
-        if uid not in rules_enabled_by_user:
-            continue
-
-        user_enabled_map = rules_enabled_by_user[uid]
-
-        for i, rule in enumerate(rules_by_user[uid]):
-            rule_id = rule['rule_id']
-
-            if rule_id in user_enabled_map:
-                if rule.get('enabled', True) != bool(user_enabled_map[rule_id]):
-                    # Rules are cached across users.
-                    rule = dict(rule)
-                    rule['enabled'] = bool(user_enabled_map[rule_id])
-                    rules_by_user[uid][i] = rule
+
+    rules_by_user = {k: v for k, v in rules_by_user.items() if v is not None}
 
     defer.returnValue(rules_by_user)
 
 
 @defer.inlineCallbacks
-def evaluator_for_room_id(room_id, hs, store):
-    results = yield store.get_receipts_for_room(room_id, "m.read")
-    user_ids = [
-        row["user_id"] for row in results
-        if hs.is_mine_id(row["user_id"])
-    ]
+def evaluator_for_event(event, hs, store, current_state):
+    room_id = event.room_id
+    # We also will want to generate notifs for other people in the room so
+    # their unread countss are correct in the event stream, but to avoid
+    # generating them for bot / AS users etc, we only do so for people who've
+    # sent a read receipt into the room.
+
+    local_users_in_room = set(
+        e.state_key for e in current_state.values()
+        if e.type == EventTypes.Member and e.membership == Membership.JOIN
+        and hs.is_mine_id(e.state_key)
+    )
+
+    # users in the room who have pushers need to get push rules run because
+    # that's how their pushers work
+    if_users_with_pushers = yield store.get_if_users_have_pushers(
+        local_users_in_room
+    )
+    user_ids = set(
+        uid for uid, have_pusher in if_users_with_pushers.items() if have_pusher
+    )
+
+    users_with_receipts = yield store.get_users_with_read_receipts_in_room(room_id)
+
+    # any users with pushers must be ours: they have pushers
+    for uid in users_with_receipts:
+        if uid in local_users_in_room:
+            user_ids.add(uid)
+
+    # if this event is an invite event, we may need to run rules for the user
+    # who's been invited, otherwise they won't get told they've been invited
+    if event.type == 'm.room.member' and event.content['membership'] == 'invite':
+        invited_user = event.state_key
+        if invited_user and hs.is_mine_id(invited_user):
+            has_pusher = yield store.user_has_pusher(invited_user)
+            if has_pusher:
+                user_ids.add(invited_user)
+
     rules_by_user = yield _get_rules(room_id, user_ids, store)
 
     defer.returnValue(BulkPushRuleEvaluator(
@@ -98,16 +97,24 @@ 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 = {}
 
-        users_dict = yield self.store.are_guests(self.rules_by_user.keys())
+        # None of these users can be peeking since this list of users comes
+        # from the set of users in the room, so we know for sure they're all
+        # actually in the room.
+        user_tuples = [
+            (u, False) for u in self.rules_by_user.keys()
+        ]
 
-        filtered_by_user = yield handler.filter_events_for_clients(
-            users_dict.items(), [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)
+        room_members = set(
+            e.state_key for e in current_state.values()
+            if e.type == EventTypes.Member and e.membership == Membership.JOIN
+        )
 
         evaluator = PushRuleEvaluatorForEvent(event, len(room_members))
 
diff --git a/synapse/push/clientformat.py b/synapse/push/clientformat.py
index ae9db9ec2f..e0331b2d2d 100644
--- a/synapse/push/clientformat.py
+++ b/synapse/push/clientformat.py
@@ -13,29 +13,19 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from synapse.push.baserules import list_with_base_rules
-
 from synapse.push.rulekinds import (
     PRIORITY_CLASS_MAP, PRIORITY_CLASS_INVERSE_MAP
 )
 
 import copy
-import simplejson as json
 
 
-def format_push_rules_for_user(user, rawrules, enabled_map):
+def format_push_rules_for_user(user, ruleslist):
     """Converts a list of rawrules and a enabled map into nested dictionaries
     to match the Matrix client-server format for push rules"""
 
-    ruleslist = []
-    for rawrule in rawrules:
-        rule = dict(rawrule)
-        rule["conditions"] = json.loads(rawrule["conditions"])
-        rule["actions"] = json.loads(rawrule["actions"])
-        ruleslist.append(rule)
-
     # We're going to be mutating this a lot, so do a deep copy
-    ruleslist = copy.deepcopy(list_with_base_rules(ruleslist))
+    ruleslist = copy.deepcopy(ruleslist)
 
     rules = {'global': {}, 'device': {}}
 
@@ -60,9 +50,7 @@ def format_push_rules_for_user(user, rawrules, enabled_map):
 
         template_rule = _rule_to_template(r)
         if template_rule:
-            if r['rule_id'] in enabled_map:
-                template_rule['enabled'] = enabled_map[r['rule_id']]
-            elif 'enabled' in r:
+            if 'enabled' in r:
                 template_rule['enabled'] = r['enabled']
             else:
                 template_rule['enabled'] = True
diff --git a/synapse/push/emailpusher.py b/synapse/push/emailpusher.py
new file mode 100644
index 0000000000..12a3ec7fd8
--- /dev/null
+++ b/synapse/push/emailpusher.py
@@ -0,0 +1,283 @@
+# -*- 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 = 10 * 60 * 1000
+
+# THROTTLE is the minimum time between mail notifications sent for a given room.
+# Each room maintains its own throttle counter, but each new mail notification
+# sends the pending notifications for all rooms.
+THROTTLE_START_MS = 10 * 60 * 1000
+THROTTLE_MAX_MS = 24 * 60 * 60 * 1000  # 24h
+# THROTTLE_MULTIPLIER = 6              # 10 mins, 1 hour, 6 hours, 24 hours
+THROTTLE_MULTIPLIER = 144              # 10 mins, 24 hours - i.e. jump straight to 1 day
+
+# If no event triggers a notification for this long after the previous,
+# the throttle is released.
+# 12 hours - a gap of 12 hours in conversation is surely enough to merit a new
+# notification when things get going again...
+THROTTLE_RESET_AFTER_MS = (12 * 60 * 60 * 1000)
+
+# does each email include all unread notifs, or just the ones which have happened
+# since the last mail?
+# XXX: this is currently broken as it includes ones from parted rooms(!)
+INCLUDE_ALL_UNREAD_NOTIFS = False
+
+
+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:
+            if 'data' in pusherdict and 'brand' in pusherdict['data']:
+                app_name = pusherdict['data']['brand']
+            else:
+                app_name = self.hs.config.email_app_name
+
+            self.mailer = Mailer(self.hs, app_name)
+        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.
+        """
+        start = 0 if INCLUDE_ALL_UNREAD_NOTIFS else self.last_stream_ordering
+        unprocessed = yield self.store.get_unread_push_actions_for_user_in_range(
+            self.user_id, start, 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.
+
+                reason = {
+                    'room_id': push_action['room_id'],
+                    'now': self.clock.time_msec(),
+                    'received_at': received_at,
+                    'delay_before_mail_ms': DELAY_BEFORE_MAIL_MS,
+                    'last_sent_ts': self.get_room_last_sent_ts(push_action['room_id']),
+                    'throttle_ms': self.get_room_throttle_ms(push_action['room_id']),
+                }
+
+                yield self.send_notification(unprocessed, reason)
+
+                yield self.save_last_stream_ordering_and_success(max([
+                    ea['stream_ordering'] for ea in unprocessed
+                ]))
+
+                # we update the throttle on all the possible unprocessed push actions
+                for ea in unprocessed:
+                    yield self.sent_notif_update_throttle(
+                        ea['room_id'], ea
+                    )
+                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: The timestamp when we are next allowed to send an email notif
+        for this room
+        """
+        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, reason):
+        logger.info("Sending notif email for user %r", self.user_id)
+
+        yield self.mailer.send_notification_mail(
+            self.app_id, self.user_id, self.email, push_actions, reason
+        )
diff --git a/synapse/push/httppusher.py b/synapse/push/httppusher.py
index 9be4869360..3992804845 100644
--- a/synapse/push/httppusher.py
+++ b/synapse/push/httppusher.py
@@ -13,60 +13,239 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from synapse.push import Pusher, PusherConfigException
+from synapse.push import PusherConfigException
 
-from twisted.internet import defer
+from twisted.internet import defer, reactor
 
 import logging
+import push_rule_evaluator
+import push_tools
+
+from synapse.util.logcontext import LoggingContext
+from synapse.util.metrics import Measure
 
 logger = logging.getLogger(__name__)
 
 
-class HttpPusher(Pusher):
-    def __init__(self, _hs, user_id, app_id,
-                 app_display_name, device_display_name, pushkey, pushkey_ts,
-                 data, last_token, last_success, failing_since):
-        super(HttpPusher, self).__init__(
-            _hs,
-            user_id,
-            app_id,
-            app_display_name,
-            device_display_name,
-            pushkey,
-            pushkey_ts,
-            data,
-            last_token,
-            last_success,
-            failing_since
+class HttpPusher(object):
+    INITIAL_BACKOFF_SEC = 1  # in seconds because that's what Twisted takes
+    MAX_BACKOFF_SEC = 60 * 60
+
+    # This one's in ms because we compare it against the clock
+    GIVE_UP_AFTER_MS = 24 * 60 * 60 * 1000
+
+    def __init__(self, hs, pusherdict):
+        self.hs = hs
+        self.store = self.hs.get_datastore()
+        self.clock = self.hs.get_clock()
+        self.user_id = pusherdict['user_name']
+        self.app_id = pusherdict['app_id']
+        self.app_display_name = pusherdict['app_display_name']
+        self.device_display_name = pusherdict['device_display_name']
+        self.pushkey = pusherdict['pushkey']
+        self.pushkey_ts = pusherdict['ts']
+        self.data = pusherdict['data']
+        self.last_stream_ordering = pusherdict['last_stream_ordering']
+        self.backoff_delay = HttpPusher.INITIAL_BACKOFF_SEC
+        self.failing_since = pusherdict['failing_since']
+        self.timed_call = None
+        self.processing = False
+
+        # This is the highest stream ordering we know it's safe to process.
+        # When new events arrive, we'll be given a window of new events: we
+        # should honour this rather than just looking for anything higher
+        # because of potential out-of-order event serialisation. This starts
+        # off as None though as we don't know any better.
+        self.max_stream_ordering = None
+
+        if 'data' not in pusherdict:
+            raise PusherConfigException(
+                "No 'data' key for HTTP pusher"
+            )
+        self.data = pusherdict['data']
+
+        self.name = "%s/%s/%s" % (
+            pusherdict['user_name'],
+            pusherdict['app_id'],
+            pusherdict['pushkey'],
         )
-        if 'url' not in data:
+
+        if 'url' not in self.data:
             raise PusherConfigException(
                 "'url' required in data for HTTP pusher"
             )
-        self.url = data['url']
-        self.http_client = _hs.get_simple_http_client()
+        self.url = self.data['url']
+        self.http_client = hs.get_simple_http_client()
         self.data_minus_url = {}
         self.data_minus_url.update(self.data)
         del self.data_minus_url['url']
 
     @defer.inlineCallbacks
-    def _build_notification_dict(self, event, tweaks, badge):
-        # we probably do not want to push for every presence update
-        # (we may want to be able to set up notifications when specific
-        # people sign in, but we'd want to only deliver the pertinent ones)
-        # Actually, presence events will not get this far now because we
-        # need to filter them out in the main Pusher code.
-        if 'event_id' not in event:
-            defer.returnValue(None)
+    def on_started(self):
+        yield self._process()
+
+    @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()
+
+    @defer.inlineCallbacks
+    def on_new_receipts(self, min_stream_id, max_stream_id):
+        # Note that the min here shouldn't be relied upon to be accurate.
+
+        # We could check the receipts are actually m.read receipts here,
+        # but currently that's the only type of receipt anyway...
+        with LoggingContext("push.on_new_receipts"):
+            with Measure(self.clock, "push.on_new_receipts"):
+                badge = yield push_tools.get_badge_count(
+                    self.hs.get_datastore(), self.user_id
+                )
+            yield self._send_badge(badge)
+
+    @defer.inlineCallbacks
+    def on_timer(self):
+        yield self._process()
+
+    def on_stop(self):
+        if self.timed_call:
+            self.timed_call.cancel()
 
-        ctx = yield self.get_context_for_event(event)
+    @defer.inlineCallbacks
+    def _process(self):
+        if self.processing:
+            return
+
+        with LoggingContext("push._process"):
+            with Measure(self.clock, "push._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):
+        """
+        Looks for unset notifications and dispatch them, in order
+        Never call this directly: use _process which will only allow this to
+        run once per pusher.
+        """
+
+        unprocessed = yield self.store.get_unread_push_actions_for_user_in_range(
+            self.user_id, self.last_stream_ordering, self.max_stream_ordering
+        )
+
+        for push_action in unprocessed:
+            processed = yield self._process_one(push_action)
+            if processed:
+                self.backoff_delay = HttpPusher.INITIAL_BACKOFF_SEC
+                self.last_stream_ordering = push_action['stream_ordering']
+                yield self.store.update_pusher_last_stream_ordering_and_success(
+                    self.app_id, self.pushkey, self.user_id,
+                    self.last_stream_ordering,
+                    self.clock.time_msec()
+                )
+                if self.failing_since:
+                    self.failing_since = None
+                    yield self.store.update_pusher_failing_since(
+                        self.app_id, self.pushkey, self.user_id,
+                        self.failing_since
+                    )
+            else:
+                if not self.failing_since:
+                    self.failing_since = self.clock.time_msec()
+                    yield self.store.update_pusher_failing_since(
+                        self.app_id, self.pushkey, self.user_id,
+                        self.failing_since
+                    )
+
+                if (
+                    self.failing_since and
+                    self.failing_since <
+                    self.clock.time_msec() - HttpPusher.GIVE_UP_AFTER_MS
+                ):
+                    # we really only give up so that if the URL gets
+                    # fixed, we don't suddenly deliver a load
+                    # of old notifications.
+                    logger.warn("Giving up on a notification to user %s, "
+                                "pushkey %s",
+                                self.user_id, self.pushkey)
+                    self.backoff_delay = HttpPusher.INITIAL_BACKOFF_SEC
+                    self.last_stream_ordering = push_action['stream_ordering']
+                    yield self.store.update_pusher_last_stream_ordering(
+                        self.app_id,
+                        self.pushkey,
+                        self.user_id,
+                        self.last_stream_ordering
+                    )
+
+                    self.failing_since = None
+                    yield self.store.update_pusher_failing_since(
+                        self.app_id,
+                        self.pushkey,
+                        self.user_id,
+                        self.failing_since
+                    )
+                else:
+                    logger.info("Push failed: delaying for %ds", self.backoff_delay)
+                    self.timed_call = reactor.callLater(self.backoff_delay, self.on_timer)
+                    self.backoff_delay = min(self.backoff_delay * 2, self.MAX_BACKOFF_SEC)
+                    break
+
+    @defer.inlineCallbacks
+    def _process_one(self, push_action):
+        if 'notify' not in push_action['actions']:
+            defer.returnValue(True)
+
+        tweaks = push_rule_evaluator.tweaks_for_actions(push_action['actions'])
+        badge = yield push_tools.get_badge_count(self.hs.get_datastore(), self.user_id)
+
+        event = yield self.store.get_event(push_action['event_id'], allow_none=True)
+        if event is None:
+            defer.returnValue(True)  # It's been redacted
+        rejected = yield self.dispatch_push(event, tweaks, badge)
+        if rejected is False:
+            defer.returnValue(False)
+
+        if isinstance(rejected, list) or isinstance(rejected, tuple):
+            for pk in rejected:
+                if pk != self.pushkey:
+                    # for sanity, we only remove the pushkey if it
+                    # was the one we actually sent...
+                    logger.warn(
+                        ("Ignoring rejected pushkey %s because we"
+                         " didn't send it"), pk
+                    )
+                else:
+                    logger.info(
+                        "Pushkey %s was rejected: removing",
+                        pk
+                    )
+                    yield self.hs.remove_pusher(
+                        self.app_id, pk, self.user_id
+                    )
+        defer.returnValue(True)
+
+    @defer.inlineCallbacks
+    def _build_notification_dict(self, event, tweaks, badge):
+        ctx = yield push_tools.get_context_for_event(self.hs.get_datastore(), event)
 
         d = {
             'notification': {
-                'id': event['event_id'],
-                'room_id': event['room_id'],
-                'type': event['type'],
-                'sender': event['user_id'],
+                'id': event.event_id,  # deprecated: remove soon
+                'event_id': event.event_id,
+                'room_id': event.room_id,
+                'type': event.type,
+                'sender': event.user_id,
                 'counts': {  # -- we don't mark messages as read yet so
                              # we have no way of knowing
                     # Just set the badge to 1 until we have read receipts
@@ -84,11 +263,11 @@ class HttpPusher(Pusher):
                 ]
             }
         }
-        if event['type'] == 'm.room.member':
-            d['notification']['membership'] = event['content']['membership']
-            d['notification']['user_is_target'] = event['state_key'] == self.user_id
+        if event.type == 'm.room.member':
+            d['notification']['membership'] = event.content['membership']
+            d['notification']['user_is_target'] = event.state_key == self.user_id
         if 'content' in event:
-            d['notification']['content'] = event['content']
+            d['notification']['content'] = event.content
 
         if len(ctx['aliases']):
             d['notification']['room_alias'] = ctx['aliases'][0]
@@ -115,7 +294,7 @@ class HttpPusher(Pusher):
         defer.returnValue(rejected)
 
     @defer.inlineCallbacks
-    def send_badge(self, badge):
+    def _send_badge(self, badge):
         logger.info("Sending updated badge count %d to %r", badge, self.user_id)
         d = {
             'notification': {
diff --git a/synapse/push/mailer.py b/synapse/push/mailer.py
new file mode 100644
index 0000000000..e5c3929cd7
--- /dev/null
+++ b/synapse/push/mailer.py
@@ -0,0 +1,514 @@
+# -*- 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 %(room)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 = "You have messages on %(app)s in the %(room)s room..."
+MESSAGES_IN_ROOM_AND_OTHERS = \
+    "You have messages on %(app)s in the %(room)s room and others..."
+MESSAGES_FROM_PERSON_AND_OTHERS = \
+    "You have messages on %(app)s from %(person)s and others..."
+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, app_name):
+        self.hs = hs
+        self.store = self.hs.get_datastore()
+        self.auth_handler = self.hs.get_auth_handler()
+        self.state_handler = self.hs.get_state_handler()
+        loader = jinja2.FileSystemLoader(self.hs.config.email_template_dir)
+        self.app_name = app_name
+        logger.info("Created Mailer for app_name %s" % 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, app_id, user_id, email_address,
+                               push_actions, reason):
+        try:
+            from_string = self.hs.config.email_notif_from % {
+                "app": self.app_name
+            }
+        except TypeError:
+            from_string = self.hs.config.email_notif_from
+
+        raw_from = email.utils.parseaddr(from_string)[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
+            )
+            if user_display_name is None:
+                user_display_name = user_id
+        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 less realtime than sync so we can afford to wait a bit.
+        yield concurrently_execute(_fetch_room_state, rooms_in_order, 3)
+
+        # actually sort our so-called rooms_in_order list, most recent room first
+        rooms_in_order.sort(
+            key=lambda r: -(notifs_by_room[r][-1]['received_ts'] or 0)
+        )
+
+        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)
+
+        reason['room_name'] = calculate_room_name(
+            state_by_room[reason['room_id']], user_id, fallback_to_members=True
+        )
+
+        summary_text = self.make_summary_text(
+            notifs_by_room, state_by_room, notif_events, user_id, reason
+        )
+
+        template_vars = {
+            "user_display_name": user_display_name,
+            "unsubscribe_link": self.make_unsubscribe_link(
+                user_id, app_id, email_address
+            ),
+            "summary_text": summary_text,
+            "app_name": self.app_name,
+            "rooms": rooms,
+            "reason": reason,
+        }
+
+        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'] = from_string
+        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 = None
+        if "avatar_url" in sender_state_event.content:
+            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, reason):
+        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
+
+            # ...but we still refer to the 'reason' room which triggered the mail
+            if reason['room_name'] is not None:
+                return MESSAGES_IN_ROOM_AND_OTHERS % {
+                    "room": reason['room_name'],
+                    "app": self.app_name,
+                }
+            else:
+                # If the reason 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[reason['room_id']]
+                ]))
+
+                return MESSAGES_FROM_PERSON_AND_OTHERS % {
+                    "person": descriptor_from_member_events([
+                        state_by_room[reason['room_id']][("m.room.member", s)]
+                        for s in sender_ids
+                    ]),
+                    "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, user_id, app_id, email_address):
+        params = {
+            "access_token": self.auth_handler.generate_delete_pusher_token(user_id),
+            "app_id": app_id,
+            "pushkey": email_address,
+        }
+
+        # XXX: make r0 once API is stable
+        return "%s_matrix/client/unstable/pushers/remove?%s" % (
+            self.hs.config.public_baseurl,
+            urllib.urlencode(params),
+        )
+
+    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/push_rule_evaluator.py b/synapse/push/push_rule_evaluator.py
index 51f73a5b78..4db76f18bd 100644
--- a/synapse/push/push_rule_evaluator.py
+++ b/synapse/push/push_rule_evaluator.py
@@ -13,12 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from twisted.internet import defer
-
-from .baserules import list_with_base_rules
-
 import logging
-import simplejson as json
 import re
 
 from synapse.types import UserID
@@ -32,22 +27,6 @@ IS_GLOB = re.compile(r'[\?\*\[\]]')
 INEQUALITY_EXPR = re.compile("^([=<>]*)([0-9]*)$")
 
 
-@defer.inlineCallbacks
-def evaluator_for_user_id(user_id, room_id, store):
-    rawrules = yield store.get_push_rules_for_user(user_id)
-    enabled_map = yield store.get_push_rules_enabled_for_user(user_id)
-    our_member_event = yield store.get_current_state(
-        room_id=room_id,
-        event_type='m.room.member',
-        state_key=user_id,
-    )
-
-    defer.returnValue(PushRuleEvaluator(
-        user_id, rawrules, enabled_map,
-        room_id, our_member_event, store
-    ))
-
-
 def _room_member_count(ev, condition, room_member_count):
     if 'is' not in condition:
         return False
@@ -74,110 +53,14 @@ def _room_member_count(ev, condition, room_member_count):
         return False
 
 
-class PushRuleEvaluator:
-    DEFAULT_ACTIONS = []
-
-    def __init__(self, user_id, raw_rules, enabled_map, room_id,
-                 our_member_event, store):
-        self.user_id = user_id
-        self.room_id = room_id
-        self.our_member_event = our_member_event
-        self.store = store
-
-        rules = []
-        for raw_rule in raw_rules:
-            rule = dict(raw_rule)
-            rule['conditions'] = json.loads(raw_rule['conditions'])
-            rule['actions'] = json.loads(raw_rule['actions'])
-            rules.append(rule)
-
-        self.rules = list_with_base_rules(rules)
-
-        self.enabled_map = enabled_map
-
-    @staticmethod
-    def tweaks_for_actions(actions):
-        tweaks = {}
-        for a in actions:
-            if not isinstance(a, dict):
-                continue
-            if 'set_tweak' in a and 'value' in a:
-                tweaks[a['set_tweak']] = a['value']
-        return tweaks
-
-    @defer.inlineCallbacks
-    def actions_for_event(self, ev):
-        """
-        This should take into account notification settings that the user
-        has configured both globally and per-room when we have the ability
-        to do such things.
-        """
-        if ev['user_id'] == self.user_id:
-            # let's assume you probably know about messages you sent yourself
-            defer.returnValue([])
-
-        room_id = ev['room_id']
-
-        # get *our* member event for display name matching
-        my_display_name = None
-
-        if self.our_member_event:
-            my_display_name = self.our_member_event[0].content.get("displayname")
-
-        room_members = yield self.store.get_users_in_room(room_id)
-        room_member_count = len(room_members)
-
-        evaluator = PushRuleEvaluatorForEvent(ev, room_member_count)
-
-        for r in self.rules:
-            enabled = self.enabled_map.get(r['rule_id'], None)
-            if enabled is not None and not enabled:
-                continue
-
-            if not r.get("enabled", True):
-                continue
-
-            conditions = r['conditions']
-            actions = r['actions']
-
-            # ignore rules with no actions (we have an explict 'dont_notify')
-            if len(actions) == 0:
-                logger.warn(
-                    "Ignoring rule id %s with no actions for user %s",
-                    r['rule_id'], self.user_id
-                )
-                continue
-
-            matches = True
-            for c in conditions:
-                matches = evaluator.matches(
-                    c, self.user_id, my_display_name
-                )
-                if not matches:
-                    break
-
-            logger.debug(
-                "Rule %s %s",
-                r['rule_id'], "matches" if matches else "doesn't match"
-            )
-
-            if matches:
-                logger.debug(
-                    "%s matches for user %s, event %s",
-                    r['rule_id'], self.user_id, ev['event_id']
-                )
-
-                # filter out dont_notify as we treat an empty actions list
-                # as dont_notify, and this doesn't take up a row in our database
-                actions = [x for x in actions if x != 'dont_notify']
-
-                defer.returnValue(actions)
-
-        logger.debug(
-            "No rules match for user %s, event %s",
-            self.user_id, ev['event_id']
-        )
-        defer.returnValue(PushRuleEvaluator.DEFAULT_ACTIONS)
+def tweaks_for_actions(actions):
+    tweaks = {}
+    for a in actions:
+        if not isinstance(a, dict):
+            continue
+        if 'set_tweak' in a and 'value' in a:
+            tweaks[a['set_tweak']] = a['value']
+    return tweaks
 
 
 class PushRuleEvaluatorForEvent(object):
diff --git a/synapse/push/push_tools.py b/synapse/push/push_tools.py
new file mode 100644
index 0000000000..89a3b5e90a
--- /dev/null
+++ b/synapse/push/push_tools.py
@@ -0,0 +1,68 @@
+# -*- 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 twisted.internet import defer
+
+
+@defer.inlineCallbacks
+def get_badge_count(store, user_id):
+    invites, joins = yield defer.gatherResults([
+        store.get_invited_rooms_for_user(user_id),
+        store.get_rooms_for_user(user_id),
+    ], consumeErrors=True)
+
+    my_receipts_by_room = yield store.get_receipts_for_user(
+        user_id, "m.read",
+    )
+
+    badge = len(invites)
+
+    for r in joins:
+        if r.room_id in my_receipts_by_room:
+            last_unread_event_id = my_receipts_by_room[r.room_id]
+
+            notifs = yield (
+                store.get_unread_event_push_actions_by_room_for_user(
+                    r.room_id, user_id, last_unread_event_id
+                )
+            )
+            # return one badge count per conversation, as count per
+            # message is so noisy as to be almost useless
+            badge += 1 if notifs["notify_count"] else 0
+    defer.returnValue(badge)
+
+
+@defer.inlineCallbacks
+def get_context_for_event(store, ev):
+    name_aliases = yield store.get_room_name_and_aliases(
+        ev.room_id
+    )
+
+    ctx = {'aliases': name_aliases[1]}
+    if name_aliases[0] is not None:
+        ctx['name'] = name_aliases[0]
+
+    their_member_events_for_room = yield store.get_current_state(
+        room_id=ev.room_id,
+        event_type='m.room.member',
+        state_key=ev.user_id
+    )
+    for mev in their_member_events_for_room:
+        if mev.content['membership'] == 'join' and 'displayname' in mev.content:
+            dn = mev.content['displayname']
+            if dn is not None:
+                ctx['sender_display_name'] = dn
+
+    defer.returnValue(ctx)
diff --git a/synapse/push/pusher.py b/synapse/push/pusher.py
new file mode 100644
index 0000000000..de9c33b936
--- /dev/null
+++ b/synapse/push/pusher.py
@@ -0,0 +1,47 @@
+# -*- 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
+
+import logging
+logger = logging.getLogger(__name__)
+
+# We try importing this if we can (it will fail if we don't
+# have the optional email dependencies installed). We don't
+# yet have the config to know if we need the email pusher,
+# but importing this after daemonizing seems to fail
+# (even though a simple test of importing from a daemonized
+# process works fine)
+try:
+    from synapse.push.emailpusher import EmailPusher
+except:
+    pass
+
+
+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:
+        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 0b463c6fdb..5853ec36a9 100644
--- a/synapse/push/pusherpool.py
+++ b/synapse/push/pusherpool.py
@@ -16,9 +16,9 @@
 
 from twisted.internet import defer
 
-from .httppusher import HttpPusher
-from synapse.push import PusherConfigException
+import pusher
 from synapse.util.logcontext import preserve_fn
+from synapse.util.async import run_on_reactor
 
 import logging
 
@@ -28,10 +28,10 @@ logger = logging.getLogger(__name__)
 class PusherPool:
     def __init__(self, _hs):
         self.hs = _hs
+        self.start_pushers = _hs.config.start_pushers
         self.store = self.hs.get_datastore()
         self.clock = self.hs.get_clock()
         self.pushers = {}
-        self.last_pusher_started = -1
 
     @defer.inlineCallbacks
     def start(self):
@@ -48,7 +48,8 @@ class PusherPool:
         # will then get pulled out of the database,
         # recreated, added and started: this means we have only one
         # code path adding pushers.
-        self._create_pusher({
+        pusher.create_pusher(self.hs, {
+            "id": None,
             "user_name": user_id,
             "kind": kind,
             "app_id": app_id,
@@ -58,10 +59,18 @@ class PusherPool:
             "ts": time_now_msec,
             "lang": lang,
             "data": data,
-            "last_token": None,
+            "last_stream_ordering": None,
             "last_success": None,
             "failing_since": None
         })
+
+        # create the pusher setting last_stream_ordering to the current maximum
+        # stream ordering in event_push_actions, so it will process
+        # pushes from this point onwards.
+        last_stream_ordering = (
+            yield self.store.get_latest_push_action_stream_ordering()
+        )
+
         yield self.store.add_pusher(
             user_id=user_id,
             access_token=access_token,
@@ -73,6 +82,7 @@ class PusherPool:
             pushkey_ts=time_now_msec,
             lang=lang,
             data=data,
+            last_stream_ordering=last_stream_ordering,
             profile_tag=profile_tag,
         )
         yield self._refresh_pusher(app_id, pushkey, user_id)
@@ -106,26 +116,51 @@ class PusherPool:
                 )
                 yield self.remove_pusher(p['app_id'], p['pushkey'], p['user_name'])
 
-    def _create_pusher(self, pusherdict):
-        if pusherdict['kind'] == 'http':
-            return HttpPusher(
-                self.hs,
-                user_id=pusherdict['user_name'],
-                app_id=pusherdict['app_id'],
-                app_display_name=pusherdict['app_display_name'],
-                device_display_name=pusherdict['device_display_name'],
-                pushkey=pusherdict['pushkey'],
-                pushkey_ts=pusherdict['ts'],
-                data=pusherdict['data'],
-                last_token=pusherdict['last_token'],
-                last_success=pusherdict['last_success'],
-                failing_since=pusherdict['failing_since']
+    @defer.inlineCallbacks
+    def on_new_notifications(self, min_stream_id, max_stream_id):
+        yield run_on_reactor()
+        try:
+            users_affected = yield self.store.get_push_action_users_in_range(
+                min_stream_id, max_stream_id
             )
-        else:
-            raise PusherConfigException(
-                "Unknown pusher type '%s' for user %s" %
-                (pusherdict['kind'], pusherdict['user_name'])
+
+            deferreds = []
+
+            for u in users_affected:
+                if u in self.pushers:
+                    for p in self.pushers[u].values():
+                        deferreds.append(
+                            p.on_new_notifications(min_stream_id, max_stream_id)
+                        )
+
+            yield defer.gatherResults(deferreds)
+        except:
+            logger.exception("Exception in pusher on_new_notifications")
+
+    @defer.inlineCallbacks
+    def on_new_receipts(self, min_stream_id, max_stream_id, affected_room_ids):
+        yield run_on_reactor()
+        try:
+            # Need to subtract 1 from the minimum because the lower bound here
+            # is not inclusive
+            updated_receipts = yield self.store.get_all_updated_receipts(
+                min_stream_id - 1, max_stream_id
             )
+            # This returns a tuple, user_id is at index 3
+            users_affected = set([r[3] for r in updated_receipts])
+
+            deferreds = []
+
+            for u in users_affected:
+                if u in self.pushers:
+                    for p in self.pushers[u].values():
+                        deferreds.append(
+                            p.on_new_receipts(min_stream_id, max_stream_id)
+                        )
+
+            yield defer.gatherResults(deferreds)
+        except:
+            logger.exception("Exception in pusher on_new_receipts")
 
     @defer.inlineCallbacks
     def _refresh_pusher(self, app_id, pushkey, user_id):
@@ -143,33 +178,40 @@ class PusherPool:
             self._start_pushers([p])
 
     def _start_pushers(self, pushers):
+        if not self.start_pushers:
+            logger.info("Not starting pushers because they are disabled in the config")
+            return
         logger.info("Starting %d pushers", len(pushers))
         for pusherdict in pushers:
             try:
-                p = self._create_pusher(pusherdict)
-            except PusherConfigException:
-                logger.exception("Couldn't start a pusher: caught PusherConfigException")
+                p = pusher.create_pusher(self.hs, pusherdict)
+            except:
+                logger.exception("Couldn't start a pusher: caught Exception")
                 continue
             if p:
-                fullid = "%s:%s:%s" % (
+                appid_pushkey = "%s:%s" % (
                     pusherdict['app_id'],
                     pusherdict['pushkey'],
-                    pusherdict['user_name']
                 )
-                if fullid in self.pushers:
-                    self.pushers[fullid].stop()
-                self.pushers[fullid] = p
-                preserve_fn(p.start)()
+                byuser = self.pushers.setdefault(pusherdict['user_name'], {})
+
+                if appid_pushkey in byuser:
+                    byuser[appid_pushkey].on_stop()
+                byuser[appid_pushkey] = p
+                preserve_fn(p.on_started)()
 
         logger.info("Started pushers")
 
     @defer.inlineCallbacks
     def remove_pusher(self, app_id, pushkey, user_id):
-        fullid = "%s:%s:%s" % (app_id, pushkey, user_id)
-        if fullid in self.pushers:
-            logger.info("Stopping pusher %s", fullid)
-            self.pushers[fullid].stop()
-            del self.pushers[fullid]
+        appid_pushkey = "%s:%s" % (app_id, pushkey)
+
+        byuser = self.pushers.get(user_id, {})
+
+        if appid_pushkey in byuser:
+            logger.info("Stopping pusher %s / %s", user_id, appid_pushkey)
+            byuser[appid_pushkey].on_stop()
+            del byuser[appid_pushkey]
         yield self.store.delete_pusher_by_app_id_pushkey_user_id(
             app_id, pushkey, user_id
         )