diff options
| -rw-r--r-- | synapse/handlers/_base.py | 8 | ||||
| -rw-r--r-- | synapse/handlers/federation.py | 10 | ||||
| -rw-r--r-- | synapse/handlers/sync.py | 59 | ||||
| -rw-r--r-- | synapse/push/__init__.py | 22 | ||||
| -rw-r--r-- | synapse/push/action_generator.py | 51 | ||||
| -rw-r--r-- | synapse/push/bulk_push_rule_evaluator.py | 107 | ||||
| -rw-r--r-- | synapse/push/push_rule_evaluator.py | 22 | ||||
| -rw-r--r-- | synapse/rest/client/v2_alpha/sync.py | 1 | ||||
| -rw-r--r-- | synapse/storage/__init__.py | 2 | ||||
| -rw-r--r-- | synapse/storage/event_actions.py | 98 | ||||
| -rw-r--r-- | synapse/storage/push_rule.py | 41 | ||||
| -rw-r--r-- | synapse/storage/registration.py | 12 | ||||
| -rw-r--r-- | synapse/storage/schema/delta/27/event_actions.sql | 26 |
13 files changed, 435 insertions, 24 deletions
diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py index 5fd20285d2..24c4c62698 100644 --- a/synapse/handlers/_base.py +++ b/synapse/handlers/_base.py @@ -19,9 +19,12 @@ from synapse.api.errors import LimitExceededError, SynapseError, AuthError from synapse.crypto.event_signing import add_hashes_and_signatures from synapse.api.constants import Membership, EventTypes from synapse.types import UserID, RoomAlias +from synapse.push.action_generator import ActionGenerator from synapse.util.logcontext import PreserveLoggingContext +from synapse.events.utils import serialize_event + import logging @@ -264,6 +267,11 @@ class BaseHandler(object): event, context=context ) + action_generator = ActionGenerator(self.hs, self.store) + yield action_generator.handle_event(serialize_event( + event, self.clock.time_msec() + )) + destinations = set(extra_destinations) for k, s in context.current_state.items(): try: diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index 28f2ff68d6..0b1221deb5 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -32,10 +32,12 @@ from synapse.crypto.event_signing import ( ) from synapse.types import UserID -from synapse.events.utils import prune_event +from synapse.events.utils import prune_event, serialize_event from synapse.util.retryutils import NotRetryingDestination +from synapse.push.action_generator import ActionGenerator + from twisted.internet import defer import itertools @@ -242,6 +244,12 @@ class FederationHandler(BaseHandler): user = UserID.from_string(event.state_key) yield user_joined_room(self.distributor, user, event.room_id) + if not backfilled and not event.internal_metadata.is_outlier(): + action_generator = ActionGenerator(self.hs, self.store) + yield action_generator.handle_event(serialize_event( + event, self.clock.time_msec()) + ) + @defer.inlineCallbacks def _filter_events_for_server(self, server_name, room_id, events): event_to_state = yield self.store.get_state_for_events( diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index feea407ea2..b1bfdce85b 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -54,6 +54,7 @@ class JoinedSyncResult(collections.namedtuple("JoinedSyncResult", [ "state", # dict[(str, str), FrozenEvent] "ephemeral", "account_data", + "unread_notification_count", ])): __slots__ = [] @@ -66,6 +67,8 @@ class JoinedSyncResult(collections.namedtuple("JoinedSyncResult", [ or self.state or self.ephemeral or self.account_data + # nb the notification count does not, er, count: if there's nothing + # else in the result, we don't need to send it. ) @@ -188,6 +191,18 @@ class SyncHandler(BaseHandler): else: return self.incremental_sync_with_gap(sync_config, since_token) + def last_read_event_id_for_room_and_user(self, room_id, user_id, ephemeral_by_room): + if room_id not in ephemeral_by_room: + return None + for e in ephemeral_by_room[room_id]: + if e['type'] != 'm.receipt': + continue + for receipt_event_id, val in e['content'].items(): + if 'm.read' in val: + if user_id in val['m.read']: + return receipt_event_id + return None + @defer.inlineCallbacks def full_state_sync(self, sync_config, timeline_since_token): """Get a sync for a client which is starting without any state. @@ -314,6 +329,13 @@ class SyncHandler(BaseHandler): room_id, sync_config, now_token, since_token=timeline_since_token ) + notifs = yield self.unread_notifs_for_room_id( + room_id, sync_config, ephemeral_by_room + ) + notif_count = None + if notifs is not None: + notif_count = len(notifs) + current_state = yield self.get_state_at(room_id, now_token) defer.returnValue(JoinedSyncResult( @@ -324,6 +346,7 @@ class SyncHandler(BaseHandler): account_data=self.account_data_for_room( room_id, tags_by_room, account_data_by_room ), + unread_notification_count=notif_count )) def account_data_for_user(self, account_data): @@ -493,6 +516,17 @@ class SyncHandler(BaseHandler): ) now_token = now_token.copy_and_replace("presence_key", presence_key) + # We now fetch all ephemeral events for this room in order to get + # this users current read receipt. This could almost certainly be + # optimised. + _, all_ephemeral_by_room = yield self.ephemeral_by_room( + sync_config, now_token + ) + + now_token, ephemeral_by_room = yield self.ephemeral_by_room( + sync_config, now_token, since_token + ) + rm_handler = self.hs.get_handlers().room_member_handler app_service = yield self.store.get_app_service_by_user_id( sync_config.user.to_string() @@ -552,6 +586,13 @@ class SyncHandler(BaseHandler): else: prev_batch = now_token + notifs = yield self.unread_notifs_for_room_id( + room_id, sync_config, all_ephemeral_by_room + ) + notif_count = None + if notifs is not None: + notif_count = len(notifs) + just_joined = yield self.check_joined_room(sync_config, state) if just_joined: logger.debug("User has just joined %s: needs full state", @@ -572,6 +613,7 @@ class SyncHandler(BaseHandler): account_data=self.account_data_for_room( room_id, tags_by_room, account_data_by_room ), + unread_notification_count=notif_count ) logger.debug("Result for room %s: %r", room_id, room_sync) @@ -845,3 +887,20 @@ class SyncHandler(BaseHandler): if join_event.content["membership"] == Membership.JOIN: return True return False + + @defer.inlineCallbacks + def unread_notifs_for_room_id(self, room_id, sync_config, ephemeral_by_room): + last_unread_event_id = self.last_read_event_id_for_room_and_user( + room_id, sync_config.user.to_string(), ephemeral_by_room + ) + + notifs = [] + if last_unread_event_id: + notifs = yield self.store.get_unread_event_actions_by_room_for_user( + room_id, sync_config.user.to_string(), last_unread_event_id + ) + else: + # There is no new information in this period, so your notification + # count is whatever it was last time. + defer.returnValue(None) + defer.returnValue(notifs) diff --git a/synapse/push/__init__.py b/synapse/push/__init__.py index e7c964bcd2..635dedd523 100644 --- a/synapse/push/__init__.py +++ b/synapse/push/__init__.py @@ -26,7 +26,9 @@ import random logger = logging.getLogger(__name__) - +# Pushers could now be moved to pull out of the event_actions table instead +# of listening on the event stream: this would avoid them having to run the +# rules again. class Pusher(object): INITIAL_BACKOFF = 1000 MAX_BACKOFF = 60 * 60 * 1000 @@ -157,21 +159,7 @@ class Pusher(object): actions = yield rule_evaluator.actions_for_event(single_event) tweaks = rule_evaluator.tweaks_for_actions(actions) - if len(actions) == 0: - logger.warn("Empty actions! Using default action.") - actions = Pusher.DEFAULT_ACTIONS - - if 'notify' not in actions and 'dont_notify' not in actions: - logger.warn("Neither notify nor dont_notify in actions: adding default") - actions.extend(Pusher.DEFAULT_ACTIONS) - - if 'dont_notify' in actions: - logger.debug( - "%s for %s: dont_notify", - single_event['event_id'], self.user_name - ) - processed = True - else: + if 'notify' in actions: rejected = yield self.dispatch_push(single_event, tweaks) self.has_unread = True if isinstance(rejected, list) or isinstance(rejected, tuple): @@ -192,6 +180,8 @@ class Pusher(object): yield self.hs.get_pusherpool().remove_pusher( self.app_id, pk, self.user_name ) + else: + processed = True if not self.alive: return diff --git a/synapse/push/action_generator.py b/synapse/push/action_generator.py new file mode 100644 index 0000000000..148b1bda8e --- /dev/null +++ b/synapse/push/action_generator.py @@ -0,0 +1,51 @@ +# -*- 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 twisted.internet import defer + +import bulk_push_rule_evaluator + +import logging + +logger = logging.getLogger(__name__) + + +class ActionGenerator: + def __init__(self, hs, store): + self.hs = hs + self.store = store + # really we want to get all user ids and all profile tags too, + # since we want the actions for each profile tag for every user and + # also actions for a client with no profile tag for each user. + # Currently the event stream doesn't support profile tags on an + # event stream, so we just run the rules for a client with no profile + # tag (ie. we just need all the users). + + @defer.inlineCallbacks + def handle_event(self, event): + users = yield self.store.get_users_in_room(event['room_id']) + + bulk_evaluator = yield bulk_push_rule_evaluator.evaluator_for_room_id( + event['room_id'], self.hs, self.store + ) + + actions_by_user = bulk_evaluator.action_for_event_by_user(event) + + yield self.store.set_actions_for_event_and_users( + event, + [ + (uid, None, actions) for uid, actions in actions_by_user.items() + ] + ) diff --git a/synapse/push/bulk_push_rule_evaluator.py b/synapse/push/bulk_push_rule_evaluator.py new file mode 100644 index 0000000000..1c0fa72b25 --- /dev/null +++ b/synapse/push/bulk_push_rule_evaluator.py @@ -0,0 +1,107 @@ +# -*- 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. + +import logging +import simplejson as json + +from twisted.internet import defer + +from synapse.types import UserID + +import baserules +from push_rule_evaluator import PushRuleEvaluator + +logger = logging.getLogger(__name__) + + +def decode_rule_json(rule): + rule['conditions'] = json.loads(rule['conditions']) + rule['actions'] = json.loads(rule['actions']) + return rule + + +@defer.inlineCallbacks +def evaluator_for_room_id(room_id, hs, store): + users = yield store.get_users_in_room(room_id) + rules_by_user = yield store.bulk_get_push_rules(users) + rules_by_user = { + uid: baserules.list_with_base_rules( + [decode_rule_json(rule_list) for rule_list in rules_by_user[uid]] + if uid in rules_by_user else [], + UserID.from_string(uid) + ) + for uid in users + } + member_events = yield store.get_current_state( + room_id=room_id, + event_type='m.room.member', + ) + display_names = {} + for ev in member_events: + if ev.content.get("displayname"): + display_names[ev.state_key] = ev.content.get("displayname") + + defer.returnValue(BulkPushRuleEvaluator( + room_id, rules_by_user, display_names, users + )) + + +class BulkPushRuleEvaluator: + """ + Runs push rules for all users in a room. + This is faster than running PushRuleEvaluator for each user because it + fetches all the rules for all the users in one (batched) db query + rarher than doing multiple queries per-user. It currently uses + the same logic to run the actual rules, but could be optimised further + (see https://matrix.org/jira/browse/SYN-562) + """ + def __init__(self, room_id, rules_by_user, display_names, users_in_room): + self.room_id = room_id + self.rules_by_user = rules_by_user + self.display_names = display_names + self.users_in_room = users_in_room + + def action_for_event_by_user(self, event): + actions_by_user = {} + + for uid, rules in self.rules_by_user.items(): + display_name = None + if uid in self.display_names: + display_name = self.display_names[uid] + + for rule in rules: + if 'enabled' in rule and not rule['enabled']: + continue + + # XXX: profile tags + if BulkPushRuleEvaluator.event_matches_rule( + event, rule, + display_name, len(self.users_in_room), None + ): + actions = [x for x in rule['actions'] if x != 'dont_notify'] + if len(actions) > 0: + actions_by_user[uid] = actions + break + return actions_by_user + + @staticmethod + def event_matches_rule(event, rule, + display_name, room_member_count, profile_tag): + matches = True + for cond in rule['conditions']: + matches &= PushRuleEvaluator._event_fulfills_condition( + event, cond, display_name, room_member_count, profile_tag + ) + return matches \ No newline at end of file diff --git a/synapse/push/push_rule_evaluator.py b/synapse/push/push_rule_evaluator.py index 92c7fd048f..40c7622ec4 100644 --- a/synapse/push/push_rule_evaluator.py +++ b/synapse/push/push_rule_evaluator.py @@ -43,7 +43,7 @@ def evaluator_for_user_name_and_profile_tag(user_name, profile_tag, room_id, sto class PushRuleEvaluator: - DEFAULT_ACTIONS = ['dont_notify'] + DEFAULT_ACTIONS = [] INEQUALITY_EXPR = re.compile("^([=<>]*)([0-9]*)$") def __init__(self, user_name, profile_tag, raw_rules, enabled_map, room_id, @@ -85,7 +85,7 @@ class PushRuleEvaluator: """ if ev['user_id'] == self.user_name: # let's assume you probably know about messages you sent yourself - defer.returnValue(['dont_notify']) + defer.returnValue([]) room_id = ev['room_id'] @@ -113,7 +113,8 @@ class PushRuleEvaluator: for c in conditions: matches &= self._event_fulfills_condition( ev, c, display_name=my_display_name, - room_member_count=room_member_count + room_member_count=room_member_count, + profile_tag=self.profile_tag ) logger.debug( "Rule %s %s", @@ -131,6 +132,11 @@ class PushRuleEvaluator: "%s matches for user %s, event %s", r['rule_id'], self.user_name, ev['event_id'] ) + + # filter out dont_notify as we treat an empty actions list + # as dont_notify, and this doesn't take up a row in our database + actions = [x for x in actions if x != 'dont_notify'] + defer.returnValue(actions) logger.info( @@ -151,16 +157,18 @@ class PushRuleEvaluator: re.sub(r'\\\-', '-', x.group(2)))), r) return r - def _event_fulfills_condition(self, ev, condition, display_name, room_member_count): + @staticmethod + def _event_fulfills_condition(ev, condition, + display_name, room_member_count, profile_tag): if condition['kind'] == 'event_match': if 'pattern' not in condition: logger.warn("event_match condition with no pattern") return False # XXX: optimisation: cache our pattern regexps if condition['key'] == 'content.body': - r = r'\b%s\b' % self._glob_to_regexp(condition['pattern']) + r = r'\b%s\b' % PushRuleEvaluator._glob_to_regexp(condition['pattern']) else: - r = r'^%s$' % self._glob_to_regexp(condition['pattern']) + r = r'^%s$' % PushRuleEvaluator._glob_to_regexp(condition['pattern']) val = _value_for_dotted_key(condition['key'], ev) if val is None: return False @@ -169,7 +177,7 @@ class PushRuleEvaluator: elif condition['kind'] == 'device': if 'profile_tag' not in condition: return True - return condition['profile_tag'] == self.profile_tag + return condition['profile_tag'] == profile_tag elif condition['kind'] == 'contains_display_name': # This is special because display names can be different diff --git a/synapse/rest/client/v2_alpha/sync.py b/synapse/rest/client/v2_alpha/sync.py index 35a70ffad1..cd3aef9e07 100644 --- a/synapse/rest/client/v2_alpha/sync.py +++ b/synapse/rest/client/v2_alpha/sync.py @@ -311,6 +311,7 @@ class SyncRestServlet(RestServlet): }, "state": {"events": serialized_state}, "account_data": {"events": account_data}, + "unread_notification_count": room.unread_notification_count } if joined: diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index c46b653f11..a112dd237f 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_actions import EventActionsStore from .state import StateStore from .signatures import SignatureStore @@ -75,6 +76,7 @@ class DataStore(RoomMemberStore, RoomStore, SearchStore, TagsStore, AccountDataStore, + EventActionsStore ): def __init__(self, hs): diff --git a/synapse/storage/event_actions.py b/synapse/storage/event_actions.py new file mode 100644 index 0000000000..3efa445c18 --- /dev/null +++ b/synapse/storage/event_actions.py @@ -0,0 +1,98 @@ +# -*- coding: utf-8 -*- +# Copyright 2014 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 EventActionsStore(SQLBaseStore): + @defer.inlineCallbacks + def set_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, + EventActionsTable.table_name, + values + ) + + @defer.inlineCallbacks + def get_unread_event_actions_by_room_for_user( + self, room_id, user_id, last_read_event_id + ): + def _get_unread_event_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_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_actions_by_room", + _get_unread_event_actions_by_room + ) + defer.returnValue(ret) + + +class EventActionsTable(object): + table_name = "event_actions" diff --git a/synapse/storage/push_rule.py b/synapse/storage/push_rule.py index 5305b7e122..9dec4aa685 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(map(lambda x: "pr."+x, 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 = max(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/registration.py b/synapse/storage/registration.py index 09a05b08ef..4676f225b9 100644 --- a/synapse/storage/registration.py +++ b/synapse/storage/registration.py @@ -292,6 +292,18 @@ class RegistrationStore(SQLBaseStore): defer.returnValue(None) @defer.inlineCallbacks + def get_all_user_ids(self): + """Returns all user ids registered on this homeserver""" + return self.runInteraction( + "get_all_user_ids", + self._get_all_user_ids_txn + ) + + def _get_all_user_ids_txn(self, txn): + txn.execute("SELECT name from users") + return [r[0] for r in txn.fetchall()] + + @defer.inlineCallbacks def count_all_users(self): """Counts all users registered on the homeserver.""" def _count_users(txn): diff --git a/synapse/storage/schema/delta/27/event_actions.sql b/synapse/storage/schema/delta/27/event_actions.sql new file mode 100644 index 0000000000..bbdaee990e --- /dev/null +++ b/synapse/storage/schema/delta/27/event_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_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_actions_room_id_event_id_user_id_profile_tag on event_actions(room_id, event_id, user_id, profile_tag); |