Merge branch 'rejections_storage' of github.com:matrix-org/synapse into replication_split

11 files changed, 559 insertions, 43 deletions

diff --git a/synapse/rest/client/v1/__init__.py b/synapse/rest/client/v1/__init__.py
index 8bb89b2f6a..d8d01cdd16 100644
--- a/synapse/rest/client/v1/__init__.py
+++ b/synapse/rest/client/v1/__init__.py
@@ -13,10 +13,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-
 from . import (
     room, events, register, login, profile, presence, initial_sync, directory,
-    voip, admin,
+    voip, admin, pusher, push_rule
 )
 
 from synapse.http.server import JsonResource
@@ -41,3 +40,5 @@ class ClientV1RestResource(JsonResource):
         directory.register_servlets(hs, client_resource)
         voip.register_servlets(hs, client_resource)
         admin.register_servlets(hs, client_resource)
+        pusher.register_servlets(hs, client_resource)
+        push_rule.register_servlets(hs, client_resource)
diff --git a/synapse/rest/client/v1/admin.py b/synapse/rest/client/v1/admin.py
index 1051d96f96..2ce754b028 100644
--- a/synapse/rest/client/v1/admin.py
+++ b/synapse/rest/client/v1/admin.py
@@ -31,7 +31,7 @@ class WhoisRestServlet(ClientV1RestServlet):
     @defer.inlineCallbacks
     def on_GET(self, request, user_id):
         target_user = UserID.from_string(user_id)
-        auth_user = yield self.auth.get_user_by_req(request)
+        auth_user, client = yield self.auth.get_user_by_req(request)
         is_admin = yield self.auth.is_server_admin(auth_user)
 
         if not is_admin and target_user != auth_user:
diff --git a/synapse/rest/client/v1/directory.py b/synapse/rest/client/v1/directory.py
index 15ae8749b8..8f65efec5f 100644
--- a/synapse/rest/client/v1/directory.py
+++ b/synapse/rest/client/v1/directory.py
@@ -45,7 +45,7 @@ class ClientDirectoryServer(ClientV1RestServlet):
 
     @defer.inlineCallbacks
     def on_PUT(self, request, room_alias):
-        user = yield self.auth.get_user_by_req(request)
+        user, client = yield self.auth.get_user_by_req(request)
 
         content = _parse_json(request)
         if not "room_id" in content:
@@ -85,7 +85,7 @@ class ClientDirectoryServer(ClientV1RestServlet):
 
     @defer.inlineCallbacks
     def on_DELETE(self, request, room_alias):
-        user = yield self.auth.get_user_by_req(request)
+        user, client = yield self.auth.get_user_by_req(request)
 
         is_admin = yield self.auth.is_server_admin(user)
         if not is_admin:
diff --git a/synapse/rest/client/v1/events.py b/synapse/rest/client/v1/events.py
index c69de56863..77b7c25a03 100644
--- a/synapse/rest/client/v1/events.py
+++ b/synapse/rest/client/v1/events.py
@@ -19,6 +19,7 @@ from twisted.internet import defer
 from synapse.api.errors import SynapseError
 from synapse.streams.config import PaginationConfig
 from .base import ClientV1RestServlet, client_path_pattern
+from synapse.events.utils import serialize_event
 
 import logging
 
@@ -33,7 +34,7 @@ class EventStreamRestServlet(ClientV1RestServlet):
 
     @defer.inlineCallbacks
     def on_GET(self, request):
-        auth_user = yield self.auth.get_user_by_req(request)
+        auth_user, client = yield self.auth.get_user_by_req(request)
         try:
             handler = self.handlers.event_stream_handler
             pagin_config = PaginationConfig.from_request(request)
@@ -64,14 +65,19 @@ class EventStreamRestServlet(ClientV1RestServlet):
 class EventRestServlet(ClientV1RestServlet):
     PATTERN = client_path_pattern("/events/(?P<event_id>[^/]*)$")
 
+    def __init__(self, hs):
+        super(EventRestServlet, self).__init__(hs)
+        self.clock = hs.get_clock()
+
     @defer.inlineCallbacks
     def on_GET(self, request, event_id):
-        auth_user = yield self.auth.get_user_by_req(request)
+        auth_user, client = yield self.auth.get_user_by_req(request)
         handler = self.handlers.event_handler
         event = yield handler.get_event(auth_user, event_id)
 
+        time_now = self.clock.time_msec()
         if event:
