after a few rethinks, a working implementation of pushers.

4 files changed, 104 insertions, 30 deletions

diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index 4881f03368..eb8cc4a9f3 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -195,6 +195,51 @@ class SQLBaseStore(object):
         txn.execute(sql, values.values())
         return txn.lastrowid
 
+    def _simple_upsert(self, table, keyvalues, values):
+        """
+        :param table: The table to upsert into
+        :param keyvalues: Dict of the unique key tables and their new values
+        :param values: Dict of all the nonunique columns and their new values
+        :return: A deferred
+        """
+        return self.runInteraction(
+            "_simple_upsert",
+            self._simple_upsert_txn, table, keyvalues, values
+        )
+
+    def _simple_upsert_txn(self, txn, table, keyvalues, values):
+        # Try to update
+        sql = "UPDATE %s SET %s WHERE %s" % (
+            table,
+            ", ".join("%s = ?" % (k) for k in values),
+            " AND ".join("%s = ?" % (k) for k in keyvalues)
+        )
+        sqlargs = values.values() + keyvalues.values()
+        logger.debug(
+            "[SQL] %s Args=%s",
+            sql, sqlargs,
+        )
+
+        txn.execute(sql, sqlargs)
+        if txn.rowcount == 0:
+            # We didn't update and rows so insert a new one
+            allvalues = {}
+            allvalues.update(keyvalues)
+            allvalues.update(values)
+
+            sql = "INSERT INTO %s (%s) VALUES (%s)" % (
+                table,
+                ", ".join(k for k in allvalues),
+                ", ".join("?" for _ in allvalues)
+            )
+            logger.debug(
+                "[SQL] %s Args=%s",
+                sql, keyvalues.values(),
+            )
+            txn.execute(sql, allvalues.values())
+
+
+
     def _simple_select_one(self, table, keyvalues, retcols,
                            allow_none=False):
         """Executes a SELECT query on the named table, which is expected to
diff --git a/synapse/storage/pusher.py b/synapse/storage/pusher.py
index a858e46f3b..deabd9cd2e 100644
--- a/synapse/storage/pusher.py
+++ b/synapse/storage/pusher.py
@@ -28,16 +28,48 @@ logger = logging.getLogger(__name__)
 
 class PusherStore(SQLBaseStore):
     @defer.inlineCallbacks
-    def get_all_pushers_after_id(self, min_id):
+    def get_pushers_by_app_id_and_pushkey(self, app_id_and_pushkey):
         sql = (
-            "SELECT id, user_name, kind, app_id, app_instance_id,"
+            "SELECT id, user_name, kind, app_id,"
             "app_display_name, device_display_name, pushkey, data, "
             "last_token, last_success, failing_since "
             "FROM pushers "
-            "WHERE id > ?"
+            "WHERE app_id = ? AND pushkey = ?"
         )
 
-        rows = yield self._execute(None, sql, min_id)
+        rows = yield self._execute(
+            None, sql, app_id_and_pushkey[0], app_id_and_pushkey[1]
+        )
+
+        ret = [
+            {
+                "id": r[0],
+                "user_name": r[1],
+                "kind": r[2],
+                "app_id": r[3],
+                "app_display_name": r[4],
+                "device_display_name": r[5],
+                "pushkey": r[6],
+                "data": r[7],
+                "last_token": r[8],
+                "last_success": r[9],
+                "failing_since": r[10]
+            }
+            for r in rows
+        ]
+
+        defer.returnValue(ret[0])
+
+    @defer.inlineCallbacks
+    def get_all_pushers(self):
+        sql = (
+            "SELECT id, user_name, kind, app_id,"
+            "app_display_name, device_display_name, pushkey, data, "
+            "last_token, last_success, failing_since "
+            "FROM pushers"
+        )
+
+        rows = yield self._execute(None, sql)
 
         ret = [
             {
@@ -45,14 +77,13 @@ class PusherStore(SQLBaseStore):
                 "user_name": r[1],
                 "kind": r[2],
                 "app_id": r[3],
-                "app_instance_id": r[4],
-                "app_display_name": r[5],
-                "device_display_name": r[6],
-                "pushkey": r[7],
-                "data": r[8],
-                "last_token": r[9],
-                "last_success": r[10],
-                "failing_since": r[11]
+                "app_display_name": r[4],
+                "device_display_name": r[5],
+                "pushkey": r[6],
+                "data": r[7],
+                "last_token": r[8],
+                "last_success": r[9],
+                "failing_since": r[10]
             }
             for r in rows
         ]
@@ -60,21 +91,22 @@ class PusherStore(SQLBaseStore):
         defer.returnValue(ret)
 
     @defer.inlineCallbacks
-    def add_pusher(self, user_name, kind, app_id, app_instance_id,
+    def add_pusher(self, user_name, kind, app_id,
                    app_display_name, device_display_name, pushkey, data):
         try:
-            yield self._simple_insert(PushersTable.table_name, dict(
-                user_name=user_name,
-                kind=kind,
-                app_id=app_id,
-                app_instance_id=app_instance_id,
-                app_display_name=app_display_name,
-                device_display_name=device_display_name,
-                pushkey=pushkey,
-                data=data
-            ))
-        except IntegrityError:
-            raise StoreError(409, "Pushkey in use.")
+            yield self._simple_upsert(
+                PushersTable.table_name,
+                dict(
+                    app_id=app_id,
+                    pushkey=pushkey,
+                ),
+                dict(
+                    user_name=user_name,
+                    kind=kind,
+                    app_display_name=app_display_name,
+                    device_display_name=device_display_name,
+                    data=data
+                ))
         except Exception as e:
             logger.error("create_pusher with failed: %s", e)
             raise StoreError(500, "Problem creating pusher.")
@@ -113,7 +145,6 @@ class PushersTable(Table):
         "user_name",
         "kind",
         "app_id",
-        "app_instance_id",
         "app_display_name",
         "device_display_name",
         "pushkey",
diff --git a/synapse/storage/schema/delta/v7.sql b/synapse/storage/schema/delta/v7.sql
index b60aeda756..799e48d780 100644
--- a/synapse/storage/schema/delta/v7.sql
+++ b/synapse/storage/schema/delta/v7.sql
@@ -18,7 +18,6 @@ CREATE TABLE IF NOT EXISTS pushers (
   user_name TEXT NOT NULL,
   kind varchar(8) NOT NULL,
   app_id varchar(64) NOT NULL,
-  app_instance_id varchar(64) NOT NULL,
   app_display_name varchar(64) NOT NULL,
   device_display_name varchar(128) NOT NULL,
   pushkey blob NOT NULL,
@@ -27,5 +26,5 @@ CREATE TABLE IF NOT EXISTS pushers (
   last_success BIGINT,
   failing_since BIGINT,
   FOREIGN KEY(user_name) REFERENCES users(name),
-  UNIQUE (user_name, pushkey)
+  UNIQUE (app_id, pushkey)
 );
diff --git a/synapse/storage/schema/pusher.sql b/synapse/storage/schema/pusher.sql
index b60aeda756..799e48d780 100644
--- a/synapse/storage/schema/pusher.sql
+++ b/synapse/storage/schema/pusher.sql
@@ -18,7 +18,6 @@ CREATE TABLE IF NOT EXISTS pushers (
   user_name TEXT NOT NULL,
   kind varchar(8) NOT NULL,
   app_id varchar(64) NOT NULL,
-  app_instance_id varchar(64) NOT NULL,
   app_display_name varchar(64) NOT NULL,
   device_display_name varchar(128) NOT NULL,
   pushkey blob NOT NULL,
@@ -27,5 +26,5 @@ CREATE TABLE IF NOT EXISTS pushers (
   last_success BIGINT,
   failing_since BIGINT,
   FOREIGN KEY(user_name) REFERENCES users(name),
-  UNIQUE (user_name, pushkey)
+  UNIQUE (app_id, pushkey)
 );