__init__.py 7.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205
  1. # Copyright 2015, 2016 OpenMarket Ltd
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License");
  4. # you may not use this file except in compliance with the License.
  5. # You may obtain a copy of the License at
  6. #
  7. # http://www.apache.org/licenses/LICENSE-2.0
  8. #
  9. # Unless required by applicable law or agreed to in writing, software
  10. # distributed under the License is distributed on an "AS IS" BASIS,
  11. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. # See the License for the specific language governing permissions and
  13. # limitations under the License.
  14. """
  15. This module implements the push rules & notifications portion of the Matrix
  16. specification.
  17. There's a few related features:
  18. * Push notifications (i.e. email or outgoing requests to a Push Gateway).
  19. * Calculation of unread notifications (for /sync and /notifications).
  20. When Synapse receives a new event (locally, via the Client-Server API, or via
  21. federation), the following occurs:
  22. 1. The push rules get evaluated to generate a set of per-user actions.
  23. 2. The event is persisted into the database.
  24. 3. (In the background) The notifier is notified about the new event.
  25. The per-user actions are initially stored in the event_push_actions_staging table,
  26. before getting moved into the event_push_actions table when the event is persisted.
  27. The event_push_actions table is periodically summarised into the event_push_summary
  28. and event_push_summary_stream_ordering tables.
  29. Since push actions block an event from being persisted the generation of push
  30. actions is performance sensitive.
  31. The general interaction of the classes are:
  32. +---------------------------------------------+
  33. | FederationEventHandler/EventCreationHandler |
  34. +---------------------------------------------+
  35. |
  36. v
  37. +-----------------------+ +---------------------------+
  38. | BulkPushRuleEvaluator |---->| PushRuleEvaluatorForEvent |
  39. +-----------------------+ +---------------------------+
  40. |
  41. v
  42. +-----------------------------+
  43. | EventPushActionsWorkerStore |
  44. +-----------------------------+
  45. The notifier notifies the pusher pool of the new event, which checks for affected
  46. users. Each user-configured pusher of the affected users then performs the
  47. previously calculated action.
  48. The general interaction of the classes are:
  49. +----------+
  50. | Notifier |
  51. +----------+
  52. |
  53. v
  54. +------------+ +--------------+
  55. | PusherPool |---->| PusherConfig |
  56. +------------+ +--------------+
  57. |
  58. | +---------------+
  59. +<--->| PusherFactory |
  60. | +---------------+
  61. v
  62. +------------------------+ +-----------------------------------------------+
  63. | EmailPusher/HttpPusher |---->| EventPushActionsWorkerStore/PusherWorkerStore |
  64. +------------------------+ +-----------------------------------------------+
  65. |
  66. v
  67. +-------------------------+
  68. | Mailer/SimpleHttpClient |
  69. +-------------------------+
  70. The Pusher instance also calls out to various utilities for generating payloads
  71. (or email templates), but those interactions are not detailed in this diagram
  72. (and are specific to the type of pusher).
  73. """
  74. import abc
  75. from typing import TYPE_CHECKING, Any, Dict, Optional
  76. import attr
  77. from synapse.types import JsonDict, RoomStreamToken
  78. if TYPE_CHECKING:
  79. from synapse.server import HomeServer
  80. @attr.s(slots=True, auto_attribs=True)
  81. class PusherConfig:
  82. """Parameters necessary to configure a pusher."""
  83. id: Optional[int]
  84. user_name: str
  85. profile_tag: str
  86. kind: str
  87. app_id: str
  88. app_display_name: str
  89. device_display_name: str
  90. pushkey: str
  91. ts: int
  92. lang: Optional[str]
  93. data: Optional[JsonDict]
  94. last_stream_ordering: int
  95. last_success: Optional[int]
  96. failing_since: Optional[int]
  97. enabled: bool
  98. device_id: Optional[str]
  99. # XXX(quenting): The access_token is not persisted anymore for new pushers, but we
  100. # keep it when reading from the database, so that we don't get stale pushers
  101. # while the "set_device_id_for_pushers" background update is running.
  102. access_token: Optional[int]
  103. def as_dict(self) -> Dict[str, Any]:
  104. """Information that can be retrieved about a pusher after creation."""
  105. return {
  106. "app_display_name": self.app_display_name,
  107. "app_id": self.app_id,
  108. "data": self.data,
  109. "device_display_name": self.device_display_name,
  110. "kind": self.kind,
  111. "lang": self.lang,
  112. "profile_tag": self.profile_tag,
  113. "pushkey": self.pushkey,
  114. "enabled": self.enabled,
  115. "device_id": self.device_id,
  116. }
  117. @attr.s(slots=True, auto_attribs=True)
  118. class ThrottleParams:
  119. """Parameters for controlling the rate of sending pushes via email."""
  120. last_sent_ts: int
  121. throttle_ms: int
  122. class Pusher(metaclass=abc.ABCMeta):
  123. def __init__(self, hs: "HomeServer", pusher_config: PusherConfig):
  124. self.hs = hs
  125. self.store = self.hs.get_datastores().main
  126. self.clock = self.hs.get_clock()
  127. self.pusher_id = pusher_config.id
  128. self.user_id = pusher_config.user_name
  129. self.app_id = pusher_config.app_id
  130. self.pushkey = pusher_config.pushkey
  131. self.last_stream_ordering = pusher_config.last_stream_ordering
  132. # This is the highest stream ordering we know it's safe to process.
  133. # When new events arrive, we'll be given a window of new events: we
  134. # should honour this rather than just looking for anything higher
  135. # because of potential out-of-order event serialisation.
  136. self.max_stream_ordering = self.store.get_room_max_stream_ordering()
  137. def on_new_notifications(self, max_token: RoomStreamToken) -> None:
  138. # We just use the minimum stream ordering and ignore the vector clock
  139. # component. This is safe to do as long as we *always* ignore the vector
  140. # clock components.
  141. max_stream_ordering = max_token.stream
  142. self.max_stream_ordering = max(max_stream_ordering, self.max_stream_ordering)
  143. self._start_processing()
  144. @abc.abstractmethod
  145. def _start_processing(self) -> None:
  146. """Start processing push notifications."""
  147. raise NotImplementedError()
  148. @abc.abstractmethod
  149. def on_new_receipts(self) -> None:
  150. raise NotImplementedError()
  151. @abc.abstractmethod
  152. def on_started(self, have_notifs: bool) -> None:
  153. """Called when this pusher has been started.
  154. Args:
  155. should_check_for_notifs: Whether we should immediately
  156. check for push to send. Set to False only if it's known there
  157. is nothing to send
  158. """
  159. raise NotImplementedError()
  160. @abc.abstractmethod
  161. def on_stop(self) -> None:
  162. raise NotImplementedError()
  163. class PusherConfigException(Exception):
  164. """An error occurred when creating a pusher."""