-            defer.returnValue((200, self.hs.serialize_event(event)))
+            defer.returnValue((200, serialize_event(event, time_now)))
         else:
             defer.returnValue((404, "Event not found."))
 
diff --git a/synapse/rest/client/v1/initial_sync.py b/synapse/rest/client/v1/initial_sync.py
index 357fa845b4..4a259bba64 100644
--- a/synapse/rest/client/v1/initial_sync.py
+++ b/synapse/rest/client/v1/initial_sync.py
@@ -25,7 +25,7 @@ class InitialSyncRestServlet(ClientV1RestServlet):
 
     @defer.inlineCallbacks
     def on_GET(self, request):
-        user = yield self.auth.get_user_by_req(request)
+        user, client = yield self.auth.get_user_by_req(request)
         with_feedback = "feedback" in request.args
         as_client_event = "raw" not in request.args
         pagination_config = PaginationConfig.from_request(request)
diff --git a/synapse/rest/client/v1/presence.py b/synapse/rest/client/v1/presence.py
index b6c207e662..7feb4aadb1 100644
--- a/synapse/rest/client/v1/presence.py
+++ b/synapse/rest/client/v1/presence.py
@@ -32,7 +32,7 @@ class PresenceStatusRestServlet(ClientV1RestServlet):
 
     @defer.inlineCallbacks
     def on_GET(self, request, user_id):
