fix push notifications for invites from remote servers

changelog.d/11410.bugfix

@@ -0,0 +1 @@
+Fix not sending out push notifications for invites from remote servers.

synapse/handlers/federation.py

@@ -818,6 +818,8 @@ async def on_invite_request(
             event.room_id, [(event, context)]
         )
 
+        await self._federation_event_handler.notify_remote_invite(event, context)
+
         return event
 
     async def do_remotely_reject_invite(

synapse/handlers/federation_event.py

@@ -58,11 +58,14 @@
 from synapse.logging.context import nested_logging_context, run_in_background
 from synapse.logging.utils import log_function
 from synapse.metrics.background_process_metrics import run_as_background_process
+from synapse.push.bulk_push_rule_evaluator import BulkPushRuleEvaluator
+from synapse.push.httppusher import HttpPusher
 from synapse.replication.http.devices import ReplicationUserDevicesResyncRestServlet
 from synapse.replication.http.federation import (
     ReplicationFederationSendEventsRestServlet,
 )
 from synapse.state import StateResolutionStore
+from synapse.storage.databases.main.event_push_actions import BasePushAction
 from synapse.storage.databases.main.events_worker import EventRedactBehaviour
 from synapse.types import (
     PersistedEventPosition,
@@ -96,6 +99,7 @@ class FederationEventHandler:
     """
 
     def __init__(self, hs: "HomeServer"):
+        self._hs = hs
         self._store = hs.get_datastore()
         self._storage = hs.get_storage()
         self._state_store = self._storage.state
@@ -136,6 +140,25 @@ def __init__(self, hs: "HomeServer"):
 
         self._room_pdu_linearizer = Linearizer("fed_room_pdu")
 
+    async def notify_remote_invite(
+        self, event: EventBase, context: EventContext
+    ) -> None:
+        actions_by_user, _ = await BulkPushRuleEvaluator(
+            self._hs
+        ).get_action_for_event_by_user(event, context)
+        if event.state_key not in actions_by_user:
+            # No pushers for target user. Nothing to do.
+            return
+        # The pusher expects a differently structured push_action, so we wrangle things into the correct shape here
+        action_for_pusher: BasePushAction = {
+            "actions": actions_by_user[event.state_key],
+            "event_id": event.event_id,
+        }
+        pusher_configs = await self._store.get_pushers_by_user_id(event.state_key)
+        for pusher_config in pusher_configs:
+            p = HttpPusher(self._hs, pusher_config)
+            await p.process_push_action(action_for_pusher)
+
     async def on_receive_pdu(self, origin: str, pdu: EventBase) -> None:
         """Process a PDU received via a federation /send/ transaction
 

synapse/push/action_generator.py

@@ -41,4 +41,4 @@ async def handle_push_actions_for_event(
         self, event: EventBase, context: EventContext
     ) -> None:
         with Measure(self.clock, "action_for_event_by_user"):
-            await self.bulk_evaluator.action_for_event_by_user(event, context)
+            await self.bulk_evaluator.store_action_for_event_by_user(event, context)

synapse/push/bulk_push_rule_evaluator.py

@@ -184,15 +184,30 @@ async def _get_power_levels_and_sender_level(
 
         return pl_event.content if pl_event else {}, sender_level
 
-    async def action_for_event_by_user(
+    async def store_action_for_event_by_user(
         self, event: EventBase, context: EventContext
     ) -> 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)
+        actions_by_user, count_as_unread = await self.get_action_for_event_by_user(
+            event, context
+        )
+
+        # 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,
+        )
 
+    async def get_action_for_event_by_user(
+        self, event: EventBase, context: EventContext
+    ) -> Tuple[Dict[str, List[Union[dict, str]]], bool]:
+        count_as_unread = _should_count_as_unread(event, context)
         rules_by_user = await self._get_rules_for_event(event, context)
         actions_by_user: Dict[str, List[Union[dict, str]]] = {}
 
@@ -256,14 +271,7 @@ async def action_for_event_by_user(
                         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,
-        )
+        return actions_by_user, count_as_unread
 
 
 def _condition_checker(

synapse/push/httppusher.py

@@ -26,7 +26,7 @@
 from synapse.logging import opentracing
 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 synapse.storage.databases.main.event_push_actions import BasePushAction
 
 from . import push_rule_evaluator, push_tools
 
@@ -204,7 +204,7 @@ async def _unsafe_process(self) -> None:
                     "app_display_name": self.app_display_name,
                 },
             ):
-                processed = await self._process_one(push_action)
+                processed = await self.process_push_action(push_action)
 
             if processed:
                 http_push_processed_counter.inc()
@@ -274,7 +274,7 @@ async def _unsafe_process(self) -> None:
                     )
                     break
 
-    async def _process_one(self, push_action: HttpPushAction) -> bool:
+    async def process_push_action(self, push_action: BasePushAction) -> bool:
         if "notify" not in push_action["actions"]:
             return True