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