-        auth_user = yield self.auth.get_user_by_req(request)
+        auth_user, client = yield self.auth.get_user_by_req(request)
         user = UserID.from_string(user_id)
 
         state = yield self.handlers.presence_handler.get_state(
@@ -42,7 +42,7 @@ class PresenceStatusRestServlet(ClientV1RestServlet):
 
     @defer.inlineCallbacks
     def on_PUT(self, request, user_id):
-        auth_user = yield self.auth.get_user_by_req(request)
+        auth_user, client = yield self.auth.get_user_by_req(request)
         user = UserID.from_string(user_id)
 
         state = {}
@@ -77,7 +77,7 @@ class PresenceListRestServlet(ClientV1RestServlet):
 
     @defer.inlineCallbacks
     def on_GET(self, request, user_id):
-        auth_user = yield self.auth.get_user_by_req(request)
+        auth_user, client = yield self.auth.get_user_by_req(request)
         user = UserID.from_string(user_id)
 
         if not self.hs.is_mine(user):
@@ -97,7 +97,7 @@ class PresenceListRestServlet(ClientV1RestServlet):
 
     @defer.inlineCallbacks
     def on_POST(self, request, user_id):
-        auth_user = yield self.auth.get_user_by_req(request)
+        auth_user, client = yield self.auth.get_user_by_req(request)
         user = UserID.from_string(user_id)
 
         if not self.hs.is_mine(user):
diff --git a/synapse/rest/client/v1/profile.py b/synapse/rest/client/v1/profile.py
index 24f8d56952..15d6f3fc6c 100644
--- a/synapse/rest/client/v1/profile.py
+++ b/synapse/rest/client/v1/profile.py
@@ -37,7 +37,7 @@ class ProfileDisplaynameRestServlet(ClientV1RestServlet):
 
     @defer.inlineCallbacks
     def on_PUT(self, request, user_id):
-        auth_user = yield self.auth.get_user_by_req(request)
+        auth_user, client = yield self.auth.get_user_by_req(request)
         user = UserID.from_string(user_id)
 
         try:
@@ -70,7 +70,7 @@ class ProfileAvatarURLRestServlet(ClientV1RestServlet):
 
     @defer.inlineCallbacks
     def on_PUT(self, request, user_id):
-        auth_user = yield self.auth.get_user_by_req(request)
+        auth_user, client = yield self.auth.get_user_by_req(request)
         user = UserID.from_string(user_id)
 
         try:
diff --git a/synapse/rest/client/v1/push_rule.py b/synapse/rest/client/v1/push_rule.py
new file mode 100644
index 0000000000..0f78fa667c
--- /dev/null
+++ b/synapse/rest/client/v1/push_rule.py
@@ -0,0 +1,401 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014 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.errors import SynapseError, Codes, UnrecognizedRequestError, NotFoundError, \
+    StoreError
+from .base import ClientV1RestServlet, client_path_pattern
+from synapse.storage.push_rule import InconsistentRuleException, RuleNotFoundException
+
+import json
+
+
+class PushRuleRestServlet(ClientV1RestServlet):
+    PATTERN = client_path_pattern("/pushrules/.*$")
+    PRIORITY_CLASS_MAP = {
+        'underride': 1,
+        'sender': 2,
+        'room': 3,
+        'content': 4,
+        'override': 5,
+    }
+    PRIORITY_CLASS_INVERSE_MAP = {v: k for k, v in PRIORITY_CLASS_MAP.items()}
+    SLIGHTLY_PEDANTIC_TRAILING_SLASH_ERROR = (
+        "Unrecognised request: You probably wanted a trailing slash")
+
+    def rule_spec_from_path(self, path):
+        if len(path) < 2:
+            raise UnrecognizedRequestError()
+        if path[0] != 'pushrules':
+            raise UnrecognizedRequestError()
+
+        scope = path[1]
+        path = path[2:]
+        if scope not in ['global', 'device']:
+            raise UnrecognizedRequestError()
+
+        device = None
+        if scope == 'device':
+            if len(path) == 0:
+                raise UnrecognizedRequestError()
+            device = path[0]
+            path = path[1:]
+
+        if len(path) == 0:
+            raise UnrecognizedRequestError()
+
+        template = path[0]
+        path = path[1:]
+
+        if len(path) == 0:
+            raise UnrecognizedRequestError()
+
+        rule_id = path[0]
+
+        spec = {
+            'scope': scope,
+            'template': template,
+            'rule_id': rule_id
+        }
+        if device:
+            spec['device'] = device
+        return spec
+
+    def rule_tuple_from_request_object(self, rule_template, rule_id, req_obj, device=None):
+        if rule_template in ['override', 'underride']:
+            if 'conditions' not in req_obj:
+                raise InvalidRuleException("Missing 'conditions'")
+            conditions = req_obj['conditions']
+            for c in conditions:
+                if 'kind' not in c:
+                    raise InvalidRuleException("Condition without 'kind'")
+        elif rule_template == 'room':
+            conditions = [{
+                'kind': 'event_match',
+                'key': 'room_id',
+                'pattern': rule_id
+            }]
+        elif rule_template == 'sender':
+            conditions = [{
+                'kind': 'event_match',
+                'key': 'user_id',
+                'pattern': rule_id
+            }]
+        elif rule_template == 'content':
+            if 'pattern' not in req_obj:
+                raise InvalidRuleException("Content rule missing 'pattern'")
+            pat = req_obj['pattern']
+            if pat.strip("*?[]") == pat:
+                # no special glob characters so we assume the user means
+                # 'contains this string' rather than 'is this string'
+                pat = "*%s*" % (pat,)
+            conditions = [{
+                'kind': 'event_match',
+                'key': 'content.body',
+                'pattern': pat
+            }]
+        else:
+            raise InvalidRuleException("Unknown rule template: %s" % (rule_template,))
+
+        if device:
+            conditions.append({
+                'kind': 'device',
+                'instance_handle': device
+            })
+
+        if 'actions' not in req_obj:
+            raise InvalidRuleException("No actions found")
+        actions = req_obj['actions']
+
+        for a in actions:
+            if a in ['notify', 'dont_notify', 'coalesce']:
+                pass
+            elif isinstance(a, dict) and 'set_sound' in a:
+                pass
+            else:
+                raise InvalidRuleException("Unrecognised action")
+
+        return conditions, actions
+
+    @defer.inlineCallbacks
+    def on_PUT(self, request):
+        spec = self.rule_spec_from_path(request.postpath)
+        try:
+            priority_class = _priority_class_from_spec(spec)
+        except InvalidRuleException as e:
+            raise SynapseError(400, e.message)
+
+        user, _ = yield self.auth.get_user_by_req(request)
+
+        content = _parse_json(request)
+
+        try:
+            (conditions, actions) = self.rule_tuple_from_request_object(
+                spec['template'],
+                spec['rule_id'],
+                content,
+                device=spec['device'] if 'device' in spec else None
+            )
+        except InvalidRuleException as e:
+            raise SynapseError(400, e.message)
+
+        before = request.args.get("before", None)
+        if before and len(before):
+            before = before[0]
+        after = request.args.get("after", None)
+        if after and len(after):
+            after = after[0]
+
+        try:
+            yield self.hs.get_datastore().add_push_rule(
+                user_name=user.to_string(),
+                rule_id=spec['rule_id'],
+                priority_class=priority_class,
+                conditions=conditions,
+                actions=actions,
+                before=before,
+                after=after
+            )
+        except InconsistentRuleException as e:
+            raise SynapseError(400, e.message)
+        except RuleNotFoundException as e:
+            raise SynapseError(400, e.message)
+
+        defer.returnValue((200, {}))
+
+    @defer.inlineCallbacks
+    def on_DELETE(self, request):
+        spec = self.rule_spec_from_path(request.postpath)
+        try:
+            priority_class = _priority_class_from_spec(spec)
+        except InvalidRuleException as e:
+            raise SynapseError(400, e.message)
+
+        user, _ = yield self.auth.get_user_by_req(request)
+
+        if 'device' in spec:
+            rules = yield self.hs.get_datastore().get_push_rules_for_user_name(
+                user.to_string()
+            )
+
+            for r in rules:
+                conditions = json.loads(r['conditions'])
+                ih = _instance_handle_from_conditions(conditions)
+                if ih == spec['device'] and r['priority_class'] == priority_class:
+                    yield self.hs.get_datastore().delete_push_rule(
+                        user.to_string(), spec['rule_id']
+                    )
+                    defer.returnValue((200, {}))
+            raise NotFoundError()
+        else:
+            try:
+                yield self.hs.get_datastore().delete_push_rule(
+                    user.to_string(), spec['rule_id'],
+                    priority_class=priority_class
+                )
+                defer.returnValue((200, {}))
+            except StoreError as e:
+                if e.code == 404:
+                    raise NotFoundError()
+                else:
+                    raise
+
+    @defer.inlineCallbacks
+    def on_GET(self, request):
+        user, _ = yield self.auth.get_user_by_req(request)
+
+        # we build up the full structure and then decide which bits of it
+        # to send which means doing unnecessary work sometimes but is
+        # is probably not going to make a whole lot of difference
+        rawrules = yield self.hs.get_datastore().get_push_rules_for_user_name(user.to_string())
+
+        rules = {'global': {}, 'device': {}}
+
+        rules['global'] = _add_empty_priority_class_arrays(rules['global'])
+
+        for r in rawrules:
+            rulearray = None
+
+            r["conditions"] = json.loads(r["conditions"])
+            r["actions"] = json.loads(r["actions"])
+
+            template_name = _priority_class_to_template_name(r['priority_class'])
+
+            if r['priority_class'] > PushRuleRestServlet.PRIORITY_CLASS_MAP['override']:
+                # per-device rule
+                instance_handle = _instance_handle_from_conditions(r["conditions"])
+                r = _strip_device_condition(r)
+                if not instance_handle:
+                    continue
+                if instance_handle not in rules['device']:
+                    rules['device'][instance_handle] = {}
+                    rules['device'][instance_handle] = (
+                        _add_empty_priority_class_arrays(
+                            rules['device'][instance_handle]
+                        )
+                    )
+
+                rulearray = rules['device'][instance_handle][template_name]
+            else:
+                rulearray = rules['global'][template_name]
+
+            template_rule = _rule_to_template(r)
+            if template_rule:
+                rulearray.append(template_rule)
+
+        path = request.postpath[1:]
+
+        if path == []:
+            # we're a reference impl: pedantry is our job.
+            raise UnrecognizedRequestError(
+                PushRuleRestServlet.SLIGHTLY_PEDANTIC_TRAILING_SLASH_ERROR
+            )
+
+        if path[0] == '':
+            defer.returnValue((200, rules))
+        elif path[0] == 'global':
+            path = path[1:]
+            result = _filter_ruleset_with_path(rules['global'], path)
+            defer.returnValue((200, result))
+        elif path[0] == 'device':
+            path = path[1:]
+            if path == []:
+                raise UnrecognizedRequestError(
+                    PushRuleRestServlet.SLIGHTLY_PEDANTIC_TRAILING_SLASH_ERROR
+                )
+            if path[0] == '':
+                defer.returnValue((200, rules['device']))
+
+            instance_handle = path[0]
+            path = path[1:]
+            if instance_handle not in rules['device']:
+                ret = {}
+                ret = _add_empty_priority_class_arrays(ret)
+                defer.returnValue((200, ret))
+            ruleset = rules['device'][instance_handle]
+            result = _filter_ruleset_with_path(ruleset, path)
+            defer.returnValue((200, result))
+        else:
+            raise UnrecognizedRequestError()
+
+    def on_OPTIONS(self, _):
+        return 200, {}
+
+
+def _add_empty_priority_class_arrays(d):
+    for pc in PushRuleRestServlet.PRIORITY_CLASS_MAP.keys():
+        d[pc] = []
+    return d
+
+
+def _instance_handle_from_conditions(conditions):
+    """
+    Given a list of conditions, return the instance handle of the
+    device rule if there is one
+    """
+    for c in conditions:
+        if c['kind'] == 'device':
+            return c['instance_handle']
+    return None
+
+
+def _filter_ruleset_with_path(ruleset, path):
+    if path == []:
+        raise UnrecognizedRequestError(
+            PushRuleRestServlet.SLIGHTLY_PEDANTIC_TRAILING_SLASH_ERROR
+        )
+
+    if path[0] == '':
+        return ruleset
+    template_kind = path[0]
+    if template_kind not in ruleset:
+        raise UnrecognizedRequestError()
+    path = path[1:]
+    if path == []:
+        raise UnrecognizedRequestError(
+            PushRuleRestServlet.SLIGHTLY_PEDANTIC_TRAILING_SLASH_ERROR
+        )
+    if path[0] == '':
+        return ruleset[template_kind]
+    rule_id = path[0]
+    for r in ruleset[template_kind]:
+        if r['rule_id'] == rule_id:
+            return r
+    raise NotFoundError
+
+
+def _priority_class_from_spec(spec):
+    if spec['template'] not in PushRuleRestServlet.PRIORITY_CLASS_MAP.keys():
+        raise InvalidRuleException("Unknown template: %s" % (spec['kind']))
+    pc = PushRuleRestServlet.PRIORITY_CLASS_MAP[spec['template']]
+
+    if spec['scope'] == 'device':
+        pc += len(PushRuleRestServlet.PRIORITY_CLASS_MAP)
+
+    return pc
+
+
+def _priority_class_to_template_name(pc):
+    if pc > PushRuleRestServlet.PRIORITY_CLASS_MAP['override']:
+        # per-device
+        prio_class_index = pc - len(PushRuleRestServlet.PRIORITY_CLASS_MAP)
+        return PushRuleRestServlet.PRIORITY_CLASS_INVERSE_MAP[prio_class_index]
+    else:
+        return PushRuleRestServlet.PRIORITY_CLASS_INVERSE_MAP[pc]
+
+
+def _rule_to_template(rule):
+    template_name = _priority_class_to_template_name(rule['priority_class'])
+    if template_name in ['override', 'underride']:
+        return {k: rule[k] for k in ["rule_id", "conditions", "actions"]}
+    elif template_name in ["sender", "room"]:
+        return {k: rule[k] for k in ["rule_id", "actions"]}
+    elif template_name == 'content':
+        if len(rule["conditions"]) != 1:
+            return None
+        thecond = rule["conditions"][0]
+        if "pattern" not in thecond:
+            return None
+        ret = {k: rule[k] for k in ["rule_id", "actions"]}
+        ret["pattern"] = thecond["pattern"]
+        return ret
+
+
+def _strip_device_condition(rule):
+    for i, c in enumerate(rule['conditions']):
+        if c['kind'] == 'device':
+            del rule['conditions'][i]
+    return rule
+
+
+class InvalidRuleException(Exception):
+    pass
+
+
+# XXX: C+ped from rest/room.py - surely this should be common?
+def _parse_json(request):
+    try:
+        content = json.loads(request.content.read())
+        if type(content) != dict:
+            raise SynapseError(400, "Content must be a JSON object.",
+                               errcode=Codes.NOT_JSON)
+        return content
+    except ValueError:
+        raise SynapseError(400, "Content not JSON.", errcode=Codes.NOT_JSON)
+
+
+def register_servlets(hs, http_server):
+    PushRuleRestServlet(hs).register(http_server)
diff --git a/synapse/rest/client/v1/pusher.py b/synapse/rest/client/v1/pusher.py
new file mode 100644
index 0000000000..353a4a6589
--- /dev/null
+++ b/synapse/rest/client/v1/pusher.py
@@ -0,0 +1,89 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014 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.errors import SynapseError, Codes
+from synapse.push import PusherConfigException
+from .base import ClientV1RestServlet, client_path_pattern
+
+import json
+
+
+class PusherRestServlet(ClientV1RestServlet):
+    PATTERN = client_path_pattern("/pushers/set$")
+
+    @defer.inlineCallbacks
+    def on_POST(self, request):
+        user, _ = yield self.auth.get_user_by_req(request)
+
+        content = _parse_json(request)
+
+        pusher_pool = self.hs.get_pusherpool()
+
+        if ('pushkey' in content and 'app_id' in content
+                    and 'kind' in content and
+                    content['kind'] is None):
+            yield pusher_pool.remove_pusher(
+                content['app_id'], content['pushkey']
+            )
+            defer.returnValue((200, {}))
+
+        reqd = ['instance_handle', 'kind', 'app_id', 'app_display_name',
+                'device_display_name', 'pushkey', 'lang', 'data']
+        missing = []
+        for i in reqd:
+            if i not in content:
+                missing.append(i)
+        if len(missing):
+            raise SynapseError(400, "Missing parameters: "+','.join(missing),
+                               errcode=Codes.MISSING_PARAM)
+
+        try:
+            yield pusher_pool.add_pusher(
+                user_name=user.to_string(),
+                instance_handle=content['instance_handle'],
+                kind=content['kind'],
+                app_id=content['app_id'],
+                app_display_name=content['app_display_name'],
+                device_display_name=content['device_display_name'],
+                pushkey=content['pushkey'],
+                lang=content['lang'],
+                data=content['data']
+            )
+        except PusherConfigException as pce:
+            raise SynapseError(400, "Config Error: "+pce.message,
+                               errcode=Codes.MISSING_PARAM)
+
+        defer.returnValue((200, {}))
+
+    def on_OPTIONS(self, _):
+        return 200, {}
+
+
+# XXX: C+ped from rest/room.py - surely this should be common?
+def _parse_json(request):
+    try:
+        content = json.loads(request.content.read())
+        if type(content) != dict:
+            raise SynapseError(400, "Content must be a JSON object.",
+                               errcode=Codes.NOT_JSON)
+        return content
+    except ValueError:
+        raise SynapseError(400, "Content not JSON.", errcode=Codes.NOT_JSON)
+
+
+def register_servlets(hs, http_server):
+    PusherRestServlet(hs).register(http_server)
diff --git a/synapse/rest/client/v1/room.py b/synapse/rest/client/v1/room.py
index f06e3ddb98..410f19ccf6 100644
--- a/synapse/rest/client/v1/room.py
+++ b/synapse/rest/client/v1/room.py
@@ -21,6 +21,7 @@ from synapse.api.errors import SynapseError, Codes
 from synapse.streams.config import PaginationConfig
 from synapse.api.constants import EventTypes, Membership
 from synapse.types import UserID, RoomID, RoomAlias
+from synapse.events.utils import serialize_event
 
 import json
 import logging
@@ -61,7 +62,7 @@ class RoomCreateRestServlet(ClientV1RestServlet):
 
     @defer.inlineCallbacks
     def on_POST(self, request):
-        auth_user = yield self.auth.get_user_by_req(request)
+        auth_user, client = yield self.auth.get_user_by_req(request)
 
         room_config = self.get_room_config(request)
         info = yield self.make_room(room_config, auth_user, None)
@@ -124,7 +125,7 @@ class RoomStateEventRestServlet(ClientV1RestServlet):
 
     @defer.inlineCallbacks
     def on_GET(self, request, room_id, event_type, state_key):
-        user = yield self.auth.get_user_by_req(request)
+        user, client = yield self.auth.get_user_by_req(request)
 
         msg_handler = self.handlers.message_handler
         data = yield msg_handler.get_room_data(
@@ -141,8 +142,8 @@ class RoomStateEventRestServlet(ClientV1RestServlet):
         defer.returnValue((200, data.get_dict()["content"]))
 
     @defer.inlineCallbacks
-    def on_PUT(self, request, room_id, event_type, state_key):
-        user = yield self.auth.get_user_by_req(request)
+    def on_PUT(self, request, room_id, event_type, state_key, txn_id=None):
+        user, client = yield self.auth.get_user_by_req(request)
 
         content = _parse_json(request)
 
@@ -157,7 +158,9 @@ class RoomStateEventRestServlet(ClientV1RestServlet):
             event_dict["state_key"] = state_key
 
         msg_handler = self.handlers.message_handler
-        yield msg_handler.create_and_send_event(event_dict)
+        yield msg_handler.create_and_send_event(
+            event_dict, client=client, txn_id=txn_id,
+        )
 
         defer.returnValue((200, {}))
 
@@ -171,8 +174,8 @@ class RoomSendEventRestServlet(ClientV1RestServlet):
         register_txn_path(self, PATTERN, http_server, with_get=True)
 
     @defer.inlineCallbacks
-    def on_POST(self, request, room_id, event_type):
-        user = yield self.auth.get_user_by_req(request)
+    def on_POST(self, request, room_id, event_type, txn_id=None):
+        user, client = yield self.auth.get_user_by_req(request)
         content = _parse_json(request)
 
         msg_handler = self.handlers.message_handler
@@ -182,7 +185,9 @@ class RoomSendEventRestServlet(ClientV1RestServlet):
                 "content": content,
                 "room_id": room_id,
                 "sender": user.to_string(),
-            }
+            },
+            client=client,
+            txn_id=txn_id,
         )
 
         defer.returnValue((200, {"event_id": event.event_id}))
