diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py
new file mode 100644
index 0000000000..ec93f9f8a7
--- /dev/null
+++ b/synapse/storage/__init__.py
@@ -0,0 +1,117 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014 matrix.org
+#
+# 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.api.events.room import (
+ RoomMemberEvent, MessageEvent, RoomTopicEvent, FeedbackEvent,
+ RoomConfigEvent
+)
+
+from .directory import DirectoryStore
+from .feedback import FeedbackStore
+from .message import MessageStore
+from .presence import PresenceStore
+from .profile import ProfileStore
+from .registration import RegistrationStore
+from .room import RoomStore
+from .roommember import RoomMemberStore
+from .roomdata import RoomDataStore
+from .stream import StreamStore
+from .pdu import StatePduStore, PduStore
+from .transactions import TransactionStore
+
+import json
+import os
+
+
+class DataStore(RoomDataStore, RoomMemberStore, MessageStore, RoomStore,
+ RegistrationStore, StreamStore, ProfileStore, FeedbackStore,
+ PresenceStore, PduStore, StatePduStore, TransactionStore,
+ DirectoryStore):
+
+ def __init__(self, hs):
+ super(DataStore, self).__init__(hs)
+ self.event_factory = hs.get_event_factory()
+ self.hs = hs
+
+ def persist_event(self, event):
+ if event.type == MessageEvent.TYPE:
+ return self.store_message(
+ user_id=event.user_id,
+ room_id=event.room_id,
+ msg_id=event.msg_id,
+ content=json.dumps(event.content)
+ )
+ elif event.type == RoomMemberEvent.TYPE:
+ return self.store_room_member(
+ user_id=event.target_user_id,
+ sender=event.user_id,
+ room_id=event.room_id,
+ content=event.content,
+ membership=event.content["membership"]
+ )
+ elif event.type == FeedbackEvent.TYPE:
+ return self.store_feedback(
+ room_id=event.room_id,
+ msg_id=event.msg_id,
+ msg_sender_id=event.msg_sender_id,
+ fb_sender_id=event.user_id,
+ fb_type=event.feedback_type,
+ content=json.dumps(event.content)
+ )
+ elif event.type == RoomTopicEvent.TYPE:
+ return self.store_room_data(
+ room_id=event.room_id,
+ etype=event.type,
+ state_key=event.state_key,
+ content=json.dumps(event.content)
+ )
+ elif event.type == RoomConfigEvent.TYPE:
+ if "visibility" in event.content:
+ visibility = event.content["visibility"]
+ return self.store_room_config(
+ room_id=event.room_id,
+ visibility=visibility
+ )
+
+ else:
+ raise NotImplementedError(
+ "Don't know how to persist type=%s" % event.type
+ )
+
+
+def schema_path(schema):
+ """ Get a filesystem path for the named database schema
+
+ Args:
+ schema: Name of the database schema.
+ Returns:
+ A filesystem path pointing at a ".sql" file.
+
+ """
+ dir_path = os.path.dirname(__file__)
+ schemaPath = os.path.join(dir_path, "schema", schema + ".sql")
+ return schemaPath
+
+
+def read_schema(schema):
+ """ Read the named database schema.
+
+ Args:
+ schema: Name of the datbase schema.
+ Returns:
+ A string containing the database schema.
+ """
+ with open(schema_path(schema)) as schema_file:
+ return schema_file.read()
diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
new file mode 100644
index 0000000000..4d98a6fd0d
--- /dev/null
+++ b/synapse/storage/_base.py
@@ -0,0 +1,405 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014 matrix.org
+#
+# 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.
+import logging
+
+from twisted.internet import defer
+
+from synapse.api.errors import StoreError
+
+import collections
+
+logger = logging.getLogger(__name__)
+
+
+class SQLBaseStore(object):
+
+ def __init__(self, hs):
+ self._db_pool = hs.get_db_pool()
+
+ def cursor_to_dict(self, cursor):
+ """Converts a SQL cursor into an list of dicts.
+
+ Args:
+ cursor : The DBAPI cursor which has executed a query.
+ Returns:
+ A list of dicts where the key is the column header.
+ """
+ col_headers = list(column[0] for column in cursor.description)
+ results = list(
+ dict(zip(col_headers, row)) for row in cursor.fetchall()
+ )
+ return results
+
+ def _execute(self, decoder, query, *args):
+ """Runs a single query for a result set.
+
+ Args:
+ decoder - The function which can resolve the cursor results to
+ something meaningful.
+ query - The query string to execute
+ *args - Query args.
+ Returns:
+ The result of decoder(results)
+ """
+ logger.debug(
+ "[SQL] %s Args=%s Func=%s", query, args, decoder.__name__
+ )
+
+ def interaction(txn):
+ cursor = txn.execute(query, args)
+ return decoder(cursor)
+ return self._db_pool.runInteraction(interaction)
+
+ # "Simple" SQL API methods that operate on a single table with no JOINs,
+ # no complex WHERE clauses, just a dict of values for columns.
+
+ def _simple_insert(self, table, values):
+ """Executes an INSERT query on the named table.
+
+ Args:
+ table : string giving the table name
+ values : dict of new column names and values for them
+ """
+ sql = "INSERT INTO %s (%s) VALUES(%s)" % (
+ table,
+ ", ".join(k for k in values),
+ ", ".join("?" for k in values)
+ )
+
+ def func(txn):
+ txn.execute(sql, values.values())
+ return txn.lastrowid
+ return self._db_pool.runInteraction(func)
+
+ def _simple_select_one(self, table, keyvalues, retcols,
+ allow_none=False):
+ """Executes a SELECT query on the named table, which is expected to
+ return a single row, returning a single column from it.
+
+ Args:
+ table : string giving the table name
+ keyvalues : dict of column names and values to select the row with
+ retcols : list of strings giving the names of the columns to return
+
+ allow_none : If true, return None instead of failing if the SELECT
+ statement returns no rows
+ """
+ return self._simple_selectupdate_one(
+ table, keyvalues, retcols=retcols, allow_none=allow_none
+ )
+
+ @defer.inlineCallbacks
+ def _simple_select_one_onecol(self, table, keyvalues, retcol,
+ allow_none=False):
+ """Executes a SELECT query on the named table, which is expected to
+ return a single row, returning a single column from it."
+
+ Args:
+ table : string giving the table name
+ keyvalues : dict of column names and values to select the row with
+ retcol : string giving the name of the column to return
+ """
+ ret = yield self._simple_select_one(
+ table=table,
+ keyvalues=keyvalues,
+ retcols=[retcol],
+ allow_none=allow_none
+ )
+
+ if ret:
+ defer.returnValue(ret[retcol])
+ else:
+ defer.returnValue(None)
+
+ @defer.inlineCallbacks
+ def _simple_select_onecol(self, table, keyvalues, retcol):
+ """Executes a SELECT query on the named table, which returns a list
+ comprising of the values of the named column from the selected rows.
+
+ Args:
+ table (str): table name
+ keyvalues (dict): column names and values to select the rows with
+ retcol (str): column whos value we wish to retrieve.
+
+ Returns:
+ Deferred: Results in a list
+ """
+ sql = "SELECT %(retcol)s FROM %(table)s WHERE %(where)s" % {
+ "retcol": retcol,
+ "table": table,
+ "where": " AND ".join("%s = ?" % k for k in keyvalues.keys()),
+ }
+
+ def func(txn):
+ txn.execute(sql, keyvalues.values())
+ return txn.fetchall()
+
+ res = yield self._db_pool.runInteraction(func)
+
+ defer.returnValue([r[0] for r in res])
+
+ def _simple_select_list(self, table, keyvalues, retcols):
+ """Executes a SELECT query on the named table, which may return zero or
+ more rows, returning the result as a list of dicts.
+
+ Args:
+ table : string giving the table name
+ keyvalues : dict of column names and values to select the rows with
+ retcols : list of strings giving the names of the columns to return
+ """
+ sql = "SELECT %s FROM %s WHERE %s" % (
+ ", ".join(retcols),
+ table,
+ " AND ".join("%s = ?" % (k) for k in keyvalues)
+ )
+
+ def func(txn):
+ txn.execute(sql, keyvalues.values())
+ return self.cursor_to_dict(txn)
+
+ return self._db_pool.runInteraction(func)
+
+ def _simple_update_one(self, table, keyvalues, updatevalues,
+ retcols=None):
+ """Executes an UPDATE query on the named table, setting new values for
+ columns in a row matching the key values.
+
+ Args:
+ table : string giving the table name
+ keyvalues : dict of column names and values to select the row with
+ updatevalues : dict giving column names and values to update
+ retcols : optional list of column names to return
+
+ If present, retcols gives a list of column names on which to perform
+ a SELECT statement *before* performing the UPDATE statement. The values
+ of these will be returned in a dict.
+
+ These are performed within the same transaction, allowing an atomic
+ get-and-set. This can be used to implement compare-and-set by putting
+ the update column in the 'keyvalues' dict as well.
+ """
+ return self._simple_selectupdate_one(table, keyvalues, updatevalues,
+ retcols=retcols)
+
+ def _simple_selectupdate_one(self, table, keyvalues, updatevalues=None,
+ retcols=None, allow_none=False):
+ """ Combined SELECT then UPDATE."""
+ if retcols:
+ select_sql = "SELECT %s FROM %s WHERE %s" % (
+ ", ".join(retcols),
+ table,
+ " AND ".join("%s = ?" % (k) for k in keyvalues)
+ )
+
+ if updatevalues:
+ update_sql = "UPDATE %s SET %s WHERE %s" % (
+ table,
+ ", ".join("%s = ?" % (k) for k in updatevalues),
+ " AND ".join("%s = ?" % (k) for k in keyvalues)
+ )
+
+ def func(txn):
+ ret = None
+ if retcols:
+ txn.execute(select_sql, keyvalues.values())
+
+ row = txn.fetchone()
+ if not row:
+ if allow_none:
+ return None
+ raise StoreError(404, "No row found")
+ if txn.rowcount > 1:
+ raise StoreError(500, "More than one row matched")
+
+ ret = dict(zip(retcols, row))
+
+ if updatevalues:
+ txn.execute(
+ update_sql,
+ updatevalues.values() + keyvalues.values()
+ )
+
+ if txn.rowcount == 0:
+ raise StoreError(404, "No row found")
+ if txn.rowcount > 1:
+ raise StoreError(500, "More than one row matched")
+
+ return ret
+ return self._db_pool.runInteraction(func)
+
+ def _simple_delete_one(self, table, keyvalues):
+ """Executes a DELETE query on the named table, expecting to delete a
+ single row.
+
+ Args:
+ table : string giving the table name
+ keyvalues : dict of column names and values to select the row with
+ """
+ sql = "DELETE FROM %s WHERE %s" % (
+ table,
+ " AND ".join("%s = ?" % (k) for k in keyvalues)
+ )
+
+ def func(txn):
+ txn.execute(sql, keyvalues.values())
+ if txn.rowcount == 0:
+ raise StoreError(404, "No row found")
+ if txn.rowcount > 1:
+ raise StoreError(500, "more than one row matched")
+ return self._db_pool.runInteraction(func)
+
+ def _simple_max_id(self, table):
+ """Executes a SELECT query on the named table, expecting to return the
+ max value for the column "id".
+
+ Args:
+ table : string giving the table name
+ """
+ sql = "SELECT MAX(id) AS id FROM %s" % table
+
+ def func(txn):
+ txn.execute(sql)
+ max_id = self.cursor_to_dict(txn)[0]["id"]
+ if max_id is None:
+ return 0
+ return max_id
+
+ return self._db_pool.runInteraction(func)
+
+
+class Table(object):
+ """ A base class used to store information about a particular table.
+ """
+
+ table_name = None
+ """ str: The name of the table """
+
+ fields = None
+ """ list: The field names """
+
+ EntryType = None
+ """ Type: A tuple type used to decode the results """
+
+ _select_where_clause = "SELECT %s FROM %s WHERE %s"
+ _select_clause = "SELECT %s FROM %s"
+ _insert_clause = "INSERT OR REPLACE INTO %s (%s) VALUES (%s)"
+
+ @classmethod
+ def select_statement(cls, where_clause=None):
+ """
+ Args:
+ where_clause (str): The WHERE clause to use.
+
+ Returns:
+ str: An SQL statement to select rows from the table with the given
+ WHERE clause.
+ """
+ if where_clause:
+ return cls._select_where_clause % (
+ ", ".join(cls.fields),
+ cls.table_name,
+ where_clause
+ )
+ else:
+ return cls._select_clause % (
+ ", ".join(cls.fields),
+ cls.table_name,
+ )
+
+ @classmethod
+ def insert_statement(cls):
+ return cls._insert_clause % (
+ cls.table_name,
+ ", ".join(cls.fields),
+ ", ".join(["?"] * len(cls.fields)),
+ )
+
+ @classmethod
+ def decode_single_result(cls, results):
+ """ Given an iterable of tuples, return a single instance of
+ `EntryType` or None if the iterable is empty
+ Args:
+ results (list): The results list to convert to `EntryType`
+ Returns:
+ EntryType: An instance of `EntryType`
+ """
+ results = list(results)
+ if results:
+ return cls.EntryType(*results[0])
+ else:
+ return None
+
+ @classmethod
+ def decode_results(cls, results):
+ """ Given an iterable of tuples, return a list of `EntryType`
+ Args:
+ results (list): The results list to convert to `EntryType`
+
+ Returns:
+ list: A list of `EntryType`
+ """
+ return [cls.EntryType(*row) for row in results]
+
+ @classmethod
+ def get_fields_string(cls, prefix=None):
+ if prefix:
+ to_join = ("%s.%s" % (prefix, f) for f in cls.fields)
+ else:
+ to_join = cls.fields
+
+ return ", ".join(to_join)
+
+
+class JoinHelper(object):
+ """ Used to help do joins on tables by looking at the tables' fields and
+ creating a list of unique fields to use with SELECTs and a namedtuple
+ to dump the results into.
+
+ Attributes:
+ taples (list): List of `Table` classes
+ EntryType (type)
+ """
+
+ def __init__(self, *tables):
+ self.tables = tables
+
+ res = []
+ for table in self.tables:
+ res += [f for f in table.fields if f not in res]
+
+ self.EntryType = collections.namedtuple("JoinHelperEntry", res)
+
+ def get_fields(self, **prefixes):
+ """Get a string representing a list of fields for use in SELECT
+ statements with the given prefixes applied to each.
+
+ For example::
+
+ JoinHelper(PdusTable, StateTable).get_fields(
+ PdusTable="pdus",
+ StateTable="state"
+ )
+ """
+ res = []
+ for field in self.EntryType._fields:
+ for table in self.tables:
+ if field in table.fields:
+ res.append("%s.%s" % (prefixes[table.__name__], field))
+ break
+
+ return ", ".join(res)
+
+ def decode_results(self, rows):
+ return [self.EntryType(*row) for row in rows]
diff --git a/synapse/storage/directory.py b/synapse/storage/directory.py
new file mode 100644
index 0000000000..71fa9d9c9c
--- /dev/null
+++ b/synapse/storage/directory.py
@@ -0,0 +1,93 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014 matrix.org
+#
+# 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 ._base import SQLBaseStore
+from twisted.internet import defer
+
+from collections import namedtuple
+
+
+RoomAliasMapping = namedtuple(
+ "RoomAliasMapping",
+ ("room_id", "room_alias", "servers",)
+)
+
+
+class DirectoryStore(SQLBaseStore):
+
+ @defer.inlineCallbacks
+ def get_association_from_room_alias(self, room_alias):
+ """ Get's the room_id and server list for a given room_alias
+
+ Args:
+ room_alias (RoomAlias)
+
+ Returns:
+ Deferred: results in namedtuple with keys "room_id" and
+ "servers" or None if no association can be found
+ """
+ room_id = yield self._simple_select_one_onecol(
+ "room_aliases",
+ {"room_alias": room_alias.to_string()},
+ "room_id",
+ allow_none=True,
+ )
+
+ if not room_id:
+ defer.returnValue(None)
+ return
+
+ servers = yield self._simple_select_onecol(
+ "room_alias_servers",
+ {"room_alias": room_alias.to_string()},
+ "server",
+ )
+
+ if not servers:
+ defer.returnValue(None)
+ return
+
+ defer.returnValue(
+ RoomAliasMapping(room_id, room_alias.to_string(), servers)
+ )
+
+ @defer.inlineCallbacks
+ def create_room_alias_association(self, room_alias, room_id, servers):
+ """ Creates an associatin between a room alias and room_id/servers
+
+ Args:
+ room_alias (RoomAlias)
+ room_id (str)
+ servers (list)
+
+ Returns:
+ Deferred
+ """
+ yield self._simple_insert(
+ "room_aliases",
+ {
+ "room_alias": room_alias.to_string(),
+ "room_id": room_id,
+ },
+ )
+
+ for server in servers:
+ # TODO(erikj): Fix this to bulk insert
+ yield self._simple_insert(
+ "room_alias_servers",
+ {
+ "room_alias": room_alias.to_string(),
+ "server": server,
+ }
+ )
diff --git a/synapse/storage/feedback.py b/synapse/storage/feedback.py
new file mode 100644
index 0000000000..2b421e3342
--- /dev/null
+++ b/synapse/storage/feedback.py
@@ -0,0 +1,74 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014 matrix.org
+#
+# 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 ._base import SQLBaseStore, Table
+from synapse.api.events.room import FeedbackEvent
+
+import collections
+import json
+
+
+class FeedbackStore(SQLBaseStore):
+
+ def store_feedback(self, room_id, msg_id, msg_sender_id,
+ fb_sender_id, fb_type, content):
+ return self._simple_insert(FeedbackTable.table_name, dict(
+ room_id=room_id,
+ msg_id=msg_id,
+ msg_sender_id=msg_sender_id,
+ fb_sender_id=fb_sender_id,
+ fb_type=fb_type,
+ content=content,
+ ))
+
+ def get_feedback(self, room_id=None, msg_id=None, msg_sender_id=None,
+ fb_sender_id=None, fb_type=None):
+ query = FeedbackTable.select_statement(
+ "msg_sender_id = ? AND room_id = ? AND msg_id = ? " +
+ "AND fb_sender_id = ? AND feedback_type = ? " +
+ "ORDER BY id DESC LIMIT 1")
+ return self._execute(
+ FeedbackTable.decode_single_result,
+ query, msg_sender_id, room_id, msg_id, fb_sender_id, fb_type,
+ )
+
+ def get_max_feedback_id(self):
+ return self._simple_max_id(FeedbackTable.table_name)
+
+
+class FeedbackTable(Table):
+ table_name = "feedback"
+
+ fields = [
+ "id",
+ "content",
+ "feedback_type",
+ "fb_sender_id",
+ "msg_id",
+ "room_id",
+ "msg_sender_id"
+ ]
+
+ class EntryType(collections.namedtuple("FeedbackEntry", fields)):
+
+ def as_event(self, event_factory):
+ return event_factory.create_event(
+ etype=FeedbackEvent.TYPE,
+ room_id=self.room_id,
+ msg_id=self.msg_id,
+ msg_sender_id=self.msg_sender_id,
+ user_id=self.fb_sender_id,
+ feedback_type=self.feedback_type,
+ content=json.loads(self.content),
+ )
diff --git a/synapse/storage/message.py b/synapse/storage/message.py
new file mode 100644
index 0000000000..4822fa709d
--- /dev/null
+++ b/synapse/storage/message.py
@@ -0,0 +1,80 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014 matrix.org
+#
+# 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 ._base import SQLBaseStore, Table
+from synapse.api.events.room import MessageEvent
+
+import collections
+import json
+
+
+class MessageStore(SQLBaseStore):
+
+ def get_message(self, user_id, room_id, msg_id):
+ """Get a message from the store.
+
+ Args:
+ user_id (str): The ID of the user who sent the message.
+ room_id (str): The room the message was sent in.
+ msg_id (str): The unique ID for this user/room combo.
+ """
+ query = MessagesTable.select_statement(
+ "user_id = ? AND room_id = ? AND msg_id = ? " +
+ "ORDER BY id DESC LIMIT 1")
+ return self._execute(
+ MessagesTable.decode_single_result,
+ query, user_id, room_id, msg_id,
+ )
+
+ def store_message(self, user_id, room_id, msg_id, content):
+ """Store a message in the store.
+
+ Args:
+ user_id (str): The ID of the user who sent the message.
+ room_id (str): The room the message was sent in.
+ msg_id (str): The unique ID for this user/room combo.
+ content (str): The content of the message (JSON)
+ """
+ return self._simple_insert(MessagesTable.table_name, dict(
+ user_id=user_id,
+ room_id=room_id,
+ msg_id=msg_id,
+ content=content,
+ ))
+
+ def get_max_message_id(self):
+ return self._simple_max_id(MessagesTable.table_name)
+
+
+class MessagesTable(Table):
+ table_name = "messages"
+
+ fields = [
+ "id",
+ "user_id",
+ "room_id",
+ "msg_id",
+ "content"
+ ]
+
+ class EntryType(collections.namedtuple("MessageEntry", fields)):
+
+ def as_event(self, event_factory):
+ return event_factory.create_event(
+ etype=MessageEvent.TYPE,
+ room_id=self.room_id,
+ user_id=self.user_id,
+ msg_id=self.msg_id,
+ content=json.loads(self.content),
+ )
diff --git a/synapse/storage/pdu.py b/synapse/storage/pdu.py
new file mode 100644
index 0000000000..a1cdde0a3b
--- /dev/null
+++ b/synapse/storage/pdu.py
@@ -0,0 +1,993 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014 matrix.org
+#
+# 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 ._base import SQLBaseStore, Table, JoinHelper
+
+from synapse.util.logutils import log_function
+
+from collections import namedtuple
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class PduStore(SQLBaseStore):
+ """A collection of queries for handling PDUs.
+ """
+
+ def get_pdu(self, pdu_id, origin):
+ """Given a pdu_id and origin, get a PDU.
+
+ Args:
+ txn
+ pdu_id (str)
+ origin (str)
+
+ Returns:
+ PduTuple: If the pdu does not exist in the database, returns None
+ """
+
+ return self._db_pool.runInteraction(
+ self._get_pdu_tuple, pdu_id, origin
+ )
+
+ def _get_pdu_tuple(self, txn, pdu_id, origin):
+ res = self._get_pdu_tuples(txn, [(pdu_id, origin)])
+ return res[0] if res else None
+
+ def _get_pdu_tuples(self, txn, pdu_id_tuples):
+ results = []
+ for pdu_id, origin in pdu_id_tuples:
+ txn.execute(
+ PduEdgesTable.select_statement("pdu_id = ? AND origin = ?"),
+ (pdu_id, origin)
+ )
+
+ edges = [
+ (r.prev_pdu_id, r.prev_origin)
+ for r in PduEdgesTable.decode_results(txn.fetchall())
+ ]
+
+ query = (
+ "SELECT %(fields)s FROM %(pdus)s as p "
+ "LEFT JOIN %(state)s as s "
+ "ON p.pdu_id = s.pdu_id AND p.origin = s.origin "
+ "WHERE p.pdu_id = ? AND p.origin = ? "
+ ) % {
+ "fields": _pdu_state_joiner.get_fields(
+ PdusTable="p", StatePdusTable="s"),
+ "pdus": PdusTable.table_name,
+ "state": StatePdusTable.table_name,
+ }
+
+ txn.execute(query, (pdu_id, origin))
+
+ row = txn.fetchone()
+ if row:
+ results.append(PduTuple(PduEntry(*row), edges))
+
+ return results
+
+ def get_current_state_for_context(self, context):
+ """Get a list of PDUs that represent the current state for a given
+ context
+
+ Args:
+ context (str)
+
+ Returns:
+ list: A list of PduTuples
+ """
+
+ return self._db_pool.runInteraction(
+ self._get_current_state_for_context,
+ context
+ )
+
+ def _get_current_state_for_context(self, txn, context):
+ query = (
+ "SELECT pdu_id, origin FROM %s WHERE context = ?"
+ % CurrentStateTable.table_name
+ )
+
+ logger.debug("get_current_state %s, Args=%s", query, context)
+ txn.execute(query, (context,))
+
+ res = txn.fetchall()
+
+ logger.debug("get_current_state %d results", len(res))
+
+ return self._get_pdu_tuples(txn, res)
+
+ def persist_pdu(self, prev_pdus, **cols):
+ """Inserts a (non-state) PDU into the database.
+
+ Args:
+ txn,
+ prev_pdus (list)
+ **cols: The columns to insert into the PdusTable.
+ """
+ return self._db_pool.runInteraction(
+ self._persist_pdu, prev_pdus, cols
+ )
+
+ def _persist_pdu(self, txn, prev_pdus, cols):
+ entry = PdusTable.EntryType(
+ **{k: cols.get(k, None) for k in PdusTable.fields}
+ )
+
+ txn.execute(PdusTable.insert_statement(), entry)
+
+ self._handle_prev_pdus(
+ txn, entry.outlier, entry.pdu_id, entry.origin,
+ prev_pdus, entry.context
+ )
+
+ def mark_pdu_as_processed(self, pdu_id, pdu_origin):
+ """Mark a received PDU as processed.
+
+ Args:
+ txn
+ pdu_id (str)
+ pdu_origin (str)
+ """
+
+ return self._db_pool.runInteraction(
+ self._mark_as_processed, pdu_id, pdu_origin
+ )
+
+ def _mark_as_processed(self, txn, pdu_id, pdu_origin):
+ txn.execute("UPDATE %s SET have_processed = 1" % PdusTable.table_name)
+
+ def get_all_pdus_from_context(self, context):
+ """Get a list of all PDUs for a given context."""
+ return self._db_pool.runInteraction(
+ self._get_all_pdus_from_context, context,
+ )
+
+ def _get_all_pdus_from_context(self, txn, context):
+ query = (
+ "SELECT pdu_id, origin FROM %s "
+ "WHERE context = ?"
+ ) % PdusTable.table_name
+
+ txn.execute(query, (context,))
+
+ return self._get_pdu_tuples(txn, txn.fetchall())
+
+ def get_pagination(self, context, pdu_list, limit):
+ """Get a list of Pdus for a given topic that occured before (and
+ including) the pdus in pdu_list. Return a list of max size `limit`.
+
+ Args:
+ txn
+ context (str)
+ pdu_list (list)
+ limit (int)
+
+ Return:
+ list: A list of PduTuples
+ """
+ return self._db_pool.runInteraction(
+ self._get_paginate, context, pdu_list, limit
+ )
+
+ def _get_paginate(self, txn, context, pdu_list, limit):
+ logger.debug(
+ "paginate: %s, %s, %s",
+ context, repr(pdu_list), limit
+ )
+
+ # We seed the pdu_results with the things from the pdu_list.
+ pdu_results = pdu_list
+
+ front = pdu_list
+
+ query = (
+ "SELECT prev_pdu_id, prev_origin FROM %(edges_table)s "
+ "WHERE context = ? AND pdu_id = ? AND origin = ? "
+ "LIMIT ?"
+ ) % {
+ "edges_table": PduEdgesTable.table_name,
+ }
+
+ # We iterate through all pdu_ids in `front` to select their previous
+ # pdus. These are dumped in `new_front`. We continue until we reach the
+ # limit *or* new_front is empty (i.e., we've run out of things to
+ # select
+ while front and len(pdu_results) < limit:
+
+ new_front = []
+ for pdu_id, origin in front:
+ logger.debug(
+ "_paginate_interaction: i=%s, o=%s",
+ pdu_id, origin
+ )
+
+ txn.execute(
+ query,
+ (context, pdu_id, origin, limit - len(pdu_results))
+ )
+
+ for row in txn.fetchall():
+ logger.debug(
+ "_paginate_interaction: got i=%s, o=%s",
+ *row
+ )
+ new_front.append(row)
+
+ front = new_front
+ pdu_results += new_front
+
+ # We also want to update the `prev_pdus` attributes before returning.
+ return self._get_pdu_tuples(txn, pdu_results)
+
+ def get_min_depth_for_context(self, context):
+ """Get the current minimum depth for a context
+
+ Args:
+ txn
+ context (str)
+ """
+ return self._db_pool.runInteraction(
+ self._get_min_depth_for_context, context
+ )
+
+ def _get_min_depth_for_context(self, txn, context):
+ return self._get_min_depth_interaction(txn, context)
+
+ def _get_min_depth_interaction(self, txn, context):
+ txn.execute(
+ "SELECT min_depth FROM %s WHERE context = ?"
+ % ContextDepthTable.table_name,
+ (context,)
+ )
+
+ row = txn.fetchone()
+
+ return row[0] if row else None
+
+ def update_min_depth_for_context(self, context, depth):
+ """Update the minimum `depth` of the given context, which is the line
+ where we stop paginating backwards on.
+
+ Args:
+ context (str)
+ depth (int)
+ """
+ return self._db_pool.runInteraction(
+ self._update_min_depth_for_context, context, depth
+ )
+
+ def _update_min_depth_for_context(self, txn, context, depth):
+ min_depth = self._get_min_depth_interaction(txn, context)
+
+ do_insert = depth < min_depth if min_depth else True
+
+ if do_insert:
+ txn.execute(
+ "INSERT OR REPLACE INTO %s (context, min_depth) "
+ "VALUES (?,?)" % ContextDepthTable.table_name,
+ (context, depth)
+ )
+
+ def get_latest_pdus_in_context(self, context):
+ """Get's a list of the most current pdus for a given context. This is
+ used when we are sending a Pdu and need to fill out the `prev_pdus`
+ key
+
+ Args:
+ txn
+ context
+ """
+ return self._db_pool.runInteraction(
+ self._get_latest_pdus_in_context, context
+ )
+
+ def _get_latest_pdus_in_context(self, txn, context):
+ query = (
+ "SELECT p.pdu_id, p.origin, p.depth FROM %(pdus)s as p "
+ "INNER JOIN %(forward)s as f ON p.pdu_id = f.pdu_id "
+ "AND f.origin = p.origin "
+ "WHERE f.context = ?"
+ ) % {
+ "pdus": PdusTable.table_name,
+ "forward": PduForwardExtremitiesTable.table_name,
+ }
+
+ logger.debug("get_prev query: %s", query)
+
+ txn.execute(
+ query,
+ (context, )
+ )
+
+ results = txn.fetchall()
+
+ return [(row[0], row[1], row[2]) for row in results]
+
+ def get_oldest_pdus_in_context(self, context):
+ """Get a list of Pdus that we paginated beyond yet (and haven't seen).
+ This list is used when we want to paginate backwards and is the list we
+ send to the remote server.
+
+ Args:
+ txn
+ context (str)
+
+ Returns:
+ list: A list of PduIdTuple.
+ """
+ return self._db_pool.runInteraction(
+ self._get_oldest_pdus_in_context, context
+ )
+
+ def _get_oldest_pdus_in_context(self, txn, context):
+ txn.execute(
+ "SELECT pdu_id, origin FROM %(back)s WHERE context = ?"
+ % {"back": PduBackwardExtremitiesTable.table_name, },
+ (context,)
+ )
+ return [PduIdTuple(i, o) for i, o in txn.fetchall()]
+
+ def is_pdu_new(self, pdu_id, origin, context, depth):
+ """For a given Pdu, try and figure out if it's 'new', i.e., if it's
+ not something we got randomly from the past, for example when we
+ request the current state of the room that will probably return a bunch
+ of pdus from before we joined.
+
+ Args:
+ txn
+ pdu_id (str)
+ origin (str)
+ context (str)
+ depth (int)
+
+ Returns:
+ bool
+ """
+
+ return self._db_pool.runInteraction(
+ self._is_pdu_new,
+ pdu_id=pdu_id,
+ origin=origin,
+ context=context,
+ depth=depth
+ )
+
+ def _is_pdu_new(self, txn, pdu_id, origin, context, depth):
+ # If depth > min depth in back table, then we classify it as new.
+ # OR if there is nothing in the back table, then it kinda needs to
+ # be a new thing.
+ query = (
+ "SELECT min(p.depth) FROM %(edges)s as e "
+ "INNER JOIN %(back)s as b "
+ "ON e.prev_pdu_id = b.pdu_id AND e.prev_origin = b.origin "
+ "INNER JOIN %(pdus)s as p "
+ "ON e.pdu_id = p.pdu_id AND p.origin = e.origin "
+ "WHERE p.context = ?"
+ ) % {
+ "pdus": PdusTable.table_name,
+ "edges": PduEdgesTable.table_name,
+ "back": PduBackwardExtremitiesTable.table_name,
+ }
+
+ txn.execute(query, (context,))
+
+ min_depth, = txn.fetchone()
+
+ if not min_depth or depth > int(min_depth):
+ logger.debug(
+ "is_new true: id=%s, o=%s, d=%s min_depth=%s",
+ pdu_id, origin, depth, min_depth
+ )
+ return True
+
+ # If this pdu is in the forwards table, then it also is a new one
+ query = (
+ "SELECT * FROM %(forward)s WHERE pdu_id = ? AND origin = ?"
+ ) % {
+ "forward": PduForwardExtremitiesTable.table_name,
+ }
+
+ txn.execute(query, (pdu_id, origin))
+
+ # Did we get anything?
+ if txn.fetchall():
+ logger.debug(
+ "is_new true: id=%s, o=%s, d=%s was forward",
+ pdu_id, origin, depth
+ )
+ return True
+
+ logger.debug(
+ "is_new false: id=%s, o=%s, d=%s",
+ pdu_id, origin, depth
+ )
+
+ # FINE THEN. It's probably old.
+ return False
+
+ @staticmethod
+ @log_function
+ def _handle_prev_pdus(txn, outlier, pdu_id, origin, prev_pdus,
+ context):
+ txn.executemany(
+ PduEdgesTable.insert_statement(),
+ [(pdu_id, origin, p[0], p[1], context) for p in prev_pdus]
+ )
+
+ # Update the extremities table if this is not an outlier.
+ if not outlier:
+
+ # First, we delete the new one from the forwards extremities table.
+ query = (
+ "DELETE FROM %s WHERE pdu_id = ? AND origin = ?"
+ % PduForwardExtremitiesTable.table_name
+ )
+ txn.executemany(query, prev_pdus)
+
+ # We only insert as a forward extremety the new pdu if there are no
+ # other pdus that reference it as a prev pdu
+ query = (
+ "INSERT INTO %(table)s (pdu_id, origin, context) "
+ "SELECT ?, ?, ? WHERE NOT EXISTS ("
+ "SELECT 1 FROM %(pdu_edges)s WHERE "
+ "prev_pdu_id = ? AND prev_origin = ?"
+ ")"
+ ) % {
+ "table": PduForwardExtremitiesTable.table_name,
+ "pdu_edges": PduEdgesTable.table_name
+ }
+
+ logger.debug("query: %s", query)
+
+ txn.execute(query, (pdu_id, origin, context, pdu_id, origin))
+
+ # Insert all the prev_pdus as a backwards thing, they'll get
+ # deleted in a second if they're incorrect anyway.
+ txn.executemany(
+ PduBackwardExtremitiesTable.insert_statement(),
+ [(i, o, context) for i, o in prev_pdus]
+ )
+
+ # Also delete from the backwards extremities table all ones that
+ # reference pdus that we have already seen
+ query = (
+ "DELETE FROM %(pdu_back)s WHERE EXISTS ("
+ "SELECT 1 FROM %(pdus)s AS pdus "
+ "WHERE "
+ "%(pdu_back)s.pdu_id = pdus.pdu_id "
+ "AND %(pdu_back)s.origin = pdus.origin "
+ "AND not pdus.outlier "
+ ")"
+ ) % {
+ "pdu_back": PduBackwardExtremitiesTable.table_name,
+ "pdus": PdusTable.table_name,
+ }
+ txn.execute(query)
+
+
+class StatePduStore(SQLBaseStore):
+ """A collection of queries for handling state PDUs.
+ """
+
+ def persist_state(self, prev_pdus, **cols):
+ """Inserts a state PDU into the database
+
+ Args:
+ txn,
+ prev_pdus (list)
+ **cols: The columns to insert into the PdusTable and StatePdusTable
+ """
+
+ return self._db_pool.runInteraction(
+ self._persist_state, prev_pdus, cols
+ )
+
+ def _persist_state(self, txn, prev_pdus, cols):
+ pdu_entry = PdusTable.EntryType(
+ **{k: cols.get(k, None) for k in PdusTable.fields}
+ )
+ state_entry = StatePdusTable.EntryType(
+ **{k: cols.get(k, None) for k in StatePdusTable.fields}
+ )
+
+ logger.debug("Inserting pdu: %s", repr(pdu_entry))
+ logger.debug("Inserting state: %s", repr(state_entry))
+
+ txn.execute(PdusTable.insert_statement(), pdu_entry)
+ txn.execute(StatePdusTable.insert_statement(), state_entry)
+
+ self._handle_prev_pdus(
+ txn,
+ pdu_entry.outlier, pdu_entry.pdu_id, pdu_entry.origin, prev_pdus,
+ pdu_entry.context
+ )
+
+ def get_unresolved_state_tree(self, new_state_pdu):
+ return self._db_pool.runInteraction(
+ self._get_unresolved_state_tree, new_state_pdu
+ )
+
+ @log_function
+ def _get_unresolved_state_tree(self, txn, new_pdu):
+ current = self._get_current_interaction(
+ txn,
+ new_pdu.context, new_pdu.pdu_type, new_pdu.state_key
+ )
+
+ ReturnType = namedtuple(
+ "StateReturnType", ["new_branch", "current_branch"]
+ )
+ return_value = ReturnType([new_pdu], [])
+
+ if not current:
+ logger.debug("get_unresolved_state_tree No current state.")
+ return return_value
+
+ return_value.current_branch.append(current)
+
+ enum_branches = self._enumerate_state_branches(
+ txn, new_pdu, current
+ )
+
+ for branch, prev_state, state in enum_branches:
+ if state:
+ return_value[branch].append(state)
+ else:
+ break
+
+ return return_value
+
+ def update_current_state(self, pdu_id, origin, context, pdu_type,
+ state_key):
+ return self._db_pool.runInteraction(
+ self._update_current_state,
+ pdu_id, origin, context, pdu_type, state_key
+ )
+
+ def _update_current_state(self, txn, pdu_id, origin, context, pdu_type,
+ state_key):
+ query = (
+ "INSERT OR REPLACE INTO %(curr)s (%(fields)s) VALUES (%(qs)s)"
+ ) % {
+ "curr": CurrentStateTable.table_name,
+ "fields": CurrentStateTable.get_fields_string(),
+ "qs": ", ".join(["?"] * len(CurrentStateTable.fields))
+ }
+
+ query_args = CurrentStateTable.EntryType(
+ pdu_id=pdu_id,
+ origin=origin,
+ context=context,
+ pdu_type=pdu_type,
+ state_key=state_key
+ )
+
+ txn.execute(query, query_args)
+
+ def get_current_state(self, context, pdu_type, state_key):
+ """For a given context, pdu_type, state_key 3-tuple, return what is
+ currently considered the current state.
+
+ Args:
+ txn
+ context (str)
+ pdu_type (str)
+ state_key (str)
+
+ Returns:
+ PduEntry
+ """
+
+ return self._db_pool.runInteraction(
+ self._get_current_state, context, pdu_type, state_key
+ )
+
+ def _get_current_state(self, txn, context, pdu_type, state_key):
+ return self._get_current_interaction(txn, context, pdu_type, state_key)
+
+ def _get_current_interaction(self, txn, context, pdu_type, state_key):
+ logger.debug(
+ "_get_current_interaction %s %s %s",
+ context, pdu_type, state_key
+ )
+
+ fields = _pdu_state_joiner.get_fields(
+ PdusTable="p", StatePdusTable="s")
+
+ current_query = (
+ "SELECT %(fields)s FROM %(state)s as s "
+ "INNER JOIN %(pdus)s as p "
+ "ON s.pdu_id = p.pdu_id AND s.origin = p.origin "
+ "INNER JOIN %(curr)s as c "
+ "ON s.pdu_id = c.pdu_id AND s.origin = c.origin "
+ "WHERE s.context = ? AND s.pdu_type = ? AND s.state_key = ? "
+ ) % {
+ "fields": fields,
+ "curr": CurrentStateTable.table_name,
+ "state": StatePdusTable.table_name,
+ "pdus": PdusTable.table_name,
+ }
+
+ txn.execute(
+ current_query,
+ (context, pdu_type, state_key)
+ )
+
+ row = txn.fetchone()
+
+ result = PduEntry(*row) if row else None
+
+ if not result:
+ logger.debug("_get_current_interaction not found")
+ else:
+ logger.debug(
+ "_get_current_interaction found %s %s",
+ result.pdu_id, result.origin
+ )
+
+ return result
+
+ def get_next_missing_pdu(self, new_pdu):
+ """When we get a new state pdu we need to check whether we need to do
+ any conflict resolution, if we do then we need to check if we need
+ to go back and request some more state pdus that we haven't seen yet.
+
+ Args:
+ txn
+ new_pdu
+
+ Returns:
+ PduIdTuple: A pdu that we are missing, or None if we have all the
+ pdus required to do the conflict resolution.
+ """
+ return self._db_pool.runInteraction(
+ self._get_next_missing_pdu, new_pdu
+ )
+
+ def _get_next_missing_pdu(self, txn, new_pdu):
+ logger.debug(
+ "get_next_missing_pdu %s %s",
+ new_pdu.pdu_id, new_pdu.origin
+ )
+
+ current = self._get_current_interaction(
+ txn,
+ new_pdu.context, new_pdu.pdu_type, new_pdu.state_key
+ )
+
+ if (not current or not current.prev_state_id
+ or not current.prev_state_origin):
+ return None
+
+ # Oh look, it's a straight clobber, so wooooo almost no-op.
+ if (new_pdu.prev_state_id == current.pdu_id
+ and new_pdu.prev_state_origin == current.origin):
+ return None
+
+ enum_branches = self._enumerate_state_branches(txn, new_pdu, current)
+ for branch, prev_state, state in enum_branches:
+ if not state:
+ return PduIdTuple(
+ prev_state.prev_state_id,
+ prev_state.prev_state_origin
+ )
+
+ return None
+
+ def handle_new_state(self, new_pdu):
+ """Actually perform conflict resolution on the new_pdu on the
+ assumption we have all the pdus required to perform it.
+
+ Args:
+ new_pdu
+
+ Returns:
+ bool: True if the new_pdu clobbered the current state, False if not
+ """
+ return self._db_pool.runInteraction(
+ self._handle_new_state, new_pdu
+ )
+
+ def _handle_new_state(self, txn, new_pdu):
+ logger.debug(
+ "handle_new_state %s %s",
+ new_pdu.pdu_id, new_pdu.origin
+ )
+
+ current = self._get_current_interaction(
+ txn,
+ new_pdu.context, new_pdu.pdu_type, new_pdu.state_key
+ )
+
+ is_current = False
+
+ if (not current or not current.prev_state_id
+ or not current.prev_state_origin):
+ # Oh, we don't have any state for this yet.
+ is_current = True
+ elif (current.pdu_id == new_pdu.prev_state_id
+ and current.origin == new_pdu.prev_state_origin):
+ # Oh! A direct clobber. Just do it.
+ is_current = True
+ else:
+ ##
+ # Ok, now loop through until we get to a common ancestor.
+ max_new = int(new_pdu.power_level)
+ max_current = int(current.power_level)
+
+ enum_branches = self._enumerate_state_branches(
+ txn, new_pdu, current
+ )
+ for branch, prev_state, state in enum_branches:
+ if not state:
+ raise RuntimeError(
+ "Could not find state_pdu %s %s" %
+ (
+ prev_state.prev_state_id,
+ prev_state.prev_state_origin
+ )
+ )
+
+ if branch == 0:
+ max_new = max(int(state.depth), max_new)
+ else:
+ max_current = max(int(state.depth), max_current)
+
+ is_current = max_new > max_current
+
+ if is_current:
+ logger.debug("handle_new_state make current")
+
+ # Right, this is a new thing, so woo, just insert it.
+ txn.execute(
+ "INSERT OR REPLACE INTO %(curr)s (%(fields)s) VALUES (%(qs)s)"
+ % {
+ "curr": CurrentStateTable.table_name,
+ "fields": CurrentStateTable.get_fields_string(),
+ "qs": ", ".join(["?"] * len(CurrentStateTable.fields))
+ },
+ CurrentStateTable.EntryType(
+ *(new_pdu.__dict__[k] for k in CurrentStateTable.fields)
+ )
+ )
+ else:
+ logger.debug("handle_new_state not current")
+
+ logger.debug("handle_new_state done")
+
+ return is_current
+
+ @classmethod
+ @log_function
+ def _enumerate_state_branches(cls, txn, pdu_a, pdu_b):
+ branch_a = pdu_a
+ branch_b = pdu_b
+
+ get_query = (
+ "SELECT %(fields)s FROM %(pdus)s as p "
+ "LEFT JOIN %(state)s as s "
+ "ON p.pdu_id = s.pdu_id AND p.origin = s.origin "
+ "WHERE p.pdu_id = ? AND p.origin = ? "
+ ) % {
+ "fields": _pdu_state_joiner.get_fields(
+ PdusTable="p", StatePdusTable="s"),
+ "pdus": PdusTable.table_name,
+ "state": StatePdusTable.table_name,
+ }
+
+ while True:
+ if (branch_a.pdu_id == branch_b.pdu_id
+ and branch_a.origin == branch_b.origin):
+ # Woo! We found a common ancestor
+ logger.debug("_enumerate_state_branches Found common ancestor")
+ break
+
+ do_branch_a = (
+ hasattr(branch_a, "prev_state_id") and
+ branch_a.prev_state_id
+ )
+
+ do_branch_b = (
+ hasattr(branch_b, "prev_state_id") and
+ branch_b.prev_state_id
+ )
+
+ logger.debug(
+ "do_branch_a=%s, do_branch_b=%s",
+ do_branch_a, do_branch_b
+ )
+
+ if do_branch_a and do_branch_b:
+ do_branch_a = int(branch_a.depth) > int(branch_b.depth)
+
+ if do_branch_a:
+ pdu_tuple = PduIdTuple(
+ branch_a.prev_state_id,
+ branch_a.prev_state_origin
+ )
+
+ logger.debug("getting branch_a prev %s", pdu_tuple)
+ txn.execute(get_query, pdu_tuple)
+
+ prev_branch = branch_a
+
+ res = txn.fetchone()
+ branch_a = PduEntry(*res) if res else None
+
+ logger.debug("branch_a=%s", branch_a)
+
+ yield (0, prev_branch, branch_a)
+
+ if not branch_a:
+ break
+ elif do_branch_b:
+ pdu_tuple = PduIdTuple(
+ branch_b.prev_state_id,
+ branch_b.prev_state_origin
+ )
+ txn.execute(get_query, pdu_tuple)
+
+ logger.debug("getting branch_b prev %s", pdu_tuple)
+
+ prev_branch = branch_b
+
+ res = txn.fetchone()
+ branch_b = PduEntry(*res) if res else None
+
+ logger.debug("branch_b=%s", branch_b)
+
+ yield (1, prev_branch, branch_b)
+
+ if not branch_b:
+ break
+ else:
+ break
+
+
+class PdusTable(Table):
+ table_name = "pdus"
+
+ fields = [
+ "pdu_id",
+ "origin",
+ "context",
+ "pdu_type",
+ "ts",
+ "depth",
+ "is_state",
+ "content_json",
+ "unrecognized_keys",
+ "outlier",
+ "have_processed",
+ ]
+
+ EntryType = namedtuple("PdusEntry", fields)
+
+
+class PduDestinationsTable(Table):
+ table_name = "pdu_destinations"
+
+ fields = [
+ "pdu_id",
+ "origin",
+ "destination",
+ "delivered_ts",
+ ]
+
+ EntryType = namedtuple("PduDestinationsEntry", fields)
+
+
+class PduEdgesTable(Table):
+ table_name = "pdu_edges"
+
+ fields = [
+ "pdu_id",
+ "origin",
+ "prev_pdu_id",
+ "prev_origin",
+ "context"
+ ]
+
+ EntryType = namedtuple("PduEdgesEntry", fields)
+
+
+class PduForwardExtremitiesTable(Table):
+ table_name = "pdu_forward_extremities"
+
+ fields = [
+ "pdu_id",
+ "origin",
+ "context",
+ ]
+
+ EntryType = namedtuple("PduForwardExtremitiesEntry", fields)
+
+
+class PduBackwardExtremitiesTable(Table):
+ table_name = "pdu_backward_extremities"
+
+ fields = [
+ "pdu_id",
+ "origin",
+ "context",
+ ]
+
+ EntryType = namedtuple("PduBackwardExtremitiesEntry", fields)
+
+
+class ContextDepthTable(Table):
+ table_name = "context_depth"
+
+ fields = [
+ "context",
+ "min_depth",
+ ]
+
+ EntryType = namedtuple("ContextDepthEntry", fields)
+
+
+class StatePdusTable(Table):
+ table_name = "state_pdus"
+
+ fields = [
+ "pdu_id",
+ "origin",
+ "context",
+ "pdu_type",
+ "state_key",
+ "power_level",
+ "prev_state_id",
+ "prev_state_origin",
+ ]
+
+ EntryType = namedtuple("StatePdusEntry", fields)
+
+
+class CurrentStateTable(Table):
+ table_name = "current_state"
+
+ fields = [
+ "pdu_id",
+ "origin",
+ "context",
+ "pdu_type",
+ "state_key",
+ ]
+
+ EntryType = namedtuple("CurrentStateEntry", fields)
+
+_pdu_state_joiner = JoinHelper(PdusTable, StatePdusTable)
+
+
+# TODO: These should probably be put somewhere more sensible
+PduIdTuple = namedtuple("PduIdTuple", ("pdu_id", "origin"))
+
+PduEntry = _pdu_state_joiner.EntryType
+""" We are always interested in the join of the PdusTable and StatePdusTable,
+rather than just the PdusTable.
+
+This does not include a prev_pdus key.
+"""
+
+PduTuple = namedtuple(
+ "PduTuple",
+ ("pdu_entry", "prev_pdu_list")
+)
+""" This is a tuple of a `PduEntry` and a list of `PduIdTuple` that represent
+the `prev_pdus` key of a PDU.
+"""
diff --git a/synapse/storage/presence.py b/synapse/storage/presence.py
new file mode 100644
index 0000000000..e57ddaf149
--- /dev/null
+++ b/synapse/storage/presence.py
@@ -0,0 +1,103 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014 matrix.org
+#
+# 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 ._base import SQLBaseStore
+
+
+class PresenceStore(SQLBaseStore):
+ def create_presence(self, user_localpart):
+ return self._simple_insert(
+ table="presence",
+ values={"user_id": user_localpart},
+ )
+
+ def has_presence_state(self, user_localpart):
+ return self._simple_select_one(
+ table="presence",
+ keyvalues={"user_id": user_localpart},
+ retcols=["user_id"],
+ allow_none=True,
+ )
+
+ def get_presence_state(self, user_localpart):
+ return self._simple_select_one(
+ table="presence",
+ keyvalues={"user_id": user_localpart},
+ retcols=["state", "status_msg"],
+ )
+
+ def set_presence_state(self, user_localpart, new_state):
+ return self._simple_update_one(
+ table="presence",
+ keyvalues={"user_id": user_localpart},
+ updatevalues={"state": new_state["state"],
+ "status_msg": new_state["status_msg"]},
+ retcols=["state"],
+ )
+
+ def allow_presence_visible(self, observed_localpart, observer_userid):
+ return self._simple_insert(
+ table="presence_allow_inbound",
+ values={"observed_user_id": observed_localpart,
+ "observer_user_id": observer_userid},
+ )
+
+ def disallow_presence_visible(self, observed_localpart, observer_userid):
+ return self._simple_delete_one(
+ table="presence_allow_inbound",
+ keyvalues={"observed_user_id": observed_localpart,
+ "observer_user_id": observer_userid},
+ )
+
+ def is_presence_visible(self, observed_localpart, observer_userid):
+ return self._simple_select_one(
+ table="presence_allow_inbound",
+ keyvalues={"observed_user_id": observed_localpart,
+ "observer_user_id": observer_userid},
+ allow_none=True,
+ )
+
+ def add_presence_list_pending(self, observer_localpart, observed_userid):
+ return self._simple_insert(
+ table="presence_list",
+ values={"user_id": observer_localpart,
+ "observed_user_id": observed_userid,
+ "accepted": False},
+ )
+
+ def set_presence_list_accepted(self, observer_localpart, observed_userid):
+ return self._simple_update_one(
+ table="presence_list",
+ keyvalues={"user_id": observer_localpart,
+ "observed_user_id": observed_userid},
+ updatevalues={"accepted": True},
+ )
+
+ def get_presence_list(self, observer_localpart, accepted=None):
+ keyvalues = {"user_id": observer_localpart}
+ if accepted is not None:
+ keyvalues["accepted"] = accepted
+
+ return self._simple_select_list(
+ table="presence_list",
+ keyvalues=keyvalues,
+ retcols=["observed_user_id", "accepted"],
+ )
+
+ def del_presence_list(self, observer_localpart, observed_userid):
+ return self._simple_delete_one(
+ table="presence_list",
+ keyvalues={"user_id": observer_localpart,
+ "observed_user_id": observed_userid},
+ )
diff --git a/synapse/storage/profile.py b/synapse/storage/profile.py
new file mode 100644
index 0000000000..d2f24930c1
--- /dev/null
+++ b/synapse/storage/profile.py
@@ -0,0 +1,51 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014 matrix.org
+#
+# 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 ._base import SQLBaseStore
+
+
+class ProfileStore(SQLBaseStore):
+ def create_profile(self, user_localpart):
+ return self._simple_insert(
+ table="profiles",
+ values={"user_id": user_localpart},
+ )
+
+ def get_profile_displayname(self, user_localpart):
+ return self._simple_select_one_onecol(
+ table="profiles",
+ keyvalues={"user_id": user_localpart},
+ retcol="displayname",
+ )
+
+ def set_profile_displayname(self, user_localpart, new_displayname):
+ return self._simple_update_one(
+ table="profiles",
+ keyvalues={"user_id": user_localpart},
+ updatevalues={"displayname": new_displayname},
+ )
+
+ def get_profile_avatar_url(self, user_localpart):
+ return self._simple_select_one_onecol(
+ table="profiles",
+ keyvalues={"user_id": user_localpart},
+ retcol="avatar_url",
+ )
+
+ def set_profile_avatar_url(self, user_localpart, new_avatar_url):
+ return self._simple_update_one(
+ table="profiles",
+ keyvalues={"user_id": user_localpart},
+ updatevalues={"avatar_url": new_avatar_url},
+ )
diff --git a/synapse/storage/registration.py b/synapse/storage/registration.py
new file mode 100644
index 0000000000..4a970dd546
--- /dev/null
+++ b/synapse/storage/registration.py
@@ -0,0 +1,113 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014 matrix.org
+#
+# 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 twisted.internet import defer
+
+from sqlite3 import IntegrityError
+
+from synapse.api.errors import StoreError
+
+from ._base import SQLBaseStore
+
+
+class RegistrationStore(SQLBaseStore):
+
+ def __init__(self, hs):
+ super(RegistrationStore, self).__init__(hs)
+
+ self.clock = hs.get_clock()
+
+ @defer.inlineCallbacks
+ def add_access_token_to_user(self, user_id, token):
+ """Adds an access token for the given user.
+
+ Args:
+ user_id (str): The user ID.
+ token (str): The new access token to add.
+ Raises:
+ StoreError if there was a problem adding this.
+ """
+ row = yield self._simple_select_one("users", {"name": user_id}, ["id"])
+ if not row:
+ raise StoreError(400, "Bad user ID supplied.")
+ row_id = row["id"]
+ yield self._simple_insert(
+ "access_tokens",
+ {
+ "user_id": row_id,
+ "token": token
+ }
+ )
+
+ @defer.inlineCallbacks
+ def register(self, user_id, token, password_hash):
+ """Attempts to register an account.
+
+ Args:
+ user_id (str): The desired user ID to register.
+ token (str): The desired access token to use for this user.
+ password_hash (str): Optional. The password hash for this user.
+ Raises:
+ StoreError if the user_id could not be registered.
+ """
+ yield self._db_pool.runInteraction(self._register, user_id, token,
+ password_hash)
+
+ def _register(self, txn, user_id, token, password_hash):
+ now = int(self.clock.time())
+
+ try:
+ txn.execute("INSERT INTO users(name, password_hash, creation_ts) "
+ "VALUES (?,?,?)",
+ [user_id, password_hash, now])
+ except IntegrityError:
+ raise StoreError(400, "User ID already taken.")
+
+ # it's possible for this to get a conflict, but only for a single user
+ # since tokens are namespaced based on their user ID
+ txn.execute("INSERT INTO access_tokens(user_id, token) " +
+ "VALUES (?,?)", [txn.lastrowid, token])
+
+ def get_user_by_id(self, user_id):
+ query = ("SELECT users.name, users.password_hash FROM users "
+ "WHERE users.name = ?")
+ return self._execute(
+ self.cursor_to_dict,
+ query, user_id
+ )
+
+ @defer.inlineCallbacks
+ def get_user_by_token(self, token):
+ """Get a user from the given access token.
+
+ Args:
+ token (str): The access token of a user.
+ Returns:
+ str: The user ID of the user.
+ Raises:
+ StoreError if no user was found.
+ """
+ user_id = yield self._db_pool.runInteraction(self._query_for_auth,
+ token)
+ defer.returnValue(user_id)
+
+ def _query_for_auth(self, txn, token):
+ txn.execute("SELECT users.name FROM access_tokens LEFT JOIN users" +
+ " ON users.id = access_tokens.user_id WHERE token = ?",
+ [token])
+ row = txn.fetchone()
+ if row:
+ return row[0]
+
+ raise StoreError(404, "Token not found.")
diff --git a/synapse/storage/room.py b/synapse/storage/room.py
new file mode 100644
index 0000000000..174cbcf3d8
--- /dev/null
+++ b/synapse/storage/room.py
@@ -0,0 +1,129 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014 matrix.org
+#
+# 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 twisted.internet import defer
+
+from sqlite3 import IntegrityError
+
+from synapse.api.errors import StoreError
+from synapse.api.events.room import RoomTopicEvent
+
+from ._base import SQLBaseStore, Table
+
+import collections
+import json
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class RoomStore(SQLBaseStore):
+
+ @defer.inlineCallbacks
+ def store_room(self, room_id, room_creator_user_id, is_public):
+ """Stores a room.
+
+ Args:
+ room_id (str): The desired room ID, can be None.
+ room_creator_user_id (str): The user ID of the room creator.
+ is_public (bool): True to indicate that this room should appear in
+ public room lists.
+ Raises:
+ StoreError if the room could not be stored.
+ """
+ try:
+ yield self._simple_insert(RoomsTable.table_name, dict(
+ room_id=room_id,
+ creator=room_creator_user_id,
+ is_public=is_public
+ ))
+ except IntegrityError:
+ raise StoreError(409, "Room ID in use.")
+ except Exception as e:
+ logger.error("store_room with room_id=%s failed: %s", room_id, e)
+ raise StoreError(500, "Problem creating room.")
+
+ def store_room_config(self, room_id, visibility):
+ return self._simple_update_one(
+ table=RoomsTable.table_name,
+ keyvalues={"room_id": room_id},
+ updatevalues={"is_public": visibility}
+ )
+
+ def get_room(self, room_id):
+ """Retrieve a room.
+
+ Args:
+ room_id (str): The ID of the room to retrieve.
+ Returns:
+ A namedtuple containing the room information, or an empty list.
+ """
+ query = RoomsTable.select_statement("room_id=?")
+ return self._execute(
+ RoomsTable.decode_single_result, query, room_id,
+ )
+
+ @defer.inlineCallbacks
+ def get_rooms(self, is_public, with_topics):
+ """Retrieve a list of all public rooms.
+
+ Args:
+ is_public (bool): True if the rooms returned should be public.
+ with_topics (bool): True to include the current topic for the room
+ in the response.
+ Returns:
+ A list of room dicts containing at least a "room_id" key, and a
+ "topic" key if one is set and with_topic=True.
+ """
+ room_data_type = RoomTopicEvent.TYPE
+ public = 1 if is_public else 0
+
+ latest_topic = ("SELECT max(room_data.id) FROM room_data WHERE "
+ + "room_data.type = ? GROUP BY room_id")
+
+ query = ("SELECT rooms.*, room_data.content FROM rooms LEFT JOIN "
+ + "room_data ON rooms.room_id = room_data.room_id WHERE "
+ + "(room_data.id IN (" + latest_topic + ") "
+ + "OR room_data.id IS NULL) AND rooms.is_public = ?")
+
+ res = yield self._execute(
+ self.cursor_to_dict, query, room_data_type, public
+ )
+
+ # return only the keys the specification expects
+ ret_keys = ["room_id", "topic"]
+
+ # extract topic from the json (icky) FIXME
+ for i, room_row in enumerate(res):
+ try:
+ content_json = json.loads(room_row["content"])
+ room_row["topic"] = content_json["topic"]
+ except:
+ pass # no topic set
+ # filter the dict based on ret_keys
+ res[i] = {k: v for k, v in room_row.iteritems() if k in ret_keys}
+
+ defer.returnValue(res)
+
+
+class RoomsTable(Table):
+ table_name = "rooms"
+
+ fields = [
+ "room_id",
+ "is_public",
+ "creator"
+ ]
+
+ EntryType = collections.namedtuple("RoomEntry", fields)
diff --git a/synapse/storage/roomdata.py b/synapse/storage/roomdata.py
new file mode 100644
index 0000000000..781d477931
--- /dev/null
+++ b/synapse/storage/roomdata.py
@@ -0,0 +1,84 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014 matrix.org
+#
+# 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 ._base import SQLBaseStore, Table
+
+import collections
+import json
+
+
+class RoomDataStore(SQLBaseStore):
+
+ """Provides various CRUD operations for Room Events. """
+
+ def get_room_data(self, room_id, etype, state_key=""):
+ """Retrieve the data stored under this type and state_key.
+
+ Args:
+ room_id (str)
+ etype (str)
+ state_key (str)
+ Returns:
+ namedtuple: Or None if nothing exists at this path.
+ """
+ query = RoomDataTable.select_statement(
+ "room_id = ? AND type = ? AND state_key = ? "
+ "ORDER BY id DESC LIMIT 1"
+ )
+ return self._execute(
+ RoomDataTable.decode_single_result,
+ query, room_id, etype, state_key,
+ )
+
+ def store_room_data(self, room_id, etype, state_key="", content=None):
+ """Stores room specific data.
+
+ Args:
+ room_id (str)
+ etype (str)
+ state_key (str)
+ data (str)- The data to store for this path in JSON.
+ Returns:
+ The store ID for this data.
+ """
+ return self._simple_insert(RoomDataTable.table_name, dict(
+ etype=etype,
+ state_key=state_key,
+ room_id=room_id,
+ content=content,
+ ))
+
+ def get_max_room_data_id(self):
+ return self._simple_max_id(RoomDataTable.table_name)
+
+
+class RoomDataTable(Table):
+ table_name = "room_data"
+
+ fields = [
+ "id",
+ "room_id",
+ "type",
+ "state_key",
+ "content"
+ ]
+
+ class EntryType(collections.namedtuple("RoomDataEntry", fields)):
+
+ def as_event(self, event_factory):
+ return event_factory.create_event(
+ etype=self.type,
+ room_id=self.room_id,
+ content=json.loads(self.content),
+ )
diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py
new file mode 100644
index 0000000000..e6e7617797
--- /dev/null
+++ b/synapse/storage/roommember.py
@@ -0,0 +1,171 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014 matrix.org
+#
+# 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 twisted.internet import defer
+
+from synapse.types import UserID
+from synapse.api.constants import Membership
+from synapse.api.events.room import RoomMemberEvent
+
+from ._base import SQLBaseStore, Table
+
+
+import collections
+import json
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class RoomMemberStore(SQLBaseStore):
+
+ def get_room_member(self, user_id, room_id):
+ """Retrieve the current state of a room member.
+
+ Args:
+ user_id (str): The member's user ID.
+ room_id (str): The room the member is in.
+ Returns:
+ namedtuple: The room member from the database, or None if this
+ member does not exist.
+ """
+ query = RoomMemberTable.select_statement(
+ "room_id = ? AND user_id = ? ORDER BY id DESC LIMIT 1")
+ return self._execute(
+ RoomMemberTable.decode_single_result,
+ query, room_id, user_id,
+ )
+
+ def store_room_member(self, user_id, sender, room_id, membership, content):
+ """Store a room member in the database.
+
+ Args:
+ user_id (str): The member's user ID.
+ room_id (str): The room in relation to the member.
+ membership (synapse.api.constants.Membership): The new membership
+ state.
+ content (dict): The content of the membership (JSON).
+ """
+ content_json = json.dumps(content)
+ return self._simple_insert(RoomMemberTable.table_name, dict(
+ user_id=user_id,
+ sender=sender,
+ room_id=room_id,
+ membership=membership,
+ content=content_json,
+ ))
+
+ @defer.inlineCallbacks
+ def get_room_members(self, room_id, membership=None):
+ """Retrieve the current room member list for a room.
+
+ Args:
+ room_id (str): The room to get the list of members.
+ membership (synapse.api.constants.Membership): The filter to apply
+ to this list, or None to return all members with some state
+ associated with this room.
+ Returns:
+ list of namedtuples representing the members in this room.
+ """
+ query = RoomMemberTable.select_statement(
+ "id IN (SELECT MAX(id) FROM " + RoomMemberTable.table_name
+ + " WHERE room_id = ? GROUP BY user_id)"
+ )
+ res = yield self._execute(
+ RoomMemberTable.decode_results, query, room_id,
+ )
+ # strip memberships which don't match
+ if membership:
+ res = [entry for entry in res if entry.membership == membership]
+ defer.returnValue(res)
+
+ def get_rooms_for_user_where_membership_is(self, user_id, membership_list):
+ """ Get all the rooms for this user where the membership for this user
+ matches one in the membership list.
+
+ Args:
+ user_id (str): The user ID.
+ membership_list (list): A list of synapse.api.constants.Membership
+ values which the user must be in.
+ Returns:
+ A list of dicts with "room_id" and "membership" keys.
+ """
+ if not membership_list:
+ return defer.succeed(None)
+
+ args = [user_id]
+ membership_placeholder = ["membership=?"] * len(membership_list)
+ where_membership = "(" + " OR ".join(membership_placeholder) + ")"
+ for membership in membership_list:
+ args.append(membership)
+
+ query = ("SELECT room_id, membership FROM room_memberships"
+ + " WHERE user_id=? AND " + where_membership
+ + " GROUP BY room_id ORDER BY id DESC")
+ return self._execute(
+ self.cursor_to_dict, query, *args
+ )
+
+ @defer.inlineCallbacks
+ def get_joined_hosts_for_room(self, room_id):
+ query = RoomMemberTable.select_statement(
+ "id IN (SELECT MAX(id) FROM " + RoomMemberTable.table_name
+ + " WHERE room_id = ? GROUP BY user_id)"
+ )
+
+ res = yield self._execute(
+ RoomMemberTable.decode_results, query, room_id,
+ )
+
+ def host_from_user_id_string(user_id):
+ domain = UserID.from_string(entry.user_id, self.hs).domain
+ return domain
+
+ # strip memberships which don't match
+ hosts = [
+ host_from_user_id_string(entry.user_id)
+ for entry in res
+ if entry.membership == Membership.JOIN
+ ]
+
+ logger.debug("Returning hosts: %s from results: %s", hosts, res)
+
+ defer.returnValue(hosts)
+
+ def get_max_room_member_id(self):
+ return self._simple_max_id(RoomMemberTable.table_name)
+
+
+class RoomMemberTable(Table):
+ table_name = "room_memberships"
+
+ fields = [
+ "id",
+ "user_id",
+ "sender",
+ "room_id",
+ "membership",
+ "content"
+ ]
+
+ class EntryType(collections.namedtuple("RoomMemberEntry", fields)):
+
+ def as_event(self, event_factory):
+ return event_factory.create_event(
+ etype=RoomMemberEvent.TYPE,
+ room_id=self.room_id,
+ target_user_id=self.user_id,
+ user_id=self.sender,
+ content=json.loads(self.content),
+ )
diff --git a/synapse/storage/schema/edge_pdus.sql b/synapse/storage/schema/edge_pdus.sql
new file mode 100644
index 0000000000..17b3c52f0d
--- /dev/null
+++ b/synapse/storage/schema/edge_pdus.sql
@@ -0,0 +1,31 @@
+/* Copyright 2014 matrix.org
+ *
+ * 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 context_edge_pdus(
+ id INTEGER PRIMARY KEY AUTOINCREMENT, -- twistar requires this
+ pdu_id TEXT,
+ origin TEXT,
+ context TEXT,
+ CONSTRAINT context_edge_pdu_id_origin UNIQUE (pdu_id, origin)
+);
+
+CREATE TABLE IF NOT EXISTS origin_edge_pdus(
+ id INTEGER PRIMARY KEY AUTOINCREMENT, -- twistar requires this
+ pdu_id TEXT,
+ origin TEXT,
+ CONSTRAINT origin_edge_pdu_id_origin UNIQUE (pdu_id, origin)
+);
+
+CREATE INDEX IF NOT EXISTS context_edge_pdu_id ON context_edge_pdus(pdu_id, origin);
+CREATE INDEX IF NOT EXISTS origin_edge_pdu_id ON origin_edge_pdus(pdu_id, origin);
diff --git a/synapse/storage/schema/im.sql b/synapse/storage/schema/im.sql
new file mode 100644
index 0000000000..77096546b2
--- /dev/null
+++ b/synapse/storage/schema/im.sql
@@ -0,0 +1,54 @@
+/* Copyright 2014 matrix.org
+ *
+ * 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 rooms(
+ room_id TEXT PRIMARY KEY NOT NULL,
+ is_public INTEGER,
+ creator TEXT
+);
+
+CREATE TABLE IF NOT EXISTS room_memberships(
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ user_id TEXT NOT NULL, -- no foreign key to users table, it could be an id belonging to another home server
+ sender TEXT NOT NULL,
+ room_id TEXT NOT NULL,
+ membership TEXT NOT NULL,
+ content TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS messages(
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ user_id TEXT,
+ room_id TEXT,
+ msg_id TEXT,
+ content TEXT
+);
+
+CREATE TABLE IF NOT EXISTS feedback(
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ content TEXT,
+ feedback_type TEXT,
+ fb_sender_id TEXT,
+ msg_id TEXT,
+ room_id TEXT,
+ msg_sender_id TEXT
+);
+
+CREATE TABLE IF NOT EXISTS room_data(
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ room_id TEXT NOT NULL,
+ type TEXT NOT NULL,
+ state_key TEXT NOT NULL,
+ content TEXT
+);
diff --git a/synapse/storage/schema/pdu.sql b/synapse/storage/schema/pdu.sql
new file mode 100644
index 0000000000..ca3de005e9
--- /dev/null
+++ b/synapse/storage/schema/pdu.sql
@@ -0,0 +1,106 @@
+/* Copyright 2014 matrix.org
+ *
+ * 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.
+ */
+-- Stores pdus and their content
+CREATE TABLE IF NOT EXISTS pdus(
+ pdu_id TEXT,
+ origin TEXT,
+ context TEXT,
+ pdu_type TEXT,
+ ts INTEGER,
+ depth INTEGER DEFAULT 0 NOT NULL,
+ is_state BOOL,
+ content_json TEXT,
+ unrecognized_keys TEXT,
+ outlier BOOL NOT NULL,
+ have_processed BOOL,
+ CONSTRAINT pdu_id_origin UNIQUE (pdu_id, origin)
+);
+
+-- Stores what the current state pdu is for a given (context, pdu_type, key) tuple
+CREATE TABLE IF NOT EXISTS state_pdus(
+ pdu_id TEXT,
+ origin TEXT,
+ context TEXT,
+ pdu_type TEXT,
+ state_key TEXT,
+ power_level TEXT,
+ prev_state_id TEXT,
+ prev_state_origin TEXT,
+ CONSTRAINT pdu_id_origin UNIQUE (pdu_id, origin)
+ CONSTRAINT prev_pdu_id_origin UNIQUE (prev_state_id, prev_state_origin)
+);
+
+CREATE TABLE IF NOT EXISTS current_state(
+ pdu_id TEXT,
+ origin TEXT,
+ context TEXT,
+ pdu_type TEXT,
+ state_key TEXT,
+ CONSTRAINT pdu_id_origin UNIQUE (pdu_id, origin)
+ CONSTRAINT uniqueness UNIQUE (context, pdu_type, state_key) ON CONFLICT REPLACE
+);
+
+-- Stores where each pdu we want to send should be sent and the delivery status.
+create TABLE IF NOT EXISTS pdu_destinations(
+ pdu_id TEXT,
+ origin TEXT,
+ destination TEXT,
+ delivered_ts INTEGER DEFAULT 0, -- or 0 if not delivered
+ CONSTRAINT uniqueness UNIQUE (pdu_id, origin, destination) ON CONFLICT REPLACE
+);
+
+CREATE TABLE IF NOT EXISTS pdu_forward_extremities(
+ pdu_id TEXT,
+ origin TEXT,
+ context TEXT,
+ CONSTRAINT uniqueness UNIQUE (pdu_id, origin, context) ON CONFLICT REPLACE
+);
+
+CREATE TABLE IF NOT EXISTS pdu_backward_extremities(
+ pdu_id TEXT,
+ origin TEXT,
+ context TEXT,
+ CONSTRAINT uniqueness UNIQUE (pdu_id, origin, context) ON CONFLICT REPLACE
+);
+
+CREATE TABLE IF NOT EXISTS pdu_edges(
+ pdu_id TEXT,
+ origin TEXT,
+ prev_pdu_id TEXT,
+ prev_origin TEXT,
+ context TEXT,
+ CONSTRAINT uniqueness UNIQUE (pdu_id, origin, prev_pdu_id, prev_origin, context)
+);
+
+CREATE TABLE IF NOT EXISTS context_depth(
+ context TEXT,
+ min_depth INTEGER,
+ CONSTRAINT uniqueness UNIQUE (context)
+);
+
+CREATE INDEX IF NOT EXISTS context_depth_context ON context_depth(context);
+
+
+CREATE INDEX IF NOT EXISTS pdu_id ON pdus(pdu_id, origin);
+
+CREATE INDEX IF NOT EXISTS dests_id ON pdu_destinations (pdu_id, origin);
+-- CREATE INDEX IF NOT EXISTS dests ON pdu_destinations (destination);
+
+CREATE INDEX IF NOT EXISTS pdu_extrem_context ON pdu_forward_extremities(context);
+CREATE INDEX IF NOT EXISTS pdu_extrem_id ON pdu_forward_extremities(pdu_id, origin);
+
+CREATE INDEX IF NOT EXISTS pdu_edges_id ON pdu_edges(pdu_id, origin);
+
+CREATE INDEX IF NOT EXISTS pdu_b_extrem_context ON pdu_backward_extremities(context);
diff --git a/synapse/storage/schema/presence.sql b/synapse/storage/schema/presence.sql
new file mode 100644
index 0000000000..b22e3ba863
--- /dev/null
+++ b/synapse/storage/schema/presence.sql
@@ -0,0 +1,37 @@
+/* Copyright 2014 matrix.org
+ *
+ * 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 presence(
+ user_id INTEGER NOT NULL,
+ state INTEGER,
+ status_msg TEXT,
+ FOREIGN KEY(user_id) REFERENCES users(id)
+);
+
+-- For each of /my/ users which possibly-remote users are allowed to see their
+-- presence state
+CREATE TABLE IF NOT EXISTS presence_allow_inbound(
+ observed_user_id INTEGER NOT NULL,
+ observer_user_id TEXT, -- a UserID,
+ FOREIGN KEY(observed_user_id) REFERENCES users(id)
+);
+
+-- For each of /my/ users (watcher), which possibly-remote users are they
+-- watching?
+CREATE TABLE IF NOT EXISTS presence_list(
+ user_id INTEGER NOT NULL,
+ observed_user_id TEXT, -- a UserID,
+ accepted BOOLEAN,
+ FOREIGN KEY(user_id) REFERENCES users(id)
+);
diff --git a/synapse/storage/schema/profiles.sql b/synapse/storage/schema/profiles.sql
new file mode 100644
index 0000000000..1092d7672c
--- /dev/null
+++ b/synapse/storage/schema/profiles.sql
@@ -0,0 +1,20 @@
+/* Copyright 2014 matrix.org
+ *
+ * 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 profiles(
+ user_id INTEGER NOT NULL,
+ displayname TEXT,
+ avatar_url TEXT,
+ FOREIGN KEY(user_id) REFERENCES users(id)
+);
diff --git a/synapse/storage/schema/room_aliases.sql b/synapse/storage/schema/room_aliases.sql
new file mode 100644
index 0000000000..71a8b90e4d
--- /dev/null
+++ b/synapse/storage/schema/room_aliases.sql
@@ -0,0 +1,12 @@
+CREATE TABLE IF NOT EXISTS room_aliases(
+ room_alias TEXT NOT NULL,
+ room_id TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS room_alias_servers(
+ room_alias TEXT NOT NULL,
+ server TEXT NOT NULL
+);
+
+
+
diff --git a/synapse/storage/schema/transactions.sql b/synapse/storage/schema/transactions.sql
new file mode 100644
index 0000000000..4b1a2368f6
--- /dev/null
+++ b/synapse/storage/schema/transactions.sql
@@ -0,0 +1,61 @@
+/* Copyright 2014 matrix.org
+ *
+ * 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.
+ */
+-- Stores what transaction ids we have received and what our response was
+CREATE TABLE IF NOT EXISTS received_transactions(
+ transaction_id TEXT,
+ origin TEXT,
+ ts INTEGER,
+ response_code INTEGER,
+ response_json TEXT,
+ has_been_referenced BOOL default 0, -- Whether thishas been referenced by a prev_tx
+ CONSTRAINT uniquesss UNIQUE (transaction_id, origin) ON CONFLICT REPLACE
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS transactions_txid ON received_transactions(transaction_id, origin);
+CREATE INDEX IF NOT EXISTS transactions_have_ref ON received_transactions(origin, has_been_referenced);-- WHERE has_been_referenced = 0;
+
+
+-- Stores what transactions we've sent, what their response was (if we got one) and whether we have
+-- since referenced the transaction in another outgoing transaction
+CREATE TABLE IF NOT EXISTS sent_transactions(
+ id INTEGER PRIMARY KEY AUTOINCREMENT, -- This is used to apply insertion ordering
+ transaction_id TEXT,
+ destination TEXT,
+ response_code INTEGER DEFAULT 0,
+ response_json TEXT,
+ ts INTEGER
+);
+
+CREATE INDEX IF NOT EXISTS sent_transaction_dest ON sent_transactions(destination);
+CREATE INDEX IF NOT EXISTS sent_transaction_dest_referenced ON sent_transactions(
+ destination
+);
+-- So that we can do an efficient look up of all transactions that have yet to be successfully
+-- sent.
+CREATE INDEX IF NOT EXISTS sent_transaction_sent ON sent_transactions(response_code);
+
+
+-- For sent transactions only.
+CREATE TABLE IF NOT EXISTS transaction_id_to_pdu(
+ transaction_id INTEGER,
+ destination TEXT,
+ pdu_id TEXT,
+ pdu_origin TEXT
+);
+
+CREATE INDEX IF NOT EXISTS transaction_id_to_pdu_tx ON transaction_id_to_pdu(transaction_id, destination);
+CREATE INDEX IF NOT EXISTS transaction_id_to_pdu_dest ON transaction_id_to_pdu(destination);
+CREATE INDEX IF NOT EXISTS transaction_id_to_pdu_index ON transaction_id_to_pdu(transaction_id, destination);
+
diff --git a/synapse/storage/schema/users.sql b/synapse/storage/schema/users.sql
new file mode 100644
index 0000000000..46b60297cb
--- /dev/null
+++ b/synapse/storage/schema/users.sql
@@ -0,0 +1,31 @@
+/* Copyright 2014 matrix.org
+ *
+ * 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 users(
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ name TEXT,
+ password_hash TEXT,
+ creation_ts INTEGER,
+ UNIQUE(name) ON CONFLICT ROLLBACK
+);
+
+CREATE TABLE IF NOT EXISTS access_tokens(
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ user_id INTEGER NOT NULL,
+ device_id TEXT,
+ token TEXT NOT NULL,
+ last_used INTEGER,
+ FOREIGN KEY(user_id) REFERENCES users(id),
+ UNIQUE(token) ON CONFLICT ROLLBACK
+);
diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py
new file mode 100644
index 0000000000..c3b1bfeb32
--- /dev/null
+++ b/synapse/storage/stream.py
@@ -0,0 +1,282 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014 matrix.org
+#
+# 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 ._base import SQLBaseStore
+from .message import MessagesTable
+from .feedback import FeedbackTable
+from .roomdata import RoomDataTable
+from .roommember import RoomMemberTable
+
+import json
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class StreamStore(SQLBaseStore):
+
+ def get_message_stream(self, user_id, from_key, to_key, room_id, limit=0,
+ with_feedback=False):
+ """Get all messages for this user between the given keys.
+
+ Args:
+ user_id (str): The user who is requesting messages.
+ from_key (int): The ID to start returning results from (exclusive).
+ to_key (int): The ID to stop returning results (exclusive).
+ room_id (str): Gets messages only for this room. Can be None, in
+ which case all room messages will be returned.
+ Returns:
+ A tuple of rows (list of namedtuples), new_id(int)
+ """
+ if with_feedback and room_id: # with fb MUST specify a room ID
+ return self._db_pool.runInteraction(
+ self._get_message_rows_with_feedback,
+ user_id, from_key, to_key, room_id, limit
+ )
+ else:
+ return self._db_pool.runInteraction(
+ self._get_message_rows,
+ user_id, from_key, to_key, room_id, limit
+ )
+
+ def _get_message_rows(self, txn, user_id, from_pkey, to_pkey, room_id,
+ limit):
+ # work out which rooms this user is joined in on and join them with
+ # the room id on the messages table, bounded by the specified pkeys
+
+ # get all messages where the *current* membership state is 'join' for
+ # this user in that room.
+ query = ("SELECT messages.* FROM messages WHERE ? IN"
+ + " (SELECT membership from room_memberships WHERE user_id=?"
+ + " AND room_id = messages.room_id ORDER BY id DESC LIMIT 1)")
+ query_args = ["join", user_id]
+
+ if room_id:
+ query += " AND messages.room_id=?"
+ query_args.append(room_id)
+
+ (query, query_args) = self._append_stream_operations(
+ "messages", query, query_args, from_pkey, to_pkey, limit=limit
+ )
+
+ logger.debug("[SQL] %s : %s", query, query_args)
+ cursor = txn.execute(query, query_args)
+ return self._as_events(cursor, MessagesTable, from_pkey)
+
+ def _get_message_rows_with_feedback(self, txn, user_id, from_pkey, to_pkey,
+ room_id, limit):
+ # this col represents the compressed feedback JSON as per spec
+ compressed_feedback_col = (
+ "'[' || group_concat('{\"sender_id\":\"' || f.fb_sender_id"
+ + " || '\",\"feedback_type\":\"' || f.feedback_type"
+ + " || '\",\"content\":' || f.content || '}') || ']'"
+ )
+
+ global_msg_id_join = ("f.room_id = messages.room_id"
+ + " and f.msg_id = messages.msg_id"
+ + " and messages.user_id = f.msg_sender_id")
+
+ select_query = (
+ "SELECT messages.*, f.content AS fb_content, f.fb_sender_id"
+ + ", " + compressed_feedback_col + " AS compressed_fb"
+ + " FROM messages LEFT JOIN feedback f ON " + global_msg_id_join)
+
+ current_membership_sub_query = (
+ "(SELECT membership from room_memberships rm"
+ + " WHERE user_id=? AND room_id = rm.room_id"
+ + " ORDER BY id DESC LIMIT 1)")
+
+ where = (" WHERE ? IN " + current_membership_sub_query
+ + " AND messages.room_id=?")
+
+ query = select_query + where
+ query_args = ["join", user_id, room_id]
+
+ (query, query_args) = self._append_stream_operations(
+ "messages", query, query_args, from_pkey, to_pkey,
+ limit=limit, group_by=" GROUP BY messages.id "
+ )
+
+ logger.debug("[SQL] %s : %s", query, query_args)
+ cursor = txn.execute(query, query_args)
+
+ # convert the result set into events
+ entries = self.cursor_to_dict(cursor)
+ events = []
+ for entry in entries:
+ # TODO we should spec the cursor > event mapping somewhere else.
+ event = {}
+ straight_mappings = ["msg_id", "user_id", "room_id"]
+ for key in straight_mappings:
+ event[key] = entry[key]
+ event["content"] = json.loads(entry["content"])
+ if entry["compressed_fb"]:
+ event["feedback"] = json.loads(entry["compressed_fb"])
+ events.append(event)
+
+ latest_pkey = from_pkey if len(entries) == 0 else entries[-1]["id"]
+
+ return (events, latest_pkey)
+
+ def get_room_member_stream(self, user_id, from_key, to_key):
+ """Get all room membership events for this user between the given keys.
+
+ Args:
+ user_id (str): The user who is requesting membership events.
+ from_key (int): The ID to start returning results from (exclusive).
+ to_key (int): The ID to stop returning results (exclusive).
+ Returns:
+ A tuple of rows (list of namedtuples), new_id(int)
+ """
+ return self._db_pool.runInteraction(
+ self._get_room_member_rows, user_id, from_key, to_key
+ )
+
+ def _get_room_member_rows(self, txn, user_id, from_pkey, to_pkey):
+ # get all room membership events for rooms which the user is
+ # *currently* joined in on, or all invite events for this user.
+ current_membership_sub_query = (
+ "(SELECT membership FROM room_memberships"
+ + " WHERE user_id=? AND room_id = rm.room_id"
+ + " ORDER BY id DESC LIMIT 1)")
+
+ query = ("SELECT rm.* FROM room_memberships rm "
+ # all membership events for rooms you've currently joined.
+ + " WHERE (? IN " + current_membership_sub_query
+ # all invite membership events for this user
+ + " OR rm.membership=? AND user_id=?)"
+ + " AND rm.id > ?")
+ query_args = ["join", user_id, "invite", user_id, from_pkey]
+
+ if to_pkey != -1:
+ query += " AND rm.id < ?"
+ query_args.append(to_pkey)
+
+ cursor = txn.execute(query, query_args)
+ return self._as_events(cursor, RoomMemberTable, from_pkey)
+
+ def get_feedback_stream(self, user_id, from_key, to_key, room_id, limit=0):
+ return self._db_pool.runInteraction(
+ self._get_feedback_rows,
+ user_id, from_key, to_key, room_id, limit
+ )
+
+ def _get_feedback_rows(self, txn, user_id, from_pkey, to_pkey, room_id,
+ limit):
+ # work out which rooms this user is joined in on and join them with
+ # the room id on the feedback table, bounded by the specified pkeys
+
+ # get all messages where the *current* membership state is 'join' for
+ # this user in that room.
+ query = (
+ "SELECT feedback.* FROM feedback WHERE ? IN "
+ + "(SELECT membership from room_memberships WHERE user_id=?"
+ + " AND room_id = feedback.room_id ORDER BY id DESC LIMIT 1)")
+ query_args = ["join", user_id]
+
+ if room_id:
+ query += " AND feedback.room_id=?"
+ query_args.append(room_id)
+
+ (query, query_args) = self._append_stream_operations(
+ "feedback", query, query_args, from_pkey, to_pkey, limit=limit
+ )
+
+ logger.debug("[SQL] %s : %s", query, query_args)
+ cursor = txn.execute(query, query_args)
+ return self._as_events(cursor, FeedbackTable, from_pkey)
+
+ def get_room_data_stream(self, user_id, from_key, to_key, room_id,
+ limit=0):
+ return self._db_pool.runInteraction(
+ self._get_room_data_rows,
+ user_id, from_key, to_key, room_id, limit
+ )
+
+ def _get_room_data_rows(self, txn, user_id, from_pkey, to_pkey, room_id,
+ limit):
+ # work out which rooms this user is joined in on and join them with
+ # the room id on the feedback table, bounded by the specified pkeys
+
+ # get all messages where the *current* membership state is 'join' for
+ # this user in that room.
+ query = (
+ "SELECT room_data.* FROM room_data WHERE ? IN "
+ + "(SELECT membership from room_memberships WHERE user_id=?"
+ + " AND room_id = room_data.room_id ORDER BY id DESC LIMIT 1)")
+ query_args = ["join", user_id]
+
+ if room_id:
+ query += " AND room_data.room_id=?"
+ query_args.append(room_id)
+
+ (query, query_args) = self._append_stream_operations(
+ "room_data", query, query_args, from_pkey, to_pkey, limit=limit
+ )
+
+ logger.debug("[SQL] %s : %s", query, query_args)
+ cursor = txn.execute(query, query_args)
+ return self._as_events(cursor, RoomDataTable, from_pkey)
+
+ def _append_stream_operations(self, table_name, query, query_args,
+ from_pkey, to_pkey, limit=None,
+ group_by=""):
+ LATEST_ROW = -1
+ order_by = ""
+ if to_pkey > from_pkey:
+ if from_pkey != LATEST_ROW:
+ # e.g. from=5 to=9 >> from 5 to 9 >> id>5 AND id<9
+ query += (" AND %s.id > ? AND %s.id < ?" %
+ (table_name, table_name))
+ query_args.append(from_pkey)
+ query_args.append(to_pkey)
+ else:
+ # e.g. from=-1 to=5 >> from now to 5 >> id>5 ORDER BY id DESC
+ query += " AND %s.id > ? " % table_name
+ order_by = "ORDER BY id DESC"
+ query_args.append(to_pkey)
+ elif from_pkey > to_pkey:
+ if to_pkey != LATEST_ROW:
+ # from=9 to=5 >> from 9 to 5 >> id>5 AND id<9 ORDER BY id DESC
+ query += (" AND %s.id > ? AND %s.id < ? " %
+ (table_name, table_name))
+ order_by = "ORDER BY id DESC"
+ query_args.append(to_pkey)
+ query_args.append(from_pkey)
+ else:
+ # from=5 to=-1 >> from 5 to now >> id>5
+ query += " AND %s.id > ?" % table_name
+ query_args.append(from_pkey)
+
+ query += group_by + order_by
+
+ if limit and limit > 0:
+ query += " LIMIT ?"
+ query_args.append(str(limit))
+
+ return (query, query_args)
+
+ def _as_events(self, cursor, table, from_pkey):
+ data_entries = table.decode_results(cursor)
+ last_pkey = from_pkey
+ if data_entries:
+ last_pkey = data_entries[-1].id
+
+ events = [
+ entry.as_event(self.event_factory).get_dict()
+ for entry in data_entries
+ ]
+
+ return (events, last_pkey)
diff --git a/synapse/storage/transactions.py b/synapse/storage/transactions.py
new file mode 100644
index 0000000000..aa41e2ad7f
--- /dev/null
+++ b/synapse/storage/transactions.py
@@ -0,0 +1,287 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014 matrix.org
+#
+# 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 ._base import SQLBaseStore, Table
+from .pdu import PdusTable
+
+from collections import namedtuple
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class TransactionStore(SQLBaseStore):
+ """A collection of queries for handling PDUs.
+ """
+
+ def get_received_txn_response(self, transaction_id, origin):
+ """For an incoming transaction from a given origin, check if we have
+ already responded to it. If so, return the response code and response
+ body (as a dict).
+
+ Args:
+ transaction_id (str)
+ origin(str)
+
+ Returns:
+ tuple: None if we have not previously responded to
+ this transaction or a 2-tuple of (int, dict)
+ """
+
+ return self._db_pool.runInteraction(
+ self._get_received_txn_response, transaction_id, origin
+ )
+
+ def _get_received_txn_response(self, txn, transaction_id, origin):
+ where_clause = "transaction_id = ? AND origin = ?"
+ query = ReceivedTransactionsTable.select_statement(where_clause)
+
+ txn.execute(query, (transaction_id, origin))
+
+ results = ReceivedTransactionsTable.decode_results(txn.fetchall())
+
+ if results and results[0].response_code:
+ return (results[0].response_code, results[0].response_json)
+ else:
+ return None
+
+ def set_received_txn_response(self, transaction_id, origin, code,
+ response_dict):
+ """Persist the response we returened for an incoming transaction, and
+ should return for subsequent transactions with the same transaction_id
+ and origin.
+
+ Args:
+ txn
+ transaction_id (str)
+ origin (str)
+ code (int)
+ response_json (str)
+ """
+
+ return self._db_pool.runInteraction(
+ self._set_received_txn_response,
+ transaction_id, origin, code, response_dict
+ )
+
+ def _set_received_txn_response(self, txn, transaction_id, origin, code,
+ response_json):
+ query = (
+ "UPDATE %s "
+ "SET response_code = ?, response_json = ? "
+ "WHERE transaction_id = ? AND origin = ?"
+ ) % ReceivedTransactionsTable.table_name
+
+ txn.execute(query, (code, response_json, transaction_id, origin))
+
+ def prep_send_transaction(self, transaction_id, destination, ts, pdu_list):
+ """Persists an outgoing transaction and calculates the values for the
+ previous transaction id list.
+
+ This should be called before sending the transaction so that it has the
+ correct value for the `prev_ids` key.
+
+ Args:
+ transaction_id (str)
+ destination (str)
+ ts (int)
+ pdu_list (list)
+
+ Returns:
+ list: A list of previous transaction ids.
+ """
+
+ return self._db_pool.runInteraction(
+ self._prep_send_transaction,
+ transaction_id, destination, ts, pdu_list
+ )
+
+ def _prep_send_transaction(self, txn, transaction_id, destination, ts,
+ pdu_list):
+
+ # First we find out what the prev_txs should be.
+ # Since we know that we are only sending one transaction at a time,
+ # we can simply take the last one.
+ query = "%s ORDER BY id DESC LIMIT 1" % (
+ SentTransactions.select_statement("destination = ?"),
+ )
+
+ results = txn.execute(query, (destination,))
+ results = SentTransactions.decode_results(results)
+
+ prev_txns = [r.transaction_id for r in results]
+
+ # Actually add the new transaction to the sent_transactions table.
+
+ query = SentTransactions.insert_statement()
+ txn.execute(query, SentTransactions.EntryType(
+ None,
+ transaction_id=transaction_id,
+ destination=destination,
+ ts=ts,
+ response_code=0,
+ response_json=None
+ ))
+
+ # Update the tx id -> pdu id mapping
+
+ values = [
+ (transaction_id, destination, pdu[0], pdu[1])
+ for pdu in pdu_list
+ ]
+
+ logger.debug("Inserting: %s", repr(values))
+
+ query = TransactionsToPduTable.insert_statement()
+ txn.executemany(query, values)
+
+ return prev_txns
+
+ def delivered_txn(self, transaction_id, destination, code, response_dict):
+ """Persists the response for an outgoing transaction.
+
+ Args:
+ transaction_id (str)
+ destination (str)
+ code (int)
+ response_json (str)
+ """
+ return self._db_pool.runInteraction(
+ self._delivered_txn,
+ transaction_id, destination, code, response_dict
+ )
+
+ def _delivered_txn(cls, txn, transaction_id, destination,
+ code, response_json):
+ query = (
+ "UPDATE %s "
+ "SET response_code = ?, response_json = ? "
+ "WHERE transaction_id = ? AND destination = ?"
+ ) % SentTransactions.table_name
+
+ txn.execute(query, (code, response_json, transaction_id, destination))
+
+ def get_transactions_after(self, transaction_id, destination):
+ """Get all transactions after a given local transaction_id.
+
+ Args:
+ transaction_id (str)
+ destination (str)
+
+ Returns:
+ list: A list of `ReceivedTransactionsTable.EntryType`
+ """
+ return self._db_pool.runInteraction(
+ self._get_transactions_after, transaction_id, destination
+ )
+
+ def _get_transactions_after(cls, txn, transaction_id, destination):
+ where = (
+ "destination = ? AND id > (select id FROM %s WHERE "
+ "transaction_id = ? AND destination = ?)"
+ ) % (
+ SentTransactions.table_name
+ )
+ query = SentTransactions.select_statement(where)
+
+ txn.execute(query, (destination, transaction_id, destination))
+
+ return ReceivedTransactionsTable.decode_results(txn.fetchall())
+
+ def get_pdus_after_transaction(self, transaction_id, destination):
+ """For a given local transaction_id that we sent to a given destination
+ home server, return a list of PDUs that were sent to that destination
+ after it.
+
+ Args:
+ txn
+ transaction_id (str)
+ destination (str)
+
+ Returns
+ list: A list of PduTuple
+ """
+ return self._db_pool.runInteraction(
+ self._get_pdus_after_transaction,
+ transaction_id, destination
+ )
+
+ def _get_pdus_after_transaction(self, txn, transaction_id, destination):
+
+ # Query that first get's all transaction_ids with an id greater than
+ # the one given from the `sent_transactions` table. Then JOIN on this
+ # from the `tx->pdu` table to get a list of (pdu_id, origin) that
+ # specify the pdus that were sent in those transactions.
+ query = (
+ "SELECT pdu_id, pdu_origin FROM %(tx_pdu)s as tp "
+ "INNER JOIN %(sent_tx)s as st "
+ "ON tp.transaction_id = st.transaction_id "
+ "AND tp.destination = st.destination "
+ "WHERE st.id > ("
+ "SELECT id FROM %(sent_tx)s "
+ "WHERE transaction_id = ? AND destination = ?"
+ ) % {
+ "tx_pdu": TransactionsToPduTable.table_name,
+ "sent_tx": SentTransactions.table_name,
+ }
+
+ txn.execute(query, (transaction_id, destination))
+
+ pdus = PdusTable.decode_results(txn.fetchall())
+
+ return self._get_pdu_tuples(txn, pdus)
+
+
+class ReceivedTransactionsTable(Table):
+ table_name = "received_transactions"
+
+ fields = [
+ "transaction_id",
+ "origin",
+ "ts",
+ "response_code",
+ "response_json",
+ "has_been_referenced",
+ ]
+
+ EntryType = namedtuple("ReceivedTransactionsEntry", fields)
+
+
+class SentTransactions(Table):
+ table_name = "sent_transactions"
+
+ fields = [
+ "id",
+ "transaction_id",
+ "destination",
+ "ts",
+ "response_code",
+ "response_json",
+ ]
+
+ EntryType = namedtuple("SentTransactionsEntry", fields)
+
+
+class TransactionsToPduTable(Table):
+ table_name = "transaction_id_to_pdu"
+
+ fields = [
+ "transaction_id",
+ "destination",
+ "pdu_id",
+ "pdu_origin",
+ ]
+
+ EntryType = namedtuple("TransactionsToPduEntry", fields)
|
