Merge pull request #1058 from matrix-org/markjh/direct_to_device_synchrotron

Add a replication stream for direct to device messages

4 files changed, 81 insertions, 4 deletions

diff --git a/synapse/replication/resource.py b/synapse/replication/resource.py
index 84993b33b3..533676814a 100644
--- a/synapse/replication/resource.py
+++ b/synapse/replication/resource.py
@@ -42,6 +42,7 @@ STREAM_NAMES = (
     ("pushers",),
     ("state",),
     ("caches",),
+    ("to_device",),
 )
 
 
@@ -144,6 +145,7 @@ class ReplicationResource(Resource):
             pushers_token,
             state_token,
             caches_token,
+            int(stream_token.to_device_key),
         ))
 
     @request_handler()
@@ -193,6 +195,7 @@ class ReplicationResource(Resource):
         yield self.pushers(writer, current_token, limit, request_streams)
         yield self.state(writer, current_token, limit, request_streams)
         yield self.caches(writer, current_token, limit, request_streams)
+        yield self.to_device(writer, current_token, limit, request_streams)
         self.streams(writer, current_token, request_streams)
 
         logger.info("Replicated %d rows", writer.total)
@@ -398,6 +401,20 @@ class ReplicationResource(Resource):
                 "position", "cache_func", "keys", "invalidation_ts"
             ))
 
+    @defer.inlineCallbacks
+    def to_device(self, writer, current_token, limit, request_streams):
+        current_position = current_token.to_device
+
+        to_device = request_streams.get("to_device")
+
+        if to_device is not None:
+            to_device_rows = yield self.store.get_all_new_device_messages(
+                to_device, current_position, limit
+            )
+            writer.write_header_and_rows("to_device", to_device_rows, (
+                "position", "user_id", "device_id", "message_json"
+            ))
+
 
 class _Writer(object):
     """Writes the streams as a JSON object as the response to the request"""
@@ -426,7 +443,7 @@ class _Writer(object):
 
 class _ReplicationToken(collections.namedtuple("_ReplicationToken", (
     "events", "presence", "typing", "receipts", "account_data", "backfill",
-    "push_rules", "pushers", "state", "caches",
+    "push_rules", "pushers", "state", "caches", "to_device",
 ))):
     __slots__ = []
 
diff --git a/synapse/replication/slave/storage/deviceinbox.py b/synapse/replication/slave/storage/deviceinbox.py
index 7583d23708..64d8eb2af1 100644
--- a/synapse/replication/slave/storage/deviceinbox.py
+++ b/synapse/replication/slave/storage/deviceinbox.py
@@ -28,3 +28,15 @@ class SlavedDeviceInboxStore(BaseSlavedStore):
     get_to_device_stream_token = DataStore.get_to_device_stream_token.__func__
     get_new_messages_for_device = DataStore.get_new_messages_for_device.__func__
     delete_messages_for_device = DataStore.delete_messages_for_device.__func__
+
+    def stream_positions(self):
+        result = super(SlavedDeviceInboxStore, self).stream_positions()
+        result["to_device"] = self._device_inbox_id_gen.get_current_token()
+        return result
+
+    def process_replication(self, result):
+        stream = result.get("to_device")
+        if stream:
+            self._device_inbox_id_gen.advance(int(stream["position"]))
+
+        return super(SlavedDeviceInboxStore, self).process_replication(result)
diff --git a/synapse/rest/client/v2_alpha/sendtodevice.py b/synapse/rest/client/v2_alpha/sendtodevice.py
index 00533741af..7c0991ca55 100644
--- a/synapse/rest/client/v2_alpha/sendtodevice.py
+++ b/synapse/rest/client/v2_alpha/sendtodevice.py
@@ -40,6 +40,7 @@ class SendToDeviceRestServlet(servlet.RestServlet):
         self.hs = hs
         self.auth = hs.get_auth()
         self.store = hs.get_datastore()
+        self.notifier = hs.get_notifier()
         self.is_mine_id = hs.is_mine_id
         self.txns = HttpTransactionStore()
 
@@ -71,9 +72,14 @@ class SendToDeviceRestServlet(servlet.RestServlet):
                     }
                     for device_id, message_content in by_device.items()
                 }
-                local_messages[user_id] = messages_by_device
+                if messages_by_device:
+                    local_messages[user_id] = messages_by_device
 
-        yield self.store.add_messages_to_device_inbox(local_messages)
+        stream_id = yield self.store.add_messages_to_device_inbox(local_messages)
+
+        self.notifier.on_new_event(
+            "to_device", stream_id, users=local_messages.keys()
+        )
 
         response = (200, {})
         self.txns.store_client_transaction(request, txn_id, response)
diff --git a/synapse/storage/deviceinbox.py b/synapse/storage/deviceinbox.py
index e682fe1bd1..2fa0a218b9 100644
--- a/synapse/storage/deviceinbox.py
+++ b/synapse/storage/deviceinbox.py
@@ -33,7 +33,8 @@ class DeviceInboxStore(SQLBaseStore):
             messages_by_user_and_device(dict):
                 Dictionary of user_id to device_id to message.
         Returns:
-            A deferred that resolves when the messages have been inserted.
+            A deferred stream_id that resolves when the messages have been
+            inserted.
         """
 
         def select_devices_txn(txn, user_id, devices):
@@ -81,6 +82,8 @@ class DeviceInboxStore(SQLBaseStore):
                 stream_id
             )
 
+        defer.returnValue(self._device_inbox_id_gen.get_current_token())
+
     def get_new_messages_for_device(
         self, user_id, device_id, current_stream_id, limit=100
     ):
@@ -136,5 +139,44 @@ class DeviceInboxStore(SQLBaseStore):
             "delete_messages_for_device", delete_messages_for_device_txn
         )
 
+    def get_all_new_device_messages(self, last_pos, current_pos, limit):
+        """
+        Args:
+            last_pos(int):
+            current_pos(int):
+            limit(int):
+        Returns:
+            A deferred list of rows from the device inbox
+        """
+        if last_pos == current_pos:
+            return defer.succeed([])
+
+        def get_all_new_device_messages_txn(txn):
+            sql = (
+                "SELECT stream_id FROM device_inbox"
+                " WHERE ? < stream_id AND stream_id <= ?"
+                " GROUP BY stream_id"
+                " ORDER BY stream_id ASC"
+                " LIMIT ?"
+            )
+            txn.execute(sql, (last_pos, current_pos, limit))
+            stream_ids = txn.fetchall()
+            if not stream_ids:
+                return []
+            max_stream_id_in_limit = stream_ids[-1]
+
+            sql = (
+                "SELECT stream_id, user_id, device_id, message_json"
+                " FROM device_inbox"
+                " WHERE ? < stream_id AND stream_id <= ?"
+                " ORDER BY stream_id ASC"
+            )
+            txn.execute(sql, (last_pos, max_stream_id_in_limit))
+            return txn.fetchall()
+
+        return self.runInteraction(
+            "get_all_new_device_messages", get_all_new_device_messages_txn
+        )
+
     def get_to_device_stream_token(self):
         return self._device_inbox_id_gen.get_current_token()