@@ -199,7 +204,7 @@ class RoomSendEventRestServlet(ClientV1RestServlet):
         except KeyError:
             pass
 
-        response = yield self.on_POST(request, room_id, event_type)
+        response = yield self.on_POST(request, room_id, event_type, txn_id)
 
         self.txns.store_client_transaction(request, txn_id, response)
         defer.returnValue(response)
@@ -214,8 +219,8 @@ class JoinRoomAliasServlet(ClientV1RestServlet):
         register_txn_path(self, PATTERN, http_server)
 
     @defer.inlineCallbacks
-    def on_POST(self, request, room_identifier):
-        user = yield self.auth.get_user_by_req(request)
+    def on_POST(self, request, room_identifier, txn_id=None):
+        user, client = yield self.auth.get_user_by_req(request)
 
         # the identifier could be a room alias or a room id. Try one then the
         # other if it fails to parse, without swallowing other valid
@@ -244,7 +249,9 @@ class JoinRoomAliasServlet(ClientV1RestServlet):
                     "room_id": identifier.to_string(),
                     "sender": user.to_string(),
                     "state_key": user.to_string(),
-                }
+                },
+                client=client,
+                txn_id=txn_id,
             )
 
             defer.returnValue((200, {"room_id": identifier.to_string()}))
@@ -258,7 +265,7 @@ class JoinRoomAliasServlet(ClientV1RestServlet):
         except KeyError:
             pass
 
