diff options
Diffstat (limited to 'synapse/storage')
| -rw-r--r-- | synapse/storage/__init__.py | 2 | ||||
| -rw-r--r-- | synapse/storage/event_push_actions.py | 98 | ||||
| -rw-r--r-- | synapse/storage/push_rule.py | 41 | ||||
| -rw-r--r-- | synapse/storage/schema/delta/27/event_push_actions.sql | 26 |
4 files changed, 167 insertions, 0 deletions
diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index c46b653f11..43e05f144a 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -33,6 +33,7 @@ from .pusher import PusherStore from .push_rule import PushRuleStore from .media_repository import MediaRepositoryStore from .rejections import RejectionsStore +from .event_push_actions import EventPushActionsStore from .state import StateStore from .signatures import SignatureStore @@ -75,6 +76,7 @@ class DataStore(RoomMemberStore, RoomStore, SearchStore, TagsStore, AccountDataStore, + EventPushActionsStore ): def __init__(self, hs): diff --git a/synapse/storage/event_push_actions.py b/synapse/storage/event_push_actions.py new file mode 100644 index 0000000000..3075d02257 --- /dev/null +++ b/synapse/storage/event_push_actions.py @@ -0,0 +1,98 @@ +# -*- coding: utf-8 -*- +# Copyright 2015 OpenMarket Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from ._base import SQLBaseStore +from twisted.internet import defer + +import logging +import simplejson as json + +logger = logging.getLogger(__name__) + + +class EventPushActionsStore(SQLBaseStore): + @defer.inlineCallbacks + def set_push_actions_for_event_and_users(self, event, tuples): + """ + :param event: the event set actions for + :param tuples: list of tuples of (user_id, profile_tag, actions) + """ + values = [] + for uid, profile_tag, actions in tuples: + values.append({ + 'room_id': event['room_id'], + 'event_id': event['event_id'], + 'user_id': uid, + 'profile_tag': profile_tag, + 'actions': json.dumps(actions) + }) + + yield self.runInteraction( + "set_actions_for_event_and_users", + self._simple_insert_many_txn, + EventPushActionsTable.table_name, + values + ) + + @defer.inlineCallbacks + def get_unread_event_push_actions_by_room_for_user( + self, room_id, user_id, last_read_event_id + ): + def _get_unread_event_push_actions_by_room(txn): + sql = ( + "SELECT stream_ordering, topological_ordering" + " FROM events" + " WHERE room_id = ? AND event_id = ?" + ) + txn.execute( + sql, (room_id, last_read_event_id) + ) + results = txn.fetchall() + if len(results) == 0: + return [] + + stream_ordering = results[0][0] + topological_ordering = results[0][1] + + sql = ( + "SELECT ea.event_id, ea.actions" + " FROM event_push_actions ea, events e" + " WHERE ea.room_id = e.room_id" + " AND ea.event_id = e.event_id" + " AND ea.user_id = ?" + " AND ea.room_id = ?" + " AND (" + " e.topological_ordering > ?" + " OR (e.topological_ordering = ? AND e.stream_ordering > ?)" + ")" + ) + txn.execute(sql, ( + user_id, room_id, + topological_ordering, topological_ordering, stream_ordering + ) + ) + return [ + {"event_id": row[0], "actions": row[1]} for row in txn.fetchall() + ] + + ret = yield self.runInteraction( + "get_unread_event_push_actions_by_room", + _get_unread_event_push_actions_by_room + ) + defer.returnValue(ret) + + +class EventPushActionsTable(object): + table_name = "event_push_actions" diff --git a/synapse/storage/push_rule.py b/synapse/storage/push_rule.py index 5305b7e122..7c5123d644 100644 --- a/synapse/storage/push_rule.py +++ b/synapse/storage/push_rule.py @@ -56,6 +56,47 @@ class PushRuleStore(SQLBaseStore): }) @defer.inlineCallbacks + def bulk_get_push_rules(self, user_ids): + batch_size = 100 + + def f(txn, user_ids_to_fetch): + sql = ( + "SELECT " + + ",".join("pr."+x for x in PushRuleTable.fields) + + " FROM " + PushRuleTable.table_name + " pr " + + " LEFT JOIN " + PushRuleEnableTable.table_name + " pre " + + " ON pr.user_name = pre.user_name and pr.rule_id = pre.rule_id " + + " WHERE pr.user_name " + + " IN (" + ",".join("?" for _ in user_ids_to_fetch) + ")" + " AND (pre.enabled is null or pre.enabled = 1)" + " ORDER BY pr.user_name, pr.priority_class DESC, pr.priority DESC" + ) + txn.execute(sql, user_ids_to_fetch) + return txn.fetchall() + + results = {} + + batch_start = 0 + while batch_start < len(user_ids): + batch_end = min(len(user_ids), batch_size) + batch_user_ids = user_ids[batch_start:batch_end] + batch_start = batch_end + + rows = yield self.runInteraction( + "bulk_get_push_rules", f, batch_user_ids + ) + + for r in rows: + rawdict = { + PushRuleTable.fields[i]: r[i] for i in range(len(r)) + } + + if rawdict['user_name'] not in results: + results[rawdict['user_name']] = [] + results[rawdict['user_name']].append(rawdict) + defer.returnValue(results) + + @defer.inlineCallbacks def add_push_rule(self, before, after, **kwargs): vals = kwargs if 'conditions' in vals: diff --git a/synapse/storage/schema/delta/27/event_push_actions.sql b/synapse/storage/schema/delta/27/event_push_actions.sql new file mode 100644 index 0000000000..bdf6ae3f24 --- /dev/null +++ b/synapse/storage/schema/delta/27/event_push_actions.sql @@ -0,0 +1,26 @@ +/* Copyright 2015 OpenMarket Ltd + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +CREATE TABLE IF NOT EXISTS event_push_actions( + room_id TEXT NOT NULL, + event_id TEXT NOT NULL, + user_id TEXT NOT NULL, + profile_tag VARCHAR(32), + actions TEXT NOT NULL, + CONSTRAINT event_id_user_id_profile_tag_uniqueness UNIQUE (room_id, event_id, user_id, profile_tag) +); + + +CREATE INDEX event_push_actions_room_id_event_id_user_id_profile_tag on event_push_actions(room_id, event_id, user_id, profile_tag); |