summary refs log tree commit diff
path: root/synapse
diff options
context:
space:
mode:
Diffstat (limited to 'synapse')
-rw-r--r--synapse/appservice/api.py12
-rw-r--r--synapse/config/appservice.py17
-rw-r--r--synapse/push/mailer.py2
-rw-r--r--synapse/replication/resource.py19
-rw-r--r--synapse/replication/slave/storage/deviceinbox.py12
-rw-r--r--synapse/rest/client/v2_alpha/sendtodevice.py10
-rw-r--r--synapse/storage/deviceinbox.py44
-rw-r--r--synapse/storage/receipts.py2
-rw-r--r--synapse/storage/schema/delta/34/sent_txn_purge.py32
-rw-r--r--synapse/storage/transactions.py2
10 files changed, 145 insertions, 7 deletions
diff --git a/synapse/appservice/api.py b/synapse/appservice/api.py
index 775417eb21..cc4af23962 100644
--- a/synapse/appservice/api.py
+++ b/synapse/appservice/api.py
@@ -67,6 +67,8 @@ class ApplicationServiceApi(SimpleHttpClient):
 
     @defer.inlineCallbacks
     def query_user(self, service, user_id):
+        if service.url is None:
+            defer.returnValue(False)
         uri = service.url + ("/users/%s" % urllib.quote(user_id))
         response = None
         try:
@@ -86,6 +88,8 @@ class ApplicationServiceApi(SimpleHttpClient):
 
     @defer.inlineCallbacks
     def query_alias(self, service, alias):
+        if service.url is None:
+            defer.returnValue(False)
         uri = service.url + ("/rooms/%s" % urllib.quote(alias))
         response = None
         try:
@@ -113,6 +117,8 @@ class ApplicationServiceApi(SimpleHttpClient):
             raise ValueError(
                 "Unrecognised 'kind' argument %r to query_3pe()", kind
             )
+        if service.url is None:
+            defer.returnValue([])
 
         uri = "%s%s/thirdparty/%s/%s" % (
             service.url,
@@ -145,6 +151,9 @@ class ApplicationServiceApi(SimpleHttpClient):
             defer.returnValue([])
 
     def get_3pe_protocol(self, service, protocol):
+        if service.url is None:
+            defer.returnValue({})
+
         @defer.inlineCallbacks
         def _get():
             uri = "%s%s/thirdparty/protocol/%s" % (
@@ -166,6 +175,9 @@ class ApplicationServiceApi(SimpleHttpClient):
 
     @defer.inlineCallbacks
     def push_bulk(self, service, events, txn_id=None):
+        if service.url is None:
+            defer.returnValue(True)
+
         events = self._serialize(events)
 
         if txn_id is None:
diff --git a/synapse/config/appservice.py b/synapse/config/appservice.py
index dfe43b0b4c..d7537e8d44 100644
--- a/synapse/config/appservice.py
+++ b/synapse/config/appservice.py
@@ -86,7 +86,7 @@ def load_appservices(hostname, config_files):
 
 def _load_appservice(hostname, as_info, config_filename):
     required_string_fields = [
-        "id", "url", "as_token", "hs_token", "sender_localpart"
+        "id", "as_token", "hs_token", "sender_localpart"
     ]
     for field in required_string_fields:
         if not isinstance(as_info.get(field), basestring):
@@ -94,6 +94,14 @@ def _load_appservice(hostname, as_info, config_filename):
                 field, config_filename,
             ))
 
