123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501 |
- # -*- coding: utf-8 -*-
- # Copyright 2015 OpenMarket Ltd
- # Copyright 2017 New Vector 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
- from collections import namedtuple
- from prometheus_client import Counter
- from synapse.api.constants import EventTypes, Membership, RelationTypes
- from synapse.event_auth import get_user_power_level
- from synapse.events import EventBase
- from synapse.events.snapshot import EventContext
- from synapse.state import POWER_KEY
- from synapse.util.async_helpers import Linearizer
- from synapse.util.caches import register_cache
- from synapse.util.caches.descriptors import cached
- from .push_rule_evaluator import PushRuleEvaluatorForEvent
- logger = logging.getLogger(__name__)
- rules_by_room = {}
- push_rules_invalidation_counter = Counter(
- "synapse_push_bulk_push_rule_evaluator_push_rules_invalidation_counter", ""
- )
- push_rules_state_size_counter = Counter(
- "synapse_push_bulk_push_rule_evaluator_push_rules_state_size_counter", ""
- )
- # Measures whether we use the fast path of using state deltas, or if we have to
- # recalculate from scratch
- push_rules_delta_state_cache_metric = register_cache(
- "cache",
- "push_rules_delta_state_cache_metric",
- cache=[], # Meaningless size, as this isn't a cache that stores values
- resizable=False,
- )
- STATE_EVENT_TYPES_TO_MARK_UNREAD = {
- EventTypes.Topic,
- EventTypes.Name,
- EventTypes.RoomAvatar,
- EventTypes.Tombstone,
- }
- def _should_count_as_unread(event: EventBase, context: EventContext) -> bool:
- # Exclude rejected and soft-failed events.
- if context.rejected or event.internal_metadata.is_soft_failed():
- return False
- # Exclude notices.
- if (
- not event.is_state()
- and event.type == EventTypes.Message
- and event.content.get("msgtype") == "m.notice"
- ):
- return False
- # Exclude edits.
- relates_to = event.content.get("m.relates_to", {})
- if relates_to.get("rel_type") == RelationTypes.REPLACE:
- return False
- # Mark events that have a non-empty string body as unread.
- body = event.content.get("body")
- if isinstance(body, str) and body:
- return True
- # Mark some state events as unread.
- if event.is_state() and event.type in STATE_EVENT_TYPES_TO_MARK_UNREAD:
- return True
- # Mark encrypted events as unread.
- if not event.is_state() and event.type == EventTypes.Encrypted:
- return True
- return False
- class BulkPushRuleEvaluator:
- """Calculates the outcome of push rules for an event for all users in the
- room at once.
- """
- def __init__(self, hs):
- self.hs = hs
- self.store = hs.get_datastore()
- self.auth = hs.get_auth()
- self.room_push_rule_cache_metrics = register_cache(
- "cache",
- "room_push_rule_cache",
- cache=[], # Meaningless size, as this isn't a cache that stores values,
- resizable=False,
- )
- async def _get_rules_for_event(self, event, context):
- """This gets the rules for all users in the room at the time of the event,
- as well as the push rules for the invitee if the event is an invite.
- Returns:
- dict of user_id -> push_rules
- """
- room_id = event.room_id
- rules_for_room = await self._get_rules_for_room(room_id)
- rules_by_user = await rules_for_room.get_rules(event, context)
- # if this event is an invite event, we may need to run rules for the user
- # who's been invited, otherwise they won't get told they've been invited
- if event.type == "m.room.member" and event.content["membership"] == "invite":
- invited = event.state_key
- if invited and self.hs.is_mine_id(invited):
- has_pusher = await self.store.user_has_pusher(invited)
- if has_pusher:
- rules_by_user = dict(rules_by_user)
- rules_by_user[invited] = await self.store.get_push_rules_for_user(
- invited
- )
- return rules_by_user
- @cached()
- def _get_rules_for_room(self, room_id):
- """Get the current RulesForRoom object for the given room id
- Returns:
- RulesForRoom
- """
- # It's important that RulesForRoom gets added to self._get_rules_for_room.cache
- # before any lookup methods get called on it as otherwise there may be
- # a race if invalidate_all gets called (which assumes its in the cache)
- return RulesForRoom(
- self.hs,
- room_id,
- self._get_rules_for_room.cache,
- self.room_push_rule_cache_metrics,
- )
- async def _get_power_levels_and_sender_level(self, event, context):
- prev_state_ids = await context.get_prev_state_ids()
- pl_event_id = prev_state_ids.get(POWER_KEY)
- if pl_event_id:
- # fastpath: if there's a power level event, that's all we need, and
- # not having a power level event is an extreme edge case
- pl_event = await self.store.get_event(pl_event_id)
- auth_events = {POWER_KEY: pl_event}
- else:
- auth_events_ids = self.auth.compute_auth_events(
- event, prev_state_ids, for_verification=False
- )
- auth_events = await self.store.get_events(auth_events_ids)
- auth_events = {(e.type, e.state_key): e for e in auth_events.values()}
- sender_level = get_user_power_level(event.sender, auth_events)
- pl_event = auth_events.get(POWER_KEY)
- return pl_event.content if pl_event else {}, sender_level
- async def action_for_event_by_user(self, event, context) -> None:
- """Given an event and context, evaluate the push rules, check if the message
- should increment the unread count, and insert the results into the
- event_push_actions_staging table.
- """
- count_as_unread = _should_count_as_unread(event, context)
- rules_by_user = await self._get_rules_for_event(event, context)
- actions_by_user = {}
- room_members = await self.store.get_joined_users_from_context(event, context)
- (
- power_levels,
- sender_power_level,
- ) = await self._get_power_levels_and_sender_level(event, context)
- evaluator = PushRuleEvaluatorForEvent(
- event, len(room_members), sender_power_level, power_levels
- )
- condition_cache = {}
- for uid, rules in rules_by_user.items():
- if event.sender == uid:
- continue
- if not event.is_state():
- is_ignored = await self.store.is_ignored_by(event.sender, uid)
- if is_ignored:
- continue
- display_name = None
- profile_info = room_members.get(uid)
- if profile_info:
- display_name = profile_info.display_name
- if not display_name:
- # Handle the case where we are pushing a membership event to
- # that user, as they might not be already joined.
- if event.type == EventTypes.Member and event.state_key == uid:
- display_name = event.content.get("displayname", None)
- if count_as_unread:
- # Add an element for the current user if the event needs to be marked as
- # unread, so that add_push_actions_to_staging iterates over it.
- # If the event shouldn't be marked as unread but should notify the
- # current user, it'll be added to the dict later.
- actions_by_user[uid] = []
- for rule in rules:
- if "enabled" in rule and not rule["enabled"]:
- continue
- matches = _condition_checker(
- evaluator, rule["conditions"], uid, display_name, condition_cache
- )
- if matches:
- actions = [x for x in rule["actions"] if x != "dont_notify"]
- if actions and "notify" in actions:
- # Push rules say we should notify the user of this event
- actions_by_user[uid] = actions
- break
- # Mark in the DB staging area the push actions for users who should be
- # notified for this event. (This will then get handled when we persist
- # the event)
- await self.store.add_push_actions_to_staging(
- event.event_id, actions_by_user, count_as_unread,
- )
- def _condition_checker(evaluator, conditions, uid, display_name, cache):
- for cond in conditions:
- _id = cond.get("_id", None)
- if _id:
- res = cache.get(_id, None)
- if res is False:
- return False
- elif res is True:
- continue
- res = evaluator.matches(cond, uid, display_name)
- if _id:
- cache[_id] = bool(res)
- if not res:
- return False
- return True
- class RulesForRoom:
- """Caches push rules for users in a room.
- This efficiently handles users joining/leaving the room by not invalidating
- the entire cache for the room.
- """
- def __init__(self, hs, room_id, rules_for_room_cache, room_push_rule_cache_metrics):
- """
- Args:
- hs (HomeServer)
- room_id (str)
- rules_for_room_cache(Cache): The cache object that caches these
- RoomsForUser objects.
- room_push_rule_cache_metrics (CacheMetric)
- """
- self.room_id = room_id
- self.is_mine_id = hs.is_mine_id
- self.store = hs.get_datastore()
- self.room_push_rule_cache_metrics = room_push_rule_cache_metrics
- self.linearizer = Linearizer(name="rules_for_room")
- self.member_map = {} # event_id -> (user_id, state)
- self.rules_by_user = {} # user_id -> rules
- # The last state group we updated the caches for. If the state_group of
- # a new event comes along, we know that we can just return the cached
- # result.
- # On invalidation of the rules themselves (if the user changes them),
- # we invalidate everything and set state_group to `object()`
- self.state_group = object()
- # A sequence number to keep track of when we're allowed to update the
- # cache. We bump the sequence number when we invalidate the cache. If
- # the sequence number changes while we're calculating stuff we should
- # not update the cache with it.
- self.sequence = 0
- # A cache of user_ids that we *know* aren't interesting, e.g. user_ids
- # owned by AS's, or remote users, etc. (I.e. users we will never need to
- # calculate push for)
- # These never need to be invalidated as we will never set up push for
- # them.
- self.uninteresting_user_set = set()
- # We need to be clever on the invalidating caches callbacks, as
- # otherwise the invalidation callback holds a reference to the object,
- # potentially causing it to leak.
- # To get around this we pass a function that on invalidations looks ups
- # the RoomsForUser entry in the cache, rather than keeping a reference
- # to self around in the callback.
- self.invalidate_all_cb = _Invalidation(rules_for_room_cache, room_id)
- async def get_rules(self, event, context):
- """Given an event context return the rules for all users who are
- currently in the room.
- """
- state_group = context.state_group
- if state_group and self.state_group == state_group:
- logger.debug("Using cached rules for %r", self.room_id)
- self.room_push_rule_cache_metrics.inc_hits()
- return self.rules_by_user
- with (await self.linearizer.queue(())):
- if state_group and self.state_group == state_group:
- logger.debug("Using cached rules for %r", self.room_id)
- self.room_push_rule_cache_metrics.inc_hits()
- return self.rules_by_user
- self.room_push_rule_cache_metrics.inc_misses()
- ret_rules_by_user = {}
- missing_member_event_ids = {}
- if state_group and self.state_group == context.prev_group:
- # If we have a simple delta then we can reuse most of the previous
- # results.
- ret_rules_by_user = self.rules_by_user
- current_state_ids = context.delta_ids
- push_rules_delta_state_cache_metric.inc_hits()
- else:
- current_state_ids = await context.get_current_state_ids()
- push_rules_delta_state_cache_metric.inc_misses()
- push_rules_state_size_counter.inc(len(current_state_ids))
- logger.debug(
- "Looking for member changes in %r %r", state_group, current_state_ids
- )
- # Loop through to see which member events we've seen and have rules
- # for and which we need to fetch
- for key in current_state_ids:
- typ, user_id = key
- if typ != EventTypes.Member:
- continue
- if user_id in self.uninteresting_user_set:
- continue
- if not self.is_mine_id(user_id):
- self.uninteresting_user_set.add(user_id)
- continue
- if self.store.get_if_app_services_interested_in_user(user_id):
- self.uninteresting_user_set.add(user_id)
- continue
- event_id = current_state_ids[key]
- res = self.member_map.get(event_id, None)
- if res:
- user_id, state = res
- if state == Membership.JOIN:
- rules = self.rules_by_user.get(user_id, None)
- if rules:
- ret_rules_by_user[user_id] = rules
- continue
- # If a user has left a room we remove their push rule. If they
- # joined then we readd it later in _update_rules_with_member_event_ids
- ret_rules_by_user.pop(user_id, None)
- missing_member_event_ids[user_id] = event_id
- if missing_member_event_ids:
- # If we have some memebr events we haven't seen, look them up
- # and fetch push rules for them if appropriate.
- logger.debug("Found new member events %r", missing_member_event_ids)
- await self._update_rules_with_member_event_ids(
- ret_rules_by_user, missing_member_event_ids, state_group, event
- )
- else:
- # The push rules didn't change but lets update the cache anyway
- self.update_cache(
- self.sequence,
- members={}, # There were no membership changes
- rules_by_user=ret_rules_by_user,
- state_group=state_group,
- )
- if logger.isEnabledFor(logging.DEBUG):
- logger.debug(
- "Returning push rules for %r %r", self.room_id, ret_rules_by_user.keys()
- )
- return ret_rules_by_user
- async def _update_rules_with_member_event_ids(
- self, ret_rules_by_user, member_event_ids, state_group, event
- ):
- """Update the partially filled rules_by_user dict by fetching rules for
- any newly joined users in the `member_event_ids` list.
- Args:
- ret_rules_by_user (dict): Partiallly filled dict of push rules. Gets
- updated with any new rules.
- member_event_ids (dict): Dict of user id to event id for membership events
- that have happened since the last time we filled rules_by_user
- state_group: The state group we are currently computing push rules
- for. Used when updating the cache.
- """
- sequence = self.sequence
- rows = await self.store.get_membership_from_event_ids(member_event_ids.values())
- members = {row["event_id"]: (row["user_id"], row["membership"]) for row in rows}
- # If the event is a join event then it will be in current state evnts
- # map but not in the DB, so we have to explicitly insert it.
- if event.type == EventTypes.Member:
- for event_id in member_event_ids.values():
- if event_id == event.event_id:
- members[event_id] = (event.state_key, event.membership)
- if logger.isEnabledFor(logging.DEBUG):
- logger.debug("Found members %r: %r", self.room_id, members.values())
- user_ids = {
- user_id
- for user_id, membership in members.values()
- if membership == Membership.JOIN
- }
- logger.debug("Joined: %r", user_ids)
- # Previously we only considered users with pushers or read receipts in that
- # room. We can't do this anymore because we use push actions to calculate unread
- # counts, which don't rely on the user having pushers or sent a read receipt into
- # the room. Therefore we just need to filter for local users here.
- user_ids = list(filter(self.is_mine_id, user_ids))
- rules_by_user = await self.store.bulk_get_push_rules(
- user_ids, on_invalidate=self.invalidate_all_cb
- )
- ret_rules_by_user.update(
- item for item in rules_by_user.items() if item[0] is not None
- )
- self.update_cache(sequence, members, ret_rules_by_user, state_group)
- def invalidate_all(self):
- # Note: Don't hand this function directly to an invalidation callback
- # as it keeps a reference to self and will stop this instance from being
- # GC'd if it gets dropped from the rules_to_user cache. Instead use
- # `self.invalidate_all_cb`
- logger.debug("Invalidating RulesForRoom for %r", self.room_id)
- self.sequence += 1
- self.state_group = object()
- self.member_map = {}
- self.rules_by_user = {}
- push_rules_invalidation_counter.inc()
- def update_cache(self, sequence, members, rules_by_user, state_group):
- if sequence == self.sequence:
- self.member_map.update(members)
- self.rules_by_user = rules_by_user
- self.state_group = state_group
- class _Invalidation(namedtuple("_Invalidation", ("cache", "room_id"))):
- # We rely on _CacheContext implementing __eq__ and __hash__ sensibly,
- # which namedtuple does for us (i.e. two _CacheContext are the same if
- # their caches and keys match). This is important in particular to
- # dedupe when we add callbacks to lru cache nodes, otherwise the number
- # of callbacks would grow.
- def __call__(self):
- rules = self.cache.get(self.room_id, None, update_metrics=False)
- if rules:
- rules.invalidate_all()
|