-        response = yield self.on_POST(request, room_identifier)
+        response = yield self.on_POST(request, room_identifier, txn_id)
 
         self.txns.store_client_transaction(request, txn_id, response)
         defer.returnValue(response)
@@ -282,7 +289,7 @@ class RoomMemberListRestServlet(ClientV1RestServlet):
     @defer.inlineCallbacks
     def on_GET(self, request, room_id):
         # TODO support Pagination stream API (limit/tokens)
-        user = yield self.auth.get_user_by_req(request)
+        user, client = yield self.auth.get_user_by_req(request)
         handler = self.handlers.room_member_handler
         members = yield handler.get_room_members_as_pagination_chunk(
             room_id=room_id,
@@ -310,7 +317,7 @@ class RoomMessageListRestServlet(ClientV1RestServlet):
 
     @defer.inlineCallbacks
     def on_GET(self, request, room_id):
-        user = yield self.auth.get_user_by_req(request)
+        user, client = yield self.auth.get_user_by_req(request)
         pagination_config = PaginationConfig.from_request(
             request, default_limit=10,
         )
@@ -334,7 +341,7 @@ class RoomStateRestServlet(ClientV1RestServlet):
 
     @defer.inlineCallbacks
     def on_GET(self, request, room_id):
-        user = yield self.auth.get_user_by_req(request)
+        user, client = yield self.auth.get_user_by_req(request)
         handler = self.handlers.message_handler
         # Get all the current state for this room
         events = yield handler.get_state_events(
@@ -350,7 +357,7 @@ class RoomInitialSyncRestServlet(ClientV1RestServlet):
 
     @defer.inlineCallbacks
     def on_GET(self, request, room_id):
-        user = yield self.auth.get_user_by_req(request)
+        user, client = yield self.auth.get_user_by_req(request)
         pagination_config = PaginationConfig.from_request(request)
         content = yield self.handlers.message_handler.room_initial_sync(
             room_id=room_id,
@@ -363,6 +370,10 @@ class RoomInitialSyncRestServlet(ClientV1RestServlet):
 class RoomTriggerBackfill(ClientV1RestServlet):
     PATTERN = client_path_pattern("/rooms/(?P<room_id>[^/]*)/backfill$")
 
+    def __init__(self, hs):
+        super(RoomTriggerBackfill, self).__init__(hs)
+        self.clock = hs.get_clock()
+
     @defer.inlineCallbacks
     def on_GET(self, request, room_id):
         remote_server = urllib.unquote(
@@ -374,7 +385,9 @@ class RoomTriggerBackfill(ClientV1RestServlet):
         handler = self.handlers.federation_handler
         events = yield handler.backfill(remote_server, room_id, limit)
 
-        res = [self.hs.serialize_event(event) for event in events]
+        time_now = self.clock.time_msec()
+
+        res = [serialize_event(event, time_now) for event in events]
         defer.returnValue((200, res))
 
 
@@ -388,8 +401,8 @@ class RoomMembershipRestServlet(ClientV1RestServlet):
         register_txn_path(self, PATTERN, http_server)
 
     @defer.inlineCallbacks
-    def on_POST(self, request, room_id, membership_action):
-        user = yield self.auth.get_user_by_req(request)
+    def on_POST(self, request, room_id, membership_action, txn_id=None):
+        user, client = yield self.auth.get_user_by_req(request)
 
         content = _parse_json(request)
 
@@ -411,7 +424,9 @@ class RoomMembershipRestServlet(ClientV1RestServlet):
                 "room_id": room_id,
                 "sender": user.to_string(),
                 "state_key": state_key,
-            }
+            },
+            client=client,
+            txn_id=txn_id,
         )
 
         defer.returnValue((200, {}))
@@ -425,7 +440,9 @@ class RoomMembershipRestServlet(ClientV1RestServlet):
         except KeyError:
             pass
 
-        response = yield self.on_POST(request, room_id, membership_action)
+        response = yield self.on_POST(
+            request, room_id, membership_action, txn_id
+        )
 
         self.txns.store_client_transaction(request, txn_id, response)
         defer.returnValue(response)
@@ -437,8 +454,8 @@ class RoomRedactEventRestServlet(ClientV1RestServlet):
         register_txn_path(self, PATTERN, http_server)
 
     @defer.inlineCallbacks
-    def on_POST(self, request, room_id, event_id):
-        user = yield self.auth.get_user_by_req(request)
+    def on_POST(self, request, room_id, event_id, txn_id=None):
+        user, client = yield self.auth.get_user_by_req(request)
         content = _parse_json(request)
 
         msg_handler = self.handlers.message_handler
@@ -449,7 +466,9 @@ class RoomRedactEventRestServlet(ClientV1RestServlet):
                 "room_id": room_id,
                 "sender": user.to_string(),
                 "redacts": event_id,
-            }
+            },
+            client=client,
+            txn_id=txn_id,
         )
 
         defer.returnValue((200, {"event_id": event.event_id}))