+    # 'url' must either be a string or explicitly null, not missing
+    # to avoid accidentally turning off push for ASes.
+    if (not isinstance(as_info.get("url"), basestring) and
+            as_info.get("url", "") is not None):
+        raise KeyError(
+            "Required string field or explicit null: 'url' (%s)" % (config_filename,)
+        )
+
     localpart = as_info["sender_localpart"]
     if urllib.quote(localpart) != localpart:
         raise ValueError(
@@ -132,6 +140,13 @@ def _load_appservice(hostname, as_info, config_filename):
         for p in protocols:
             if not isinstance(p, str):
                 raise KeyError("Bad value for 'protocols' item")
+
+    if as_info["url"] is None:
+        logger.info(
+            "(%s) Explicitly empty 'url' provided. This application service"
+            " will not receive events or queries.",
+            config_filename,
+        )
     return ApplicationService(
         token=as_info["as_token"],
         url=as_info["url"],
diff --git a/synapse/push/mailer.py b/synapse/push/mailer.py
index 81b1af4a19..2cafcfd8f5 100644
--- a/synapse/push/mailer.py
+++ b/synapse/push/mailer.py
@@ -338,7 +338,7 @@ class Mailer(object):
             # want the generated-from-names one here otherwise we'll
             # end up with, "new message from Bob in the Bob room"
             room_name = yield calculate_room_name(
-                state_by_room[room_id], user_id, fallback_to_members=False
+                self.store, state_by_room[room_id], user_id, fallback_to_members=False
             )
 
             my_member_event = state_by_room[room_id][("m.room.member", user_id)]
diff --git a/synapse/replication/resource.py b/synapse/replication/resource.py
index 817f50428f..1ed9034bcb 100644
--- a/synapse/replication/resource.py
+++ b/synapse/replication/resource.py
@@ -41,6 +41,7 @@ STREAM_NAMES = (
     ("push_rules",),
     ("pushers",),
     ("caches",),
+    ("to_device",),
 )
 
 
@@ -142,6 +143,7 @@ class ReplicationResource(Resource):
             pushers_token,
             0,  # State stream is no longer a thing
             caches_token,
+            int(stream_token.to_device_key),
         ))
 
     @request_handler()
@@ -190,6 +192,7 @@ class ReplicationResource(Resource):
         yield self.push_rules(writer, current_token, limit, request_streams)
         yield self.pushers(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)
@@ -376,6 +379,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"""
@@ -404,7 +421,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()
diff --git a/synapse/storage/receipts.py b/synapse/storage/receipts.py
index ccc3811e84..9747a04a9a 100644
--- a/synapse/storage/receipts.py
+++ b/synapse/storage/receipts.py
@@ -145,7 +145,7 @@ class ReceiptsStore(SQLBaseStore):
 
         defer.returnValue([ev for res in results.values() for ev in res])
 
-    @cachedInlineCallbacks(num_args=3, max_entries=5000, tree=True)
+    @cachedInlineCallbacks(num_args=3, tree=True)
     def get_linearized_receipts_for_room(self, room_id, to_key, from_key=None):
         """Get receipts for a single room for sending to clients.
 
diff --git a/synapse/storage/schema/delta/34/sent_txn_purge.py b/synapse/storage/schema/delta/34/sent_txn_purge.py
new file mode 100644
index 0000000000..81948e3431
--- /dev/null
+++ b/synapse/storage/schema/delta/34/sent_txn_purge.py
@@ -0,0 +1,32 @@
+# 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 synapse.storage.engines import PostgresEngine
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+def run_create(cur, database_engine, *args, **kwargs):
+    if isinstance(database_engine, PostgresEngine):
+        cur.execute("TRUNCATE sent_transactions")
+    else:
+        cur.execute("DELETE FROM sent_transactions")
+
+    cur.execute("CREATE INDEX sent_transactions_ts ON sent_transactions(ts)")
+
+
+def run_upgrade(cur, database_engine, *args, **kwargs):
+    pass
diff --git a/synapse/storage/transactions.py b/synapse/storage/transactions.py
index 58d4de4f1d..1c588bd46b 100644
--- a/synapse/storage/transactions.py
+++ b/synapse/storage/transactions.py
@@ -387,8 +387,10 @@ class TransactionStore(SQLBaseStore):
     def _cleanup_transactions(self):
         now = self._clock.time_msec()
         month_ago = now - 30 * 24 * 60 * 60 * 1000
+        six_hours_ago = now - 6 * 60 * 60 * 1000
 
         def _cleanup_transactions_txn(txn):
             txn.execute("DELETE FROM received_transactions WHERE ts < ?", (month_ago,))
+            txn.execute("DELETE FROM sent_transactions WHERE ts < ?", (six_hours_ago,))
 
         return self.runInteraction("_persist_in_mem_txns", _cleanup_transactions_txn)