Implement push rule evaluation in Rust. (#13838)

3 files changed, 63 insertions, 381 deletions

diff --git a/synapse/push/bulk_push_rule_evaluator.py b/synapse/push/bulk_push_rule_evaluator.py
index 32313e3bcf..60f3129005 100644
--- a/synapse/push/bulk_push_rule_evaluator.py
+++ b/synapse/push/bulk_push_rule_evaluator.py
@@ -17,6 +17,7 @@ import itertools
 import logging
 from typing import (
     TYPE_CHECKING,
+    Any,
     Collection,
     Dict,
     Iterable,
@@ -37,13 +38,11 @@ from synapse.events.snapshot import EventContext
 from synapse.state import POWER_KEY
 from synapse.storage.databases.main.roommember import EventIdMembership
 from synapse.storage.state import StateFilter
-from synapse.synapse_rust.push import FilteredPushRules, PushRule
+from synapse.synapse_rust.push import FilteredPushRules, PushRule, PushRuleEvaluator
 from synapse.util.caches import register_cache
 from synapse.util.metrics import measure_func
 from synapse.visibility import filter_event_for_clients_with_state
 
-from .push_rule_evaluator import PushRuleEvaluatorForEvent
-
 if TYPE_CHECKING:
     from synapse.server import HomeServer
 
@@ -290,11 +289,11 @@ class BulkPushRuleEvaluator:
             if relation.rel_type == RelationTypes.THREAD:
                 thread_id = relation.parent_id
 
-        evaluator = PushRuleEvaluatorForEvent(
-            event,
+        evaluator = PushRuleEvaluator(
+            _flatten_dict(event),
             room_member_count,
             sender_power_level,
-            power_levels,
+            power_levels.get("notifications", {}),
             relations,
             self._relations_match_enabled,
         )
@@ -338,17 +337,10 @@ class BulkPushRuleEvaluator:
                 # current user, it'll be added to the dict later.
                 actions_by_user[uid] = []
 
-            for rule, enabled in rules.rules():
-                if not enabled:
-                    continue
-
-                matches = evaluator.check_conditions(rule.conditions, uid, display_name)
-                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
+            actions = evaluator.run(rules, uid, display_name)
+            if "notify" in actions:
+                # Push rules say we should notify the user of this event
+                actions_by_user[uid] = actions
 
         # 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
@@ -365,3 +357,21 @@ MemberMap = Dict[str, Optional[EventIdMembership]]
 Rule = Dict[str, dict]
 RulesByUser = Dict[str, List[Rule]]
 StateGroup = Union[object, int]
+
+
+def _flatten_dict(
+    d: Union[EventBase, Mapping[str, Any]],
+    prefix: Optional[List[str]] = None,
+    result: Optional[Dict[str, str]] = None,
+) -> Dict[str, str]:
+    if prefix is None:
+        prefix = []
+    if result is None:
+        result = {}
+    for key, value in d.items():
+        if isinstance(value, str):
+            result[".".join(prefix + [key])] = value.lower()
+        elif isinstance(value, Mapping):
+            _flatten_dict(value, prefix=(prefix + [key]), result=result)
+
+    return result
diff --git a/synapse/push/httppusher.py b/synapse/push/httppusher.py
index e96fb45e9f..b048b03a74 100644
--- a/synapse/push/httppusher.py
+++ b/synapse/push/httppusher.py
@@ -14,7 +14,7 @@
 # limitations under the License.
 import logging
 import urllib.parse
-from typing import TYPE_CHECKING, Any, Dict, Iterable, Optional, Union
+from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Union
 
 from prometheus_client import Counter
 
@@ -28,7 +28,7 @@ from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.push import Pusher, PusherConfig, PusherConfigException
 from synapse.storage.databases.main.event_push_actions import HttpPushAction
 
-from . import push_rule_evaluator, push_tools
+from . import push_tools
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
@@ -56,6 +56,39 @@ http_badges_failed_counter = Counter(
 )
 
 
+def tweaks_for_actions(actions: List[Union[str, Dict]]) -> Dict[str, Any]:
+    """
+    Converts a list of actions into a `tweaks` dict (which can then be passed to
+        the push gateway).
+
+    This function ignores all actions other than `set_tweak` actions, and treats
+    absent `value`s as `True`, which agrees with the only spec-defined treatment
+    of absent `value`s (namely, for `highlight` tweaks).
+
+    Args:
+        actions: list of actions
+            e.g. [
+                {"set_tweak": "a", "value": "AAA"},
+                {"set_tweak": "b", "value": "BBB"},
+                {"set_tweak": "highlight"},
+                "notify"
+            ]
+
+    Returns:
+        dictionary of tweaks for those actions
+            e.g. {"a": "AAA", "b": "BBB", "highlight": True}
+    """
+    tweaks = {}
+    for a in actions:
+        if not isinstance(a, dict):
+            continue
+        if "set_tweak" in a:
+            # value is allowed to be absent in which case the value assumed
+            # should be True.
+            tweaks[a["set_tweak"]] = a.get("value", True)
+    return tweaks
+
+
 class HttpPusher(Pusher):
     INITIAL_BACKOFF_SEC = 1  # in seconds because that's what Twisted takes
     MAX_BACKOFF_SEC = 60 * 60
@@ -281,7 +314,7 @@ class HttpPusher(Pusher):
         if "notify" not in push_action.actions:
             return True
 
-        tweaks = push_rule_evaluator.tweaks_for_actions(push_action.actions)
+        tweaks = tweaks_for_actions(push_action.actions)
         badge = await push_tools.get_badge_count(
             self.hs.get_datastores().main,
             self.user_id,
diff --git a/synapse/push/push_rule_evaluator.py b/synapse/push/push_rule_evaluator.py
deleted file mode 100644
index f8176c5a42..0000000000
--- a/synapse/push/push_rule_evaluator.py
+++ /dev/null
@@ -1,361 +0,0 @@
-# Copyright 2015, 2016 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
-import re
-from typing import (
-    Any,
-    Dict,
-    List,
-    Mapping,
-    Optional,
-    Pattern,
-    Sequence,
-    Set,
-    Tuple,
-    Union,
-)
-
-from matrix_common.regex import glob_to_regex, to_word_pattern
-
-from synapse.events import EventBase
-from synapse.types import UserID
-from synapse.util.caches.lrucache import LruCache
-
-logger = logging.getLogger(__name__)
-
-
-GLOB_REGEX = re.compile(r"\\\[(\\\!|)(.*)\\\]")
-IS_GLOB = re.compile(r"[\?\*\[\]]")
-INEQUALITY_EXPR = re.compile("^([=<>]*)([0-9]*)$")
-
-
-def _room_member_count(condition: Mapping[str, Any], room_member_count: int) -> bool:
-    return _test_ineq_condition(condition, room_member_count)
-
-
-def _sender_notification_permission(
-    condition: Mapping[str, Any],
-    sender_power_level: Optional[int],
-    power_levels: Dict[str, Union[int, Dict[str, int]]],
-) -> bool:
-    if sender_power_level is None:
-        return False
-
-    notif_level_key = condition.get("key")
-    if notif_level_key is None:
-        return False
-
-    notif_levels = power_levels.get("notifications", {})
-    assert isinstance(notif_levels, dict)
-    room_notif_level = notif_levels.get(notif_level_key, 50)
-
-    return sender_power_level >= room_notif_level
-
-
-def _test_ineq_condition(condition: Mapping[str, Any], number: int) -> bool:
-    if "is" not in condition:
-        return False
-    m = INEQUALITY_EXPR.match(condition["is"])
-    if not m:
-        return False
-    ineq = m.group(1)
-    rhs = m.group(2)
-    if not rhs.isdigit():
-        return False
-    rhs_int = int(rhs)
-
-    if ineq == "" or ineq == "==":
-        return number == rhs_int
-    elif ineq == "<":
-        return number < rhs_int
-    elif ineq == ">":
-        return number > rhs_int
-    elif ineq == ">=":
-        return number >= rhs_int
-    elif ineq == "<=":
-        return number <= rhs_int
-    else:
-        return False
-
-
-def tweaks_for_actions(actions: List[Union[str, Dict]]) -> Dict[str, Any]:
-    """
-    Converts a list of actions into a `tweaks` dict (which can then be passed to
-        the push gateway).
-
-    This function ignores all actions other than `set_tweak` actions, and treats
-    absent `value`s as `True`, which agrees with the only spec-defined treatment
-    of absent `value`s (namely, for `highlight` tweaks).
-
-    Args:
-        actions: list of actions
-            e.g. [
-                {"set_tweak": "a", "value": "AAA"},
-                {"set_tweak": "b", "value": "BBB"},
-                {"set_tweak": "highlight"},
-                "notify"
-            ]
-
-    Returns:
-        dictionary of tweaks for those actions
-            e.g. {"a": "AAA", "b": "BBB", "highlight": True}
-    """
-    tweaks = {}
-    for a in actions:
-        if not isinstance(a, dict):
-            continue
-        if "set_tweak" in a:
-            # value is allowed to be absent in which case the value assumed
-            # should be True.
-            tweaks[a["set_tweak"]] = a.get("value", True)
-    return tweaks
-
-
-class PushRuleEvaluatorForEvent:
-    def __init__(
-        self,
-        event: EventBase,
-        room_member_count: int,
-        sender_power_level: Optional[int],
-        power_levels: Dict[str, Union[int, Dict[str, int]]],
-        relations: Dict[str, Set[Tuple[str, str]]],
-        relations_match_enabled: bool,
-    ):
-        self._event = event
-        self._room_member_count = room_member_count
-        self._sender_power_level = sender_power_level
-        self._power_levels = power_levels
-        self._relations = relations
-        self._relations_match_enabled = relations_match_enabled
-
-        # Maps strings of e.g. 'content.body' -> event["content"]["body"]
-        self._value_cache = _flatten_dict(event)
-
-        # Maps cache keys to final values.
-        self._condition_cache: Dict[str, bool] = {}
-
-    def check_conditions(
-        self, conditions: Sequence[Mapping], uid: str, display_name: Optional[str]
-    ) -> bool:
-        """
-        Returns true if a user's conditions/user ID/display name match the event.
-
-        Args:
-            conditions: The user's conditions to match.
-            uid: The user's MXID.
-            display_name: The display name.
-
-        Returns:
-             True if all conditions match the event, False otherwise.
-        """
-        for cond in conditions:
-            _cache_key = cond.get("_cache_key", None)
-            if _cache_key:
-                res = self._condition_cache.get(_cache_key, None)
-                if res is False:
-                    return False
-                elif res is True:
-                    continue
-
-            res = self.matches(cond, uid, display_name)
-            if _cache_key:
-                self._condition_cache[_cache_key] = bool(res)
-
-            if not res:
-                return False
-
-        return True
-
-    def matches(
-        self, condition: Mapping[str, Any], user_id: str, display_name: Optional[str]
-    ) -> bool:
-        """
-        Returns true if a user's condition/user ID/display name match the event.
-
-        Args:
-            condition: The user's condition to match.
-            uid: The user's MXID.
-            display_name: The display name, or None if there is not one.
-
-        Returns:
-             True if the condition matches the event, False otherwise.
-        """
-        if condition["kind"] == "event_match":
-            return self._event_match(condition, user_id)
-        elif condition["kind"] == "contains_display_name":
-            return self._contains_display_name(display_name)
-        elif condition["kind"] == "room_member_count":
-            return _room_member_count(condition, self._room_member_count)
-        elif condition["kind"] == "sender_notification_permission":
-            return _sender_notification_permission(
-                condition, self._sender_power_level, self._power_levels
-            )
-        elif (
-            condition["kind"] == "org.matrix.msc3772.relation_match"
-            and self._relations_match_enabled
-        ):
-            return self._relation_match(condition, user_id)
-        else:
-            # XXX This looks incorrect -- we have reached an unknown condition
-            #     kind and are unconditionally returning that it matches. Note
-            #     that it seems possible to provide a condition to the /pushrules
-            #     endpoint with an unknown kind, see _rule_tuple_from_request_object.
-            return True
-
-    def _event_match(self, condition: Mapping, user_id: str) -> bool:
-        """
-        Check an "event_match" push rule condition.
-
-        Args:
-            condition: The "event_match" push rule condition to match.
-            user_id: The user's MXID.
-
-        Returns:
-             True if the condition matches the event, False otherwise.
-        """
-        pattern = condition.get("pattern", None)
-
-        if not pattern:
-            pattern_type = condition.get("pattern_type", None)
-            if pattern_type == "user_id":
-                pattern = user_id
-            elif pattern_type == "user_localpart":
-                pattern = UserID.from_string(user_id).localpart
-
-        if not pattern:
-            logger.warning("event_match condition with no pattern")
-            return False
-
-        # XXX: optimisation: cache our pattern regexps
-        if condition["key"] == "content.body":
-            body = self._event.content.get("body", None)
-            if not body or not isinstance(body, str):
-                return False
-
-            return _glob_matches(pattern, body, word_boundary=True)
-        else:
-            haystack = self._value_cache.get(condition["key"], None)
-            if haystack is None:
-                return False
-
-            return _glob_matches(pattern, haystack)
-
-    def _contains_display_name(self, display_name: Optional[str]) -> bool:
-        """
-        Check an "event_match" push rule condition.
-
-        Args:
-            display_name: The display name, or None if there is not one.
-
-        Returns:
-             True if the display name is found in the event body, False otherwise.
-        """
-        if not display_name:
-            return False
-
-        body = self._event.content.get("body", None)
-        if not body or not isinstance(body, str):
-            return False
-
-        # Similar to _glob_matches, but do not treat display_name as a glob.
-        r = regex_cache.get((display_name, False, True), None)
-        if not r:
-            r1 = re.escape(display_name)
-            r1 = to_word_pattern(r1)
-            r = re.compile(r1, flags=re.IGNORECASE)
-            regex_cache[(display_name, False, True)] = r
-
-        return bool(r.search(body))
-
-    def _relation_match(self, condition: Mapping, user_id: str) -> bool:
-        """
-        Check an "relation_match" push rule condition.
-
-        Args:
-            condition: The "event_match" push rule condition to match.
-            user_id: The user's MXID.
-
-        Returns:
-             True if the condition matches the event, False otherwise.
-        """
-        rel_type = condition.get("rel_type")
-        if not rel_type:
-            logger.warning("relation_match condition missing rel_type")
-            return False
-
-        sender_pattern = condition.get("sender")
-        if sender_pattern is None:
-            sender_type = condition.get("sender_type")
-            if sender_type == "user_id":
-                sender_pattern = user_id
-        type_pattern = condition.get("type")
-
-        # If any other relations matches, return True.
-        for sender, event_type in self._relations.get(rel_type, ()):
-            if sender_pattern and not _glob_matches(sender_pattern, sender):
-                continue
-            if type_pattern and not _glob_matches(type_pattern, event_type):
-                continue
-            # All values must have matched.
-            return True
-
-        # No relations matched.
-        return False
-
-
-# Caches (string, is_glob, word_boundary) -> regex for push. See _glob_matches
-regex_cache: LruCache[Tuple[str, bool, bool], Pattern] = LruCache(
-    50000, "regex_push_cache"
-)
-
-
-def _glob_matches(glob: str, value: str, word_boundary: bool = False) -> bool:
-    """Tests if value matches glob.
-
-    Args:
-        glob
-        value: String to test against glob.
-        word_boundary: Whether to match against word boundaries or entire
-            string. Defaults to False.
-    """
-
-    try:
-        r = regex_cache.get((glob, True, word_boundary), None)
-        if not r:
-            r = glob_to_regex(glob, word_boundary=word_boundary)
-            regex_cache[(glob, True, word_boundary)] = r
-        return bool(r.search(value))
-    except re.error:
-        logger.warning("Failed to parse glob to regex: %r", glob)
-        return False
-
-
-def _flatten_dict(
-    d: Union[EventBase, Mapping[str, Any]],
-    prefix: Optional[List[str]] = None,
-    result: Optional[Dict[str, str]] = None,
-) -> Dict[str, str]:
-    if prefix is None:
-        prefix = []
-    if result is None:
-        result = {}
-    for key, value in d.items():
-        if isinstance(value, str):
-            result[".".join(prefix + [key])] = value.lower()
-        elif isinstance(value, Mapping):
-            _flatten_dict(value, prefix=(prefix + [key]), result=result)
-
-    return result