Initial stab at real SQL storage implementation of user filter definitions

3 files changed, 60 insertions, 14 deletions

diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py
index efa63031bd..7c5631d014 100644
--- a/synapse/storage/__init__.py
+++ b/synapse/storage/__init__.py
@@ -61,6 +61,7 @@ SCHEMAS = [
     "event_edges",
     "event_signatures",
     "media_repository",
+    "filtering",
 ]
 
 
diff --git a/synapse/storage/filtering.py b/synapse/storage/filtering.py
index 18e0e7c298..e98eaf8032 100644
--- a/synapse/storage/filtering.py
+++ b/synapse/storage/filtering.py
@@ -17,6 +17,8 @@ from twisted.internet import defer
 
 from ._base import SQLBaseStore
 
+import json
+
 
 # TODO(paul)
 _filters_for_user = {}
@@ -25,22 +27,41 @@ _filters_for_user = {}
 class FilteringStore(SQLBaseStore):
     @defer.inlineCallbacks
     def get_user_filter(self, user_localpart, filter_id):
-        filters = _filters_for_user.get(user_localpart, None)
-
-        if not filters or filter_id >= len(filters):
-            raise KeyError()
+        def_json = yield self._simple_select_one_onecol(
+            table="user_filters",
+            keyvalues={
+                "user_id": user_localpart,
+                "filter_id": filter_id,
+            },
+            retcol="definition",
+            allow_none=False,
+        )
 
-        # trivial yield to make it a generator so d.iC works
-        yield
-        defer.returnValue(filters[filter_id])
+        defer.returnValue(json.loads(def_json))
 
-    @defer.inlineCallbacks
     def add_user_filter(self, user_localpart, definition):
-        filters = _filters_for_user.setdefault(user_localpart, [])
+        def_json = json.dumps(definition)
+
+        # Need an atomic transaction to SELECT the maximal ID so far then
+        # INSERT a new one
+        def _do_txn(txn):
+            sql = (
+                "SELECT MAX(filter_id) FROM user_filters "
+                "WHERE user_id = ?"
+            )
+            txn.execute(sql, (user_localpart,))
+            max_id = txn.fetchone()[0]
+            if max_id is None:
+                filter_id = 0
+            else:
+                filter_id = max_id + 1
+
+            sql = (
+                "INSERT INTO user_filters (user_id, filter_id, definition)"
+                "VALUES(?, ?, ?)"
+            )
+            txn.execute(sql, (user_localpart, filter_id, def_json))
 
-        filter_id = len(filters)
-        filters.append(definition)
+            return filter_id
 
-        # trivial yield, see above
-        yield
-        defer.returnValue(filter_id)
+        return self.runInteraction("add_user_filter", _do_txn)
diff --git a/synapse/storage/schema/filtering.sql b/synapse/storage/schema/filtering.sql
new file mode 100644
index 0000000000..795aca4afd
--- /dev/null
+++ b/synapse/storage/schema/filtering.sql
@@ -0,0 +1,24 @@
+/* Copyright 2015 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.
+ */
+CREATE TABLE IF NOT EXISTS user_filters(
+  user_id TEXT,
+  filter_id INTEGER,
+  definition TEXT,
+  FOREIGN KEY(user_id) REFERENCES users(id)
+);
+
+CREATE INDEX IF NOT EXISTS user_filters_by_user_id_filter_id ON user_filters(
+  user_id, filter_id
+);