@@ -463,7 +482,7 @@ class RoomRedactEventRestServlet(ClientV1RestServlet):
         except KeyError:
             pass
 
-        response = yield self.on_POST(request, room_id, event_id)
+        response = yield self.on_POST(request, room_id, event_id, txn_id)
 
         self.txns.store_client_transaction(request, txn_id, response)
         defer.returnValue(response)
@@ -476,7 +495,7 @@ class RoomTypingRestServlet(ClientV1RestServlet):
 
     @defer.inlineCallbacks
     def on_PUT(self, request, room_id, user_id):
-        auth_user = yield self.auth.get_user_by_req(request)
+        auth_user, client = yield self.auth.get_user_by_req(request)
 
         room_id = urllib.unquote(room_id)
         target_user = UserID.from_string(urllib.unquote(user_id))
diff --git a/synapse/rest/client/v1/voip.py b/synapse/rest/client/v1/voip.py
index 822d863ce6..11d08fbced 100644
--- a/synapse/rest/client/v1/voip.py
+++ b/synapse/rest/client/v1/voip.py
@@ -28,7 +28,7 @@ class VoipRestServlet(ClientV1RestServlet):
 
     @defer.inlineCallbacks
     def on_GET(self, request):
-        auth_user = yield self.auth.get_user_by_req(request)
+        auth_user, client = yield self.auth.get_user_by_req(request)
 
         turnUris = self.hs.config.turn_uris
         turnSecret = self.hs.config.turn_shared_secret