diff --git a/zerver/lib/test_classes.py b/zerver/lib/test_classes.py
index 80a5dbd383..c8b14cb932 100644
--- a/zerver/lib/test_classes.py
+++ b/zerver/lib/test_classes.py
@@ -1337,26 +1337,28 @@ Output:
             sender_is_muted=kwargs.get("sender_is_muted", False),
         )
 
-    def get_maybe_enqueue_notifications_parameters(self, **kwargs: Any) -> Dict[str, Any]:
+    def get_maybe_enqueue_notifications_parameters(
+        self, *, message_id: int, user_id: int, sender_id: int, **kwargs: Any
+    ) -> Dict[str, Any]:
         """
         Returns a dictionary with the passed parameters, after filling up the
         missing data with default values, for testing what was passed to the
         `maybe_enqueue_notifications` method.
         """
-        parameters: Dict[str, Any] = dict(
-            private_message=False,
-            mentioned=False,
-            wildcard_mention_notify=False,
-            stream_push_notify=False,
-            stream_email_notify=False,
-            stream_name=None,
-            online_push_enabled=False,
-            idle=True,
-            already_notified={"email_notified": False, "push_notified": False},
+        user_notifications_data = self.create_user_notifications_data_object(
+            user_id=user_id, **kwargs
+        )
+        return dict(
+            user_data=user_notifications_data,
+            message_id=message_id,
+            sender_id=sender_id,
+            private_message=kwargs.get("private_message", False),
+            stream_name=kwargs.get("stream_name", None),
+            idle=kwargs.get("idle", True),
+            already_notified=kwargs.get(
+                "already_notified", {"email_notified": False, "push_notified": False}
+            ),
         )
-
-        # Values from `kwargs` will replace those from `parameters`
-        return {**parameters, **kwargs}
 
 
 class WebhookTestCase(ZulipTestCase):
diff --git a/zerver/tests/test_event_queue.py b/zerver/tests/test_event_queue.py
index 102fd6f40b..e14ac14902 100644
--- a/zerver/tests/test_event_queue.py
+++ b/zerver/tests/test_event_queue.py
@@ -1,5 +1,5 @@
 import time
-from typing import Any, Callable, Collection, Dict, List, Optional, Tuple, Union
+from typing import Any, Callable, Collection, Dict, List
 from unittest import mock
 
 import orjson
@@ -25,125 +25,34 @@ class MissedMessageNotificationsTest(ZulipTestCase):
     """Tests the logic for when missed-message notifications
     should be triggered, based on user settings"""
 
-    def check_will_notify(self, **kwargs: Any) -> Tuple[str, str]:
-        hamlet = self.example_user("hamlet")
-        kwargs["user_profile_id"] = hamlet.id
-        kwargs["message_id"] = 32
+    def test_maybe_enqueue_notifications(self) -> None:
+        # We've already tested the "when to send notifications" logic as part of the
+        # notification_data module.
+        # This test is for verifying whether `maybe_enqueue_notifications` returns the
+        # `already_notified` data correctly.
+        params = self.get_maybe_enqueue_notifications_parameters(
+            message_id=1, user_id=1, sender_id=2
+        )
 
-        # Fill up the parameters which weren't sent with defaults.
-        kwargs = self.get_maybe_enqueue_notifications_parameters(**kwargs)
-
-        email_notice = None
-        mobile_notice = None
         with mock_queue_publish(
             "zerver.tornado.event_queue.queue_json_publish"
         ) as mock_queue_json_publish:
-            notified = maybe_enqueue_notifications(**kwargs)
-            for entry in mock_queue_json_publish.call_args_list:
-                args = entry[0]
-                if args[0] == "missedmessage_mobile_notifications":
-                    mobile_notice = args[1]
-                if args[0] == "missedmessage_emails":
-                    email_notice = args[1]
+            notified = maybe_enqueue_notifications(**params)
+            mock_queue_json_publish.assert_not_called()
 
-            # Now verify the return value matches the queue actions
-            if email_notice:
-                self.assertTrue(notified["email_notified"])
-            else:
-                self.assertFalse(notified.get("email_notified", False))
-            if mobile_notice:
-                self.assertTrue(notified["push_notified"])
-            else:
-                self.assertFalse(notified.get("push_notified", False))
-        return email_notice, mobile_notice
+        with mock_queue_publish(
+            "zerver.tornado.event_queue.queue_json_publish"
+        ) as mock_queue_json_publish:
+            params["private_message"] = True
+            notified = maybe_enqueue_notifications(**params)
+            self.assertTrue(mock_queue_json_publish.call_count, 2)
 
-    def test_enqueue_notifications(self) -> None:
-        # Boring message doesn't send a notice
-        email_notice, mobile_notice = self.check_will_notify()
-        self.assertTrue(email_notice is None)
-        self.assertTrue(mobile_notice is None)
+            queues_pushed = [entry[0][0] for entry in mock_queue_json_publish.call_args_list]
+            self.assertIn("missedmessage_mobile_notifications", queues_pushed)
+            self.assertIn("missedmessage_emails", queues_pushed)
 
-        # Private message sends a notice
-        email_notice, mobile_notice = self.check_will_notify(private_message=True)
-        self.assertTrue(email_notice is not None)
-        self.assertTrue(mobile_notice is not None)
-
-        # Private message won't double-send either notice if we've
-        # already sent notices before.
-        email_notice, mobile_notice = self.check_will_notify(
-            private_message=True, already_notified={"push_notified": True, "email_notified": False}
-        )
-        self.assertTrue(email_notice is not None)
-        self.assertTrue(mobile_notice is None)
-
-        email_notice, mobile_notice = self.check_will_notify(
-            private_message=True, already_notified={"push_notified": False, "email_notified": True}
-        )
-        self.assertTrue(email_notice is None)
-        self.assertTrue(mobile_notice is not None)
-
-        # Mention sends a notice
-        email_notice, mobile_notice = self.check_will_notify(mentioned=True)
-        self.assertTrue(email_notice is not None)
-        self.assertTrue(mobile_notice is not None)
-
-        # Wildcard mention triggers both email and push notices (Like a
-        # direct mention, whether the notice is actually delivered is
-        # determined later, in the email/push notification code)
-        email_notice, mobile_notice = self.check_will_notify(wildcard_mention_notify=True)
-        self.assertTrue(email_notice is not None)
-        self.assertTrue(mobile_notice is not None)
-
-        # stream_push_notify pushes but doesn't email
-        email_notice, mobile_notice = self.check_will_notify(
-            stream_push_notify=True, stream_name="Denmark"
-        )
-        self.assertTrue(email_notice is None)
-        self.assertTrue(mobile_notice is not None)
-
-        # stream_email_notify emails but doesn't push
-        email_notice, mobile_notice = self.check_will_notify(
-            stream_email_notify=True, stream_name="Denmark"
-        )
-        self.assertTrue(email_notice is not None)
-        self.assertTrue(mobile_notice is None)
-
-        # Private message doesn't send a notice if not idle
-        email_notice, mobile_notice = self.check_will_notify(private_message=True, idle=False)
-        self.assertTrue(email_notice is None)
-        self.assertTrue(mobile_notice is None)
-
-        # Mention doesn't send a notice if not idle
-        email_notice, mobile_notice = self.check_will_notify(mentioned=True, idle=False)
-        self.assertTrue(email_notice is None)
-        self.assertTrue(mobile_notice is None)
-
-        # Wildcard mention doesn't send a notice if not idle
-        email_notice, mobile_notice = self.check_will_notify(
-            wildcard_mention_notify=True, idle=False
-        )
-        self.assertTrue(email_notice is None)
-        self.assertTrue(mobile_notice is None)
-
-        # Private message sends push but not email if not idle but online_push_enabled
-        email_notice, mobile_notice = self.check_will_notify(
-            private_message=True, online_push_enabled=True, idle=False
-        )
-        self.assertTrue(email_notice is None)
-        self.assertTrue(mobile_notice is not None)
-
-        # Stream message sends push (if enabled for that stream) but not email if not
-        # idle but online_push_enabled
-        email_notice, mobile_notice = self.check_will_notify(
-            stream_push_notify=True,
-            stream_email_notify=True,
-            stream_name="Denmark",
-            online_push_enabled=True,
-            idle=False,
-            already_notified={},
-        )
-        self.assertTrue(email_notice is None)
-        self.assertTrue(mobile_notice is not None)
+            self.assertTrue(notified["email_notified"])
+            self.assertTrue(notified["push_notified"])
 
     def tornado_call(
         self,
@@ -246,23 +155,14 @@ class MissedMessageNotificationsTest(ZulipTestCase):
 
         def assert_maybe_enqueue_notifications_call_args(
             args_dict: Collection[Any],
-            user_profile_id: int,
             message_id: int,
-            **kwargs: Union[bool, Dict[str, bool], Optional[str]],
+            **kwargs: Any,
         ) -> None:
-            kwargs = self.get_maybe_enqueue_notifications_parameters(**kwargs)
-            expected_args_dict = dict(
-                user_profile_id=user_profile_id,
+            expected_args_dict = self.get_maybe_enqueue_notifications_parameters(
+                user_id=user_profile.id,
+                sender_id=iago.id,
                 message_id=message_id,
-                private_message=kwargs["private_message"],
-                mentioned=kwargs["mentioned"],
-                wildcard_mention_notify=kwargs["wildcard_mention_notify"],
-                stream_push_notify=kwargs["stream_push_notify"],
-                stream_email_notify=kwargs["stream_email_notify"],
-                stream_name=kwargs["stream_name"],
-                online_push_enabled=kwargs["online_push_enabled"],
-                idle=kwargs["idle"],
-                already_notified=kwargs["already_notified"],
+                **kwargs,
             )
             self.assertEqual(args_dict, expected_args_dict)
 
@@ -285,7 +185,6 @@ class MissedMessageNotificationsTest(ZulipTestCase):
 
             assert_maybe_enqueue_notifications_call_args(
                 args_dict=args_dict,
-                user_profile_id=user_profile.id,
                 message_id=msg_id,
                 stream_name="Denmark",
                 already_notified={"email_notified": False, "push_notified": False},
@@ -303,7 +202,6 @@ class MissedMessageNotificationsTest(ZulipTestCase):
 
             assert_maybe_enqueue_notifications_call_args(
                 args_dict=args_dict,
-                user_profile_id=user_profile.id,
                 message_id=msg_id,
                 private_message=True,
                 already_notified={"email_notified": True, "push_notified": True},
@@ -323,8 +221,8 @@ class MissedMessageNotificationsTest(ZulipTestCase):
 
             assert_maybe_enqueue_notifications_call_args(
                 args_dict=args_dict,
-                user_profile_id=user_profile.id,
                 message_id=msg_id,
+                flags=["mentioned"],
                 mentioned=True,
                 stream_name="Denmark",
                 already_notified={"email_notified": True, "push_notified": True},
@@ -342,8 +240,8 @@ class MissedMessageNotificationsTest(ZulipTestCase):
 
             assert_maybe_enqueue_notifications_call_args(
                 args_dict=args_dict,
-                user_profile_id=user_profile.id,
                 message_id=msg_id,
+                flags=["wildcard_mentioned"],
                 wildcard_mention_notify=True,
                 stream_name="Denmark",
                 already_notified={"email_notified": True, "push_notified": True},
@@ -362,7 +260,7 @@ class MissedMessageNotificationsTest(ZulipTestCase):
 
             assert_maybe_enqueue_notifications_call_args(
                 args_dict=args_dict,
-                user_profile_id=user_profile.id,
+                flags=["wildcard_mentioned"],
                 message_id=msg_id,
                 stream_name="Denmark",
                 already_notified={"email_notified": False, "push_notified": False},
@@ -383,8 +281,8 @@ class MissedMessageNotificationsTest(ZulipTestCase):
 
             assert_maybe_enqueue_notifications_call_args(
                 args_dict=args_dict,
-                user_profile_id=user_profile.id,
                 message_id=msg_id,
+                flags=["wildcard_mentioned"],
                 stream_name="Denmark",
                 already_notified={"email_notified": False, "push_notified": False},
             )
@@ -408,8 +306,8 @@ class MissedMessageNotificationsTest(ZulipTestCase):
 
             assert_maybe_enqueue_notifications_call_args(
                 args_dict=args_dict,
-                user_profile_id=user_profile.id,
                 message_id=msg_id,
+                flags=["wildcard_mentioned"],
                 wildcard_mention_notify=True,
                 stream_name="Denmark",
                 already_notified={"email_notified": True, "push_notified": True},
@@ -432,7 +330,6 @@ class MissedMessageNotificationsTest(ZulipTestCase):
 
             assert_maybe_enqueue_notifications_call_args(
                 args_dict=args_dict,
-                user_profile_id=user_profile.id,
                 message_id=msg_id,
                 stream_push_notify=True,
                 stream_email_notify=False,
@@ -455,7 +352,6 @@ class MissedMessageNotificationsTest(ZulipTestCase):
 
             assert_maybe_enqueue_notifications_call_args(
                 args_dict=args_dict,
-                user_profile_id=user_profile.id,
                 message_id=msg_id,
                 stream_push_notify=False,
                 stream_email_notify=True,
@@ -486,7 +382,6 @@ class MissedMessageNotificationsTest(ZulipTestCase):
 
             assert_maybe_enqueue_notifications_call_args(
                 args_dict=args_dict,
-                user_profile_id=user_profile.id,
                 message_id=msg_id,
                 stream_name="Denmark",
                 already_notified={"email_notified": False, "push_notified": False},
@@ -510,7 +405,6 @@ class MissedMessageNotificationsTest(ZulipTestCase):
 
             assert_maybe_enqueue_notifications_call_args(
                 args_dict=args_dict,
-                user_profile_id=user_profile.id,
                 message_id=msg_id,
                 stream_name="Denmark",
                 already_notified={"email_notified": False, "push_notified": False},
@@ -530,8 +424,17 @@ class MissedMessageNotificationsTest(ZulipTestCase):
         msg_id = self.send_personal_message(iago, user_profile)
         with mock.patch("zerver.tornado.event_queue.maybe_enqueue_notifications") as mock_enqueue:
             missedmessage_hook(user_profile.id, client_descriptor, True)
-            # If the sender is muted, we don't call `maybe_enqueue_notifications` at all.
-            mock_enqueue.assert_not_called()
+            mock_enqueue.assert_called_once()
+            args_dict = mock_enqueue.call_args_list[0][1]
+
+            assert_maybe_enqueue_notifications_call_args(
+                args_dict=args_dict,
+                message_id=msg_id,
+                private_message=True,
+                sender_is_muted=True,
+                flags=["read"],
+                already_notified={"email_notified": False, "push_notified": False},
+            )
         destroy_event_queue(client_descriptor.event_queue.id)
         result = self.api_delete(user_profile, f"/api/v1/users/me/muted_users/{iago.id}")
         self.assert_json_success(result)
diff --git a/zerver/tests/test_message_edit_notifications.py b/zerver/tests/test_message_edit_notifications.py
index 3c8f024e49..be128099e5 100644
--- a/zerver/tests/test_message_edit_notifications.py
+++ b/zerver/tests/test_message_edit_notifications.py
@@ -105,7 +105,7 @@ class EditMessageSideEffectsTest(ZulipTestCase):
         cordelia_calls = [
             call_args
             for call_args in m.call_args_list
-            if call_args[1]["user_profile_id"] == cordelia.id
+            if call_args[1]["user_data"].user_id == cordelia.id
         ]
 
         if expect_short_circuit:
@@ -184,10 +184,13 @@ class EditMessageSideEffectsTest(ZulipTestCase):
         info = notification_message_data["info"]
 
         cordelia = self.example_user("cordelia")
+        hamlet = self.example_user("hamlet")
         expected_enqueue_kwargs = self.get_maybe_enqueue_notifications_parameters(
-            user_profile_id=cordelia.id,
+            user_id=cordelia.id,
+            sender_id=hamlet.id,
             message_id=message_id,
             mentioned=True,
+            flags=["mentioned"],
             stream_name="Scotland",
             already_notified={},
         )
@@ -295,6 +298,7 @@ class EditMessageSideEffectsTest(ZulipTestCase):
 
     def test_online_push_enabled_for_fully_present_mentioned_user(self) -> None:
         cordelia = self.example_user("cordelia")
+        hamlet = self.example_user("hamlet")
 
         # Simulate Cordelia is FULLY present, not just in term of
         # browser activity, but also in terms of her client descriptors.
@@ -312,10 +316,12 @@ class EditMessageSideEffectsTest(ZulipTestCase):
         info = notification_message_data["info"]
 
         expected_enqueue_kwargs = self.get_maybe_enqueue_notifications_parameters(
-            user_profile_id=cordelia.id,
+            user_id=cordelia.id,
+            sender_id=hamlet.id,
             message_id=message_id,
             mentioned=True,
             stream_name="Scotland",
+            flags=["mentioned"],
             online_push_enabled=True,
             idle=False,
             already_notified={},
@@ -329,6 +335,7 @@ class EditMessageSideEffectsTest(ZulipTestCase):
 
     def test_online_push_enabled_for_fully_present_boring_user(self) -> None:
         cordelia = self.example_user("cordelia")
+        hamlet = self.example_user("hamlet")
 
         # Simulate Cordelia is FULLY present, not just in term of
         # browser activity, but also in terms of her client descriptors.
@@ -346,7 +353,8 @@ class EditMessageSideEffectsTest(ZulipTestCase):
         info = notification_message_data["info"]
 
         expected_enqueue_kwargs = self.get_maybe_enqueue_notifications_parameters(
-            user_profile_id=cordelia.id,
+            user_id=cordelia.id,
+            sender_id=hamlet.id,
             message_id=message_id,
             stream_name="Scotland",
             online_push_enabled=True,
@@ -381,9 +389,11 @@ class EditMessageSideEffectsTest(ZulipTestCase):
         info = notification_message_data["info"]
 
         expected_enqueue_kwargs = self.get_maybe_enqueue_notifications_parameters(
-            user_profile_id=cordelia.id,
+            user_id=cordelia.id,
             message_id=message_id,
+            sender_id=self.example_user("hamlet").id,
             mentioned=True,
+            flags=["mentioned"],
             stream_name="Scotland",
             already_notified={},
         )
@@ -395,6 +405,7 @@ class EditMessageSideEffectsTest(ZulipTestCase):
 
     def test_updates_with_wildcard_mention(self) -> None:
         cordelia = self.example_user("cordelia")
+        hamlet = self.example_user("hamlet")
 
         # We will simulate that the user still has a an active client,
         # but they don't have UserPresence rows, so we will still
@@ -411,9 +422,11 @@ class EditMessageSideEffectsTest(ZulipTestCase):
         info = notification_message_data["info"]
 
         expected_enqueue_kwargs = self.get_maybe_enqueue_notifications_parameters(
-            user_profile_id=cordelia.id,
+            user_id=cordelia.id,
+            sender_id=hamlet.id,
             message_id=message_id,
             wildcard_mention_notify=True,
+            flags=["wildcard_mentioned"],
             stream_name="Scotland",
             already_notified={},
         )
@@ -450,6 +463,7 @@ class EditMessageSideEffectsTest(ZulipTestCase):
 
     def test_updates_with_stream_mention_of_fully_present_user(self) -> None:
         cordelia = self.example_user("cordelia")
+        hamlet = self.example_user("hamlet")
 
         # Simulate Cordelia is FULLY present, not just in term of
         # browser activity, but also in terms of her client descriptors.
@@ -466,9 +480,11 @@ class EditMessageSideEffectsTest(ZulipTestCase):
         info = notification_message_data["info"]
 
         expected_enqueue_kwargs = self.get_maybe_enqueue_notifications_parameters(
-            user_profile_id=cordelia.id,
+            user_id=cordelia.id,
+            sender_id=hamlet.id,
             message_id=message_id,
             mentioned=True,
+            flags=["mentioned"],
             stream_name="Scotland",
             idle=False,
             already_notified={},
diff --git a/zerver/tests/test_muting_users.py b/zerver/tests/test_muting_users.py
index 2ff5c63abf..5ae2c776ef 100644
--- a/zerver/tests/test_muting_users.py
+++ b/zerver/tests/test_muting_users.py
@@ -312,7 +312,7 @@ class MutedUsersTests(ZulipTestCase):
             self.assert_json_success(result)
             m.assert_called_once()
             # `maybe_enqueue_notificaions` was called for Hamlet after message edit mentioned him.
-            self.assertEqual(m.call_args_list[0][1]["user_profile_id"], hamlet.id)
+            self.assertEqual(m.call_args_list[0][1]["user_data"].user_id, hamlet.id)
 
         # Hamlet mutes Cordelia.
         self.login("hamlet")
diff --git a/zerver/tornado/event_queue.py b/zerver/tornado/event_queue.py
index 2c86296d44..77aec17a9b 100644
--- a/zerver/tornado/event_queue.py
+++ b/zerver/tornado/event_queue.py
@@ -737,26 +737,26 @@ def missedmessage_hook(
         if event["type"] != "message":
             continue
         internal_data = event.get("internal_data", {})
+        sender_id = event["message"]["sender_id"]
 
-        sender_is_muted = internal_data.get("sender_is_muted", False)
-        if sender_is_muted:
-            continue
+        user_notifications_data = UserMessageNotificationsData(
+            user_id=user_profile_id,
+            flags=event.get("flags", []),
+            sender_is_muted=internal_data.get("sender_is_muted", False),
+            mentioned=internal_data.get("mentioned", False),
+            stream_push_notify=internal_data.get("stream_push_notify", False),
+            stream_email_notify=internal_data.get("stream_email_notify", False),
+            wildcard_mention_notify=internal_data.get("wildcard_mention_notify", False),
+            # Since one is by definition idle, we don't need to check online_push_enabled
+            online_push_enabled=False,
+        )
 
-        assert "flags" in event
-
-        mentioned = internal_data.get("mentioned", False)
         private_message = event["message"]["type"] == "private"
-        # stream_push_notify is set in process_message_event.
-        stream_push_notify = internal_data.get("stream_push_notify", False)
-        stream_email_notify = internal_data.get("stream_email_notify", False)
-        wildcard_mention_notify = internal_data.get("wildcard_mention_notify", False)
 
         stream_name = None
         if not private_message:
             stream_name = event["message"]["display_recipient"]
 
-        # Since one is by definition idle, we don't need to check online_push_enabled
-        online_push_enabled = False
         # Since we just GC'd the last event queue, the user is definitely idle.
         idle = True
 
@@ -767,15 +767,11 @@ def missedmessage_hook(
             email_notified=internal_data.get("email_notified", False),
         )
         maybe_enqueue_notifications(
-            user_profile_id=user_profile_id,
+            user_data=user_notifications_data,
+            sender_id=sender_id,
             message_id=message_id,
             private_message=private_message,
-            mentioned=mentioned,
-            wildcard_mention_notify=wildcard_mention_notify,
-            stream_push_notify=stream_push_notify,
-            stream_email_notify=stream_email_notify,
             stream_name=stream_name,
-            online_push_enabled=online_push_enabled,
             idle=idle,
             already_notified=already_notified,
         )
@@ -794,15 +790,11 @@ def receiver_is_off_zulip(user_profile_id: int) -> bool:
 
 def maybe_enqueue_notifications(
     *,
-    user_profile_id: int,
+    user_data: UserMessageNotificationsData,
+    sender_id: int,
     message_id: int,
     private_message: bool,
-    mentioned: bool,
-    wildcard_mention_notify: bool,
-    stream_push_notify: bool,
-    stream_email_notify: bool,
     stream_name: Optional[str],
-    online_push_enabled: bool,
     idle: bool,
     already_notified: Dict[str, bool],
 ) -> Dict[str, bool]:
@@ -815,17 +807,15 @@ def maybe_enqueue_notifications(
     """
     notified: Dict[str, bool] = {}
 
-    if (idle or online_push_enabled) and (
-        private_message or mentioned or wildcard_mention_notify or stream_push_notify
-    ):
-        notice = build_offline_notification(user_profile_id, message_id)
+    if user_data.is_push_notifiable(private_message, sender_id, idle):
+        notice = build_offline_notification(user_data.user_id, message_id)
         if private_message:
             notice["trigger"] = "private_message"
-        elif mentioned:
+        elif user_data.mentioned:
             notice["trigger"] = "mentioned"
-        elif wildcard_mention_notify:
+        elif user_data.wildcard_mention_notify:
             notice["trigger"] = "wildcard_mentioned"
-        elif stream_push_notify:
+        elif user_data.stream_push_notify:
             notice["trigger"] = "stream_push_notify"
         else:
             raise AssertionError("Unknown notification trigger!")
@@ -838,15 +828,15 @@ def maybe_enqueue_notifications(
     # mention.  Eventually, we'll add settings to allow email
     # notifications to match the model of push notifications
     # above.
-    if idle and (private_message or mentioned or wildcard_mention_notify or stream_email_notify):
-        notice = build_offline_notification(user_profile_id, message_id)
+    if user_data.is_email_notifiable(private_message, sender_id, idle):
+        notice = build_offline_notification(user_data.user_id, message_id)
         if private_message:
             notice["trigger"] = "private_message"
-        elif mentioned:
+        elif user_data.mentioned:
             notice["trigger"] = "mentioned"
-        elif wildcard_mention_notify:
+        elif user_data.wildcard_mention_notify:
             notice["trigger"] = "wildcard_mentioned"
-        elif stream_email_notify:
+        elif user_data.stream_email_notify:
             notice["trigger"] = "stream_email_notify"
         else:
             raise AssertionError("Unknown notification trigger!")
@@ -966,8 +956,8 @@ def process_message_event(
             muted_sender_user_ids=muted_sender_user_ids,
         )
 
-        # Remove fields sent through other pipes to save some space.
         internal_data = asdict(user_notifications_data)
+        # Remove fields sent through other pipes to save some space.
         internal_data.pop("flags")
         internal_data.pop("user_id")
         extra_user_data[user_profile_id] = dict(internal_data=internal_data)
@@ -985,15 +975,11 @@ def process_message_event(
 
         extra_user_data[user_profile_id]["internal_data"].update(
             maybe_enqueue_notifications(
-                user_profile_id=user_profile_id,
+                user_data=user_notifications_data,
+                sender_id=sender_id,
                 message_id=message_id,
                 private_message=private_message,
-                mentioned=user_notifications_data.mentioned,
-                wildcard_mention_notify=user_notifications_data.wildcard_mention_notify,
-                stream_push_notify=user_notifications_data.stream_push_notify,
-                stream_email_notify=user_notifications_data.stream_email_notify,
                 stream_name=stream_name,
-                online_push_enabled=user_notifications_data.online_push_enabled,
                 idle=idle,
                 already_notified={},
             )
@@ -1126,6 +1112,25 @@ def process_message_update_event(
 
     for user_data in users:
         user_profile_id = user_data["id"]
+
+        if "user_id" in event_template:
+            # This is inaccurate: the user we'll get here will be the
+            # sender if the message's content was edited, which is
+            # typically where we might send new notifications.
+            # However, for topic/stream edits, it could be another
+            # user.  We may need to adjust the format for
+            # update_message events to address this issue.
+            message_sender_id = event_template["user_id"]
+        else:
+            # This is also inaccurate, but usefully so.  Events
+            # without a `user_id` field come from the
+            # do_update_embedded_data code path, and represent not an
+            # update to the raw content, but instead just rendering
+            # previews. Setting the current user at the sender is a
+            # hack to simplify notifications logic for this code
+            # path. TODO: Change this to short-circuit more directly.
+            message_sender_id = user_profile_id
+
         user_event = dict(event_template)  # shallow copy, but deep enough for our needs
         for key in user_data.keys():
             if key != "id":
@@ -1145,6 +1150,7 @@ def process_message_update_event(
         maybe_enqueue_notifications_for_message_update(
             user_data=user_notifications_data,
             message_id=message_id,
+            sender_id=message_sender_id,
             private_message=(stream_name is None),
             stream_name=stream_name,
             presence_idle=(user_profile_id in presence_idle_user_ids),
@@ -1161,6 +1167,7 @@ def process_message_update_event(
 def maybe_enqueue_notifications_for_message_update(
     user_data: UserMessageNotificationsData,
     message_id: int,
+    sender_id: int,
     private_message: bool,
     stream_name: Optional[str],
     presence_idle: bool,
@@ -1202,15 +1209,11 @@ def maybe_enqueue_notifications_for_message_update(
     idle = presence_idle or receiver_is_off_zulip(user_data.user_id)
 
     maybe_enqueue_notifications(
-        user_profile_id=user_data.user_id,
+        user_data=user_data,
         message_id=message_id,
+        sender_id=sender_id,
         private_message=private_message,
-        mentioned=user_data.mentioned,
-        wildcard_mention_notify=user_data.wildcard_mention_notify,
-        stream_push_notify=user_data.stream_push_notify,
-        stream_email_notify=user_data.stream_email_notify,
         stream_name=stream_name,
-        online_push_enabled=user_data.online_push_enabled,
         idle=idle,
         already_notified={},
     )