Add endpoints for backfilling history (MSC2716)

changelog.d/9247.feature

@@ -0,0 +1 @@
+Implement MSC2716 to support backfilling history into rooms

scripts-dev/complement.sh

@@ -11,12 +11,10 @@ cd "$(dirname $0)/.."
 docker build -t matrixdotorg/synapse:latest -f docker/Dockerfile .
 
 # Download Complement
-wget -N https://github.com/matrix-org/complement/archive/master.tar.gz
-tar -xzf master.tar.gz
-cd complement-master
+cd ../complement
 
 # Build the Synapse image from Complement, based on the above image we just built
 docker build -t complement-synapse -f dockerfiles/Synapse.Dockerfile ./dockerfiles
 
 # Run the tests on the resulting image!
-COMPLEMENT_BASE_IMAGE=complement-synapse go test -v -count=1 ./tests
+COMPLEMENT_BASE_IMAGE=complement-synapse go test -tags msc2716 -v -count=1 ./tests/main_test.go ./tests/msc2716_test.go

synapse/events/builder.py

@@ -98,7 +98,10 @@ def is_state(self):
         return self._state_key is not None
 
     async def build(
-        self, prev_event_ids: List[str], auth_event_ids: Optional[List[str]],
+        self,
+        prev_event_ids: List[str],
+        overriding_prev_events: Literal[False] = False,
+        auth_event_ids: Optional[List[str]],
     ) -> EventBase:
         """Transform into a fully signed and hashed event
 
@@ -131,7 +134,13 @@ async def build(
             prev_events = prev_event_ids
 
         old_depth = await self._store.get_max_depth_of(prev_event_ids)
-        depth = old_depth + 1
+        # awfewafeafew depth stuff
+        # If backfilling old message, let's just use the same depth of what we're inserting next to
+        if overriding_prev_events:
+            depth = old_depth
+        # Otherwise, progress the depth as normal
+        else:
+            depth = old_depth + 1
 
         # we cap depth of generated events, to ensure that they are not
         # rejected by other servers (and so that they can be persisted in

synapse/events/utils.py

@@ -250,13 +250,13 @@ def format_event_for_client_v1(d):
 
 def format_event_for_client_v2(d):
     drop_keys = (
-        "auth_events",
-        "prev_events",
-        "hashes",
-        "signatures",
-        "depth",
-        "origin",
-        "prev_state",
+        # "auth_events",
+        # "prev_events",
+        # "hashes",
+        # "signatures",
+        # "depth",
+        # "origin",
+        # "prev_state",
     )
     for key in drop_keys:
         d.pop(key, None)

synapse/handlers/message.py

@@ -736,8 +736,12 @@ async def create_and_send_nonmember_event(
                     assert event.internal_metadata.stream_ordering
                     return event, event.internal_metadata.stream_ordering
 
+            prev_events = None
+            if "prev_events" in event_dict:
+                prev_events = event_dict["prev_events"]
+
             event, context = await self.create_event(
-                requester, event_dict, txn_id=txn_id
+                requester, event_dict, txn_id=txn_id, prev_event_ids=prev_events
             )
 
             assert self.hs.is_mine_id(event.sender), "User must be our own: %s" % (
@@ -790,7 +794,10 @@ async def create_new_client_event(
             Tuple of created event, context
         """
 
+        overriding_prev_events = False
         if prev_event_ids is not None:
+            overriding_prev_events = True
+
             assert len(prev_event_ids) <= 10, (
                 "Attempting to create an event with %i prev_events"
                 % (len(prev_event_ids),)
@@ -808,7 +815,9 @@ async def create_new_client_event(
         ), "Attempting to create an event with no prev_events"
 
         event = await builder.build(
-            prev_event_ids=prev_event_ids, auth_event_ids=auth_event_ids
+            prev_event_ids=prev_event_ids,
+            overriding_prev_events=overriding_prev_events,
+            auth_event_ids=auth_event_ids,
         )
         context = await self.state.compute_event_context(event)
         if requester:

synapse/handlers/sync.py

@@ -42,7 +42,7 @@
 from synapse.util.caches.lrucache import LruCache
 from synapse.util.caches.response_cache import ResponseCache
 from synapse.util.metrics import Measure, measure_func
-from synapse.visibility import filter_events_for_client
+from synapse.visibility import filter_events_for_client, filter_historical_events
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
@@ -534,6 +534,9 @@ async def _load_filtered_recents(
 
             prev_batch_token = now_token.copy_and_replace("room_key", room_key)
 
+        # `m.historical` events should not come down /sync
+        recents = await filter_historical_events(recents)
+
         return TimelineBatch(
             events=recents,
             prev_batch=prev_batch_token,

synapse/http/servlet.py

@@ -16,6 +16,7 @@
 """ This module contains base REST classes for constructing REST servlets. """
 
 import logging
+from typing import List
 
 from synapse.api.errors import Codes, SynapseError
 from synapse.util import json_decoder
@@ -147,38 +148,43 @@ def parse_string(
     )
 
 
-def parse_string_from_args(
+def parse_string_value(value, allowed_values, name="", encoding="ascii") -> str:
+    if encoding:
+        try:
+            value = value.decode(encoding)
+        except ValueError:
+            raise SynapseError(400, "Query parameter %r must be %s" % (name, encoding))
+
+    if allowed_values is not None and value not in allowed_values:
+        message = "Query parameter %r must be one of [%s]" % (
+            name,
+            ", ".join(repr(v) for v in allowed_values),
+        )
+        raise SynapseError(400, message)
+    else:
+        return value
+
+
+def parse_strings_from_args(
     args,
     name,
     default=None,
     required=False,
     allowed_values=None,
     param_type="string",
     encoding="ascii",
-):
+) -> List[str]:
 
     if not isinstance(name, bytes):
         name = name.encode("ascii")
 
     if name in args:
-        value = args[name][0]
-
-        if encoding:
-            try:
-                value = value.decode(encoding)
-            except ValueError:
-                raise SynapseError(
-                    400, "Query parameter %r must be %s" % (name, encoding)
-                )
-
-        if allowed_values is not None and value not in allowed_values:
-            message = "Query parameter %r must be one of [%s]" % (
-                name,
-                ", ".join(repr(v) for v in allowed_values),
-            )
-            raise SynapseError(400, message)
-        else:
-            return value
+        values = args[name]
+
+        return [
+            parse_string_value(value, allowed_values, name=name, encoding=encoding)
+            for value in values
+        ]
     else:
         if required:
             message = "Missing %s query parameter %r" % (param_type, name)
@@ -191,6 +197,32 @@ def parse_string_from_args(
             return default
 
 
+def parse_string_from_args(
+    args,
+    name,
+    default=None,
+    required=False,
+    allowed_values=None,
+    param_type="string",
+    encoding="ascii",
+):
+    strings = parse_strings_from_args(
+        args,
+        name,
+        default=default,
+        required=required,
+        allowed_values=allowed_values,
+        param_type=param_type,
+        encoding=encoding,
+    )
+
+    if isinstance(strings, list) and len(strings):
+        return strings[0]
+
+    # Return the default
+    return strings
+
+
 def parse_json_value_from_request(request, allow_empty_body=False):
     """Parse a JSON value from the body of a twisted HTTP request.
 

synapse/rest/client/v1/room.py

@@ -36,7 +36,9 @@
     RestServlet,
     assert_params_in_dict,
     parse_integer,
+    parse_integer_from_args,
     parse_json_object_from_request,
+    parse_strings_from_args,
     parse_string,
 )
 from synapse.logging.opentracing import set_tag
@@ -222,6 +224,7 @@ def register(self, http_server):
     async def on_POST(self, request, room_id, event_type, txn_id=None):
         requester = await self.auth.get_user_by_req(request, allow_guest=True)
         content = parse_json_object_from_request(request)
+        prev_events = parse_strings_from_args(request.args, "prev_event")
 
         event_dict = {
             "type": event_type,
@@ -230,7 +233,12 @@ async def on_POST(self, request, room_id, event_type, txn_id=None):
             "sender": requester.user.to_string(),
         }
 
-        if b"ts" in request.args and requester.app_service:
+        if prev_events:
+            event_dict["prev_events"] = prev_events
+
+        # TODO: Put app_service logic back in place once we figure out how to make the Complement tests
+        # run as an app service
+        if b"ts" in request.args: # and requester.app_service:
             event_dict["origin_server_ts"] = parse_integer(request, "ts", 0)
 
         try:

synapse/visibility.py

@@ -46,6 +46,16 @@
 )
 
 
+async def filter_historical_events(events):
+    filtered_events = [e for e in events if not e.content.get("m.historical", None)]
+
+    # remove the None entries
+    filtered_events = filter(operator.truth, filtered_events)
+
+    # we turn it into a list before returning it.
+    return list(filtered_events)
+
+
 async def filter_events_for_client(
     storage: Storage,
     user_id,