events: Add functionality to mark messages as unread.

Co-authored-by: Steve Howell <showell@zulip.com>
Co-authored-by: Tim Abbott <tabbott@zulip.com>

This commit adds the backend functionality to
mark messages as unread through update_message_flags
with `unread` flag and `remove` operation.

We also manage incoming events in the webapp.

Tweaked by tabbott to simplify the implementation and add an API
feature level update to the documentation.

This commit was originally drafted by showell, and showell
also finalized the changes.  Many thanks to Suyash here for
the main work here, which was to get all the tests and
documentation work moving forward.

frontend_tests/node_tests/dispatch.js

@@ -874,6 +874,20 @@ run_test("update_message (read)", ({override}) => {
     assert_same(args.message_ids, [999]);
 });
 
+run_test("update_message (unread)", ({override}) => {
+    const event = event_fixtures.update_message_flags__read_remove;
+
+    const stub = make_stub();
+    override(unread_ops, "process_unread_messages_event", stub.f);
+    dispatch(event);
+    assert.equal(stub.num_calls, 1);
+    const {args} = stub.get_args("args");
+    assert.deepEqual(args, {
+        message_ids: event.messages,
+        message_details: event.message_details,
+    });
+});
+
 run_test("update_message (add star)", ({override, override_rewire}) => {
     override_rewire(starred_messages, "rerender_ui", noop);
 

frontend_tests/node_tests/lib/events.js

@@ -69,6 +69,16 @@ exports.test_streams = {
 
 const streams = exports.test_streams;
 
+// TODO: we want to validate this better with check-schema.
+// The data should mostly be representative here, but we don't
+// really exercise it in our tests yet.
+const message_detail = {
+    type: "stream",
+    mentioned: false,
+    sender_id: test_user.id,
+    stream_id: streams.devel.test_id,
+};
+
 exports.test_realm_emojis = {
     101: {
         id: "101",
@@ -674,6 +684,16 @@ exports.fixtures = {
         all: false,
     },
 
+    update_message_flags__read_remove: {
+        type: "update_message_flags",
+        op: "remove",
+        operation: "remove",
+        flag: "read",
+        messages: [888],
+        message_details: {888: message_detail},
+        all: false,
+    },
+
     update_message_flags__starred_add: {
         type: "update_message_flags",
         op: "add",

static/js/message_live_update.js

@@ -3,7 +3,7 @@ import * as message_lists from "./message_lists";
 import * as message_store from "./message_store";
 import * as people from "./people";
 
-function rerender_messages_view() {
+export function rerender_messages_view() {
     for (const list of [message_lists.home, message_list.narrowed]) {
         if (list === undefined) {
             continue;

static/js/people.js

@@ -403,17 +403,21 @@ export function concat_huddle(user_ids, user_id) {
     return sorted_ids.join(",");
 }
 
-export function pm_lookup_key(user_ids_string) {
+export function pm_lookup_key_from_user_ids(user_ids) {
     /*
         The server will sometimes include our own user id
         in keys for PMs, but we only want our user id if
         we sent a message to ourself.
     */
-    let user_ids = split_to_ints(user_ids_string);
     user_ids = sorted_other_user_ids(user_ids);
     return user_ids.join(",");
 }
 
+export function pm_lookup_key(user_ids_string) {
+    const user_ids = split_to_ints(user_ids_string);
+    return pm_lookup_key_from_user_ids(user_ids);
+}
+
 export function all_user_ids_in_pm(message) {
     if (message.type !== "private") {
         return undefined;

static/js/server_events_dispatch.js

@@ -719,7 +719,14 @@ export function dispatch_normal_event(event) {
                     }
                     break;
                 case "read":
-                    unread_ops.process_read_messages_event(event.messages);
+                    if (event.op === "add") {
+                        unread_ops.process_read_messages_event(event.messages);
+                    } else {
+                        unread_ops.process_unread_messages_event({
+                            message_ids: event.messages,
+                            message_details: event.message_details,
+                        });
+                    }
                     break;
             }
             break;

static/js/unread.js

@@ -450,7 +450,7 @@ export function process_loaded_messages(messages) {
     }
 }
 
-function process_unread_message(message) {
+export function process_unread_message(message) {
     // The `message` here just needs to require certain fields. For example,
     // the "message" may actually be constructed from a Zulip event that doesn't
     // include fields like "content".  The caller must verify that the message

static/js/unread_ops.js

@@ -2,9 +2,11 @@ import * as channel from "./channel";
 import * as message_flags from "./message_flags";
 import * as message_list from "./message_list";
 import * as message_lists from "./message_lists";
+import * as message_live_update from "./message_live_update";
 import * as message_store from "./message_store";
 import * as message_viewport from "./message_viewport";
 import * as notifications from "./notifications";
+import * as people from "./people";
 import * as recent_topics_ui from "./recent_topics_ui";
 import * as reload from "./reload";
 import * as unread from "./unread";
@@ -73,6 +75,60 @@ export function process_read_messages_event(message_ids) {
     unread_ui.update_unread_counts();
 }
 
+export function process_unread_messages_event({message_ids, message_details}) {
+    // This is the reverse of  process_unread_messages_event.
+    message_ids = unread.get_read_message_ids(message_ids);
+    if (message_ids.length === 0) {
+        return;
+    }
+
+    for (const message_id of message_ids) {
+        const message = message_store.get(message_id);
+
+        if (message) {
+            message.unread = true;
+        }
+
+        const message_info = message_details[message_id];
+
+        let user_ids_string;
+
+        if (message_info.type === "private") {
+            user_ids_string = people.pm_lookup_key_from_user_ids(message_info.user_ids);
+        }
+
+        unread.process_unread_message({
+            id: message_id,
+            mentioned: message_info.mentioned,
+            stream_id: message_info.stream_id,
+            topic: message_info.topic,
+            type: message_info.type,
+            unread: true,
+            user_ids_string,
+        });
+
+        if (message_info.type === "stream") {
+            recent_topics_ui.update_topic_unread_count({
+                stream_id: message_info.stream_id,
+                topic: message_info.topic,
+            });
+        }
+    }
+
+    /*
+        We use a big-hammer approach now to updating the message view.
+        This is relatively harmless, since the only time we are called is
+        when the user herself marks her message as unread.  But we
+        do eventually want to be more surgical here, especially once we
+        have a final scheme for how best to structure the HTML within
+        the message to indicate read-vs.-unread.  Currently we use a
+        green border, but that may change.
+    */
+    message_live_update.rerender_messages_view();
+
+    unread_ui.update_unread_counts();
+}
+
 // Takes a list of messages and marks them as read.
 // Skips any messages that are already marked as read.
 export function notify_server_messages_read(messages, options = {}) {

templates/zerver/api/changelog.md

@@ -20,6 +20,12 @@ format used by the Zulip server that they are interacting with.
 
 ## Changes in Zulip 5.0
 
+**Feature level 121**
+
+* [`GET /events`](/api/get-events): Added `message_details` field
+  appearing in message flag update events when marking previously read
+  messages as unread.
+
 **Feature level 120**
 
 * [`GET /messages/{message_id}`](/api/get-message): This endpoint

version.py

@@ -33,7 +33,7 @@ DESKTOP_WARNING_VERSION = "5.4.3"
 # Changes should be accompanied by documentation explaining what the
 # new level means in templates/zerver/api/changelog.md, as well as
 # "**Changes**" entries in the endpoint's documentation in `zulip.yaml`.
-API_FEATURE_LEVEL = 120
+API_FEATURE_LEVEL = 121
 
 # Bump the minor PROVISION_VERSION to indicate that folks should provision
 # only when going from an old version of the code to a newer version. Bump

zerver/lib/actions.py

@@ -106,7 +106,9 @@ from zerver.lib.message import (
     SendMessageRequest,
     access_message,
     bulk_access_messages,
+    format_unread_message_details,
     get_last_message_id,
+    get_raw_unread_data,
     normalize_body,
     render_markdown,
     truncate_topic,
@@ -6445,6 +6447,15 @@ def do_update_message_flags(
         "messages": messages,
         "all": False,
     }
+
+    if flag == "read" and operation == "remove":
+        # When removing the read flag (i.e. marking messages as
+        # unread), extend the event with an additional object with
+        # details on the messages required to update the client's
+        # `unread_msgs` data structure.
+        raw_unread_data = get_raw_unread_data(user_profile, messages)
+        event["message_details"] = format_unread_message_details(user_profile.id, raw_unread_data)
+
     send_event(user_profile.realm, event, [user_profile.id])
 
     if flag == "read" and operation == "add":
@@ -6461,6 +6472,7 @@ def do_update_message_flags(
             event_time,
             increment=min(1, count),
         )
+
     return count
 
 

zerver/lib/event_schema.py

@@ -1644,10 +1644,29 @@ update_message_flags_remove_event = event_dict_type(
         ("type", Equals("update_message_flags")),
         ("op", Equals("remove")),
         ("operation", Equals("remove")),
-        ("flag", str),
+        ("flag", EnumType(["read", "starred"])),
         ("messages", ListType(int)),
         ("all", bool),
-    ]
+    ],
+    optional_keys=[
+        (
+            "message_details",
+            StringDictType(
+                DictType(
+                    required_keys=[
+                        ("type", EnumType(["private", "stream"])),
+                    ],
+                    optional_keys=[
+                        ("mentioned", bool),
+                        ("user_ids", ListType(int)),
+                        ("stream_id", int),
+                        ("topic", str),
+                        ("unmuted_stream_msg", bool),
+                    ],
+                )
+            ),
+        )
+    ],
 )
 check_update_message_flags_remove = make_checker(update_message_flags_remove_event)
 

zerver/lib/events.py

@@ -27,6 +27,7 @@ from zerver.lib.external_accounts import DEFAULT_EXTERNAL_ACCOUNTS
 from zerver.lib.hotspots import get_next_hotspots
 from zerver.lib.integrations import EMBEDDED_BOTS, WEBHOOK_INTEGRATIONS
 from zerver.lib.message import (
+    add_message_to_unread_msgs,
     aggregate_unread_data,
     apply_unread_message_event,
     extract_unread_data_from_um_rows,
@@ -1157,6 +1158,14 @@ def apply_event(
         if "raw_unread_msgs" in state and event["flag"] == "read" and event["op"] == "add":
             for remove_id in event["messages"]:
                 remove_message_id_from_unread_mgs(state["raw_unread_msgs"], remove_id)
+        if event["flag"] == "read" and event["op"] == "remove":
+            for message_id_str, message_details in event["message_details"].items():
+                add_message_to_unread_msgs(
+                    user_profile.id,
+                    state["raw_unread_msgs"],
+                    int(message_id_str),
+                    message_details,
+                )
         if event["flag"] == "starred" and "starred_messages" in state:
             if event["op"] == "add":
                 state["starred_messages"] += event["messages"]

zerver/lib/message.py

@@ -56,6 +56,15 @@ from zerver.models import (
 )
 
 
+class MessageDetailsDict(TypedDict, total=False):
+    type: str
+    mentioned: bool
+    user_ids: List[int]
+    stream_id: int
+    topic: str
+    unmuted_stream_msg: bool
+
+
 class RawReactionRow(TypedDict):
     emoji_code: str
     emoji_name: str
@@ -976,7 +985,9 @@ def get_starred_message_ids(user_profile: UserProfile) -> List[int]:
     )
 
 
-def get_raw_unread_data(user_profile: UserProfile) -> RawUnreadMessagesResult:
+def get_raw_unread_data(
+    user_profile: UserProfile, message_ids: Optional[List[int]] = None
+) -> RawUnreadMessagesResult:
     excluded_recipient_ids = get_inactive_recipient_ids(user_profile)
 
     user_msgs = (
@@ -986,9 +997,6 @@ def get_raw_unread_data(user_profile: UserProfile) -> RawUnreadMessagesResult:
         .exclude(
             message__recipient_id__in=excluded_recipient_ids,
         )
-        .extra(
-            where=[UserMessage.where_unread()],
-        )
         .values(
             "message_id",
             "message__sender_id",
@@ -1001,6 +1009,16 @@ def get_raw_unread_data(user_profile: UserProfile) -> RawUnreadMessagesResult:
         .order_by("-message_id")
     )
 
+    if message_ids is not None:
+        # When users are marking just a few messages as unread, we just need
+        # those ids, and we know they're unread.
+        user_msgs = user_msgs.filter(message_id__in=message_ids)
+    else:
+        # At page load we need all unread messages.
+        user_msgs = user_msgs.extra(
+            where=[UserMessage.where_unread()],
+        )
+
     # Limit unread messages for performance reasons.
     user_msgs = list(user_msgs[:MAX_UNREAD_MESSAGES])
 
@@ -1283,6 +1301,98 @@ def remove_message_id_from_unread_mgs(state: RawUnreadMessagesResult, message_id
     state["mentions"].discard(message_id)
 
 
+def format_unread_message_details(
+    my_user_id: int,
+    raw_unread_data: RawUnreadMessagesResult,
+) -> Dict[str, MessageDetailsDict]:
+    unread_data = {}
+
+    for message_id, private_message_details in raw_unread_data["pm_dict"].items():
+        other_user_id = private_message_details["other_user_id"]
+        if other_user_id == my_user_id:
+            user_ids = []
+        else:
+            user_ids = [other_user_id]
+
+        # Note that user_ids excludes ourself, even for the case we send messages
+        # to ourself.
+        message_details = MessageDetailsDict(
+            type="private",
+            user_ids=user_ids,
+        )
+        if message_id in raw_unread_data["mentions"]:
+            message_details["mentioned"] = True
+        unread_data[str(message_id)] = message_details
+
+    for message_id, stream_message_details in raw_unread_data["stream_dict"].items():
+        if message_id in raw_unread_data["unmuted_stream_msgs"]:
+            unmuted_stream_msg = True
+        else:
+            unmuted_stream_msg = False
+
+        message_details = MessageDetailsDict(
+            type="stream",
+            stream_id=stream_message_details["stream_id"],
+            topic=stream_message_details["topic"],
+            # Clients don't need this detail, but we need it internally for apply_events.
+            unmuted_stream_msg=unmuted_stream_msg,
+        )
+        if message_id in raw_unread_data["mentions"]:
+            message_details["mentioned"] = True
+        unread_data[str(message_id)] = message_details
+
+    for message_id, huddle_message_details in raw_unread_data["huddle_dict"].items():
+        # The client wants a list of user_ids in the conversation, excluding ourself,
+        # that is sorted in numerical order.
+        user_ids = [int(s) for s in huddle_message_details["user_ids_string"].split(",")]
+        user_ids = [user_id for user_id in user_ids if user_id != my_user_id]
+        user_ids.sort()
+        message_details = MessageDetailsDict(
+            type="private",
+            user_ids=user_ids,
+        )
+        if message_id in raw_unread_data["mentions"]:
+            message_details["mentioned"] = True
+        unread_data[str(message_id)] = message_details
+
+    return unread_data
+
+
+def add_message_to_unread_msgs(
+    my_user_id: int,
+    state: RawUnreadMessagesResult,
+    message_id: int,
+    message_details: MessageDetailsDict,
+) -> None:
+    if message_details.get("mentioned"):
+        state["mentions"].add(message_id)
+
+    if message_details["type"] == "private":
+        user_ids: List[int] = message_details["user_ids"]
+        user_ids = [user_id for user_id in user_ids if user_id != my_user_id]
+        if user_ids == []:
+            state["pm_dict"][message_id] = RawUnreadPrivateMessageDict(
+                other_user_id=my_user_id,
+            )
+        elif len(user_ids) == 1:
+            state["pm_dict"][message_id] = RawUnreadPrivateMessageDict(
+                other_user_id=user_ids[0],
+            )
+        else:
+            user_ids.append(my_user_id)
+            user_ids_string = ",".join(str(user_id) for user_id in sorted(user_ids))
+            state["huddle_dict"][message_id] = RawUnreadHuddleDict(
+                user_ids_string=user_ids_string,
+            )
+    elif message_details["type"] == "stream":
+        state["stream_dict"][message_id] = RawUnreadStreamDict(
+            stream_id=message_details["stream_id"],
+            topic=message_details["topic"],
+        )
+        if message_details["unmuted_stream_msg"]:
+            state["unmuted_stream_msgs"].add(message_id)
+
+
 def estimate_recent_messages(realm: Realm, hours: int) -> int:
     stat = COUNT_STATS["messages_sent:is_bot:hour"]
     d = timezone_now() - datetime.timedelta(hours=hours)

zerver/openapi/zulip.yaml

@@ -2541,6 +2541,16 @@ paths:
                                 removed from a message.
 
                                 [message-flags]: /api/update-message-flags#available-flags
+                              required:
+                                [
+                                  "id",
+                                  "type",
+                                  "op",
+                                  "operation",
+                                  "flag",
+                                  "messages",
+                                  "all",
+                                ]
                               properties:
                                 id:
                                   $ref: "#/components/schemas/EventIdSchema"
@@ -2567,6 +2577,9 @@ paths:
                                   type: string
                                   description: |
                                     The flag to be removed.
+                                  enum:
+                                    - starred
+                                    - read
                                 messages:
                                   type: array
                                   description: |
@@ -2579,6 +2592,70 @@ paths:
                                   description: |
                                     Whether the flag was removed from all messages.
                                     If this is true then the `messages` array will be empty.
+                                message_details:
+                                  description: |
+                                    Present if `message` and `update_message_flags` are both present in
+                                    `event_types` and the `flag` is `read` and the `op` is `remove`.
+
+                                    A set of data structures describing the messages that
+                                    are being marked as unread with additional details to
+                                    allow a client to update the `unread_msgs` data
+                                    structure for these messages (which may not be
+                                    otherwise known to the client).
+
+                                    **Changes**: New in Zulip 5.0 (feature level 121). Previously,
+                                    marking already read messages as unread was not
+                                    supported by the Zulip API.
+                                  type: object
+                                  additionalProperties:
+                                    type: object
+                                    description: |
+                                      Additional properties.
+                                    additionalProperties: false
+                                    required: ["type"]
+                                    properties:
+                                      type:
+                                        type: string
+                                        description: |
+                                          The type of this message.
+                                        enum:
+                                          - private
+                                          - stream
+                                      mentioned:
+                                        type: boolean
+                                        description: |
+                                          A flag which indicates whether the message contains a mention
+                                          of the user.
+
+                                          Present only if the message mentions the current user.
+                                      user_ids:
+                                        type: array
+                                        items:
+                                          type: integer
+                                        description: |
+                                          Present only if `type` is `private`.
+
+                                          The user IDs of every recipient of this private message, excluding yourself.
+                                          Will be the empty list for a message you had sent to only yourself.
+                                      stream_id:
+                                        type: integer
+                                        description: |
+                                          Present only if `type` is `stream`.
+
+                                          The ID of the stream where the message was sent.
+                                      topic:
+                                        type: string
+                                        description: |
+                                          Present only if `type` is `stream`.
+
+                                          Name of the topic where the message was sent.
+                                      unmuted_stream_msg:
+                                        type: boolean
+                                        deprecated: true
+                                        description: |
+                                          **Deprecated**
+                                          Internal implementation detail. Clients should
+                                          ignore this field as it will be removed in the future.
                               example:
                                 {
                                   "type": "update_message_flags",
@@ -2586,6 +2663,15 @@ paths:
                                   "operation": "remove",
                                   "flag": "starred",
                                   "messages": [63],
+                                  "message_details":
+                                    {
+                                      63:
+                                        {
+                                          "type": "stream",
+                                          "stream_id": 22,
+                                          "topic": "lunch",
+                                        },
+                                    },
                                   "all": false,
                                   "id": 0,
                                 }
@@ -9567,8 +9653,9 @@ paths:
                                 user_ids_string:
                                   type: string
                                   description: |
-                                    A string containing the ids of all users in the huddle(group PMs)
-                                    separated by commas(,). Example: "1,2,3".
+                                    A string containing the IDs of all users in the group
+                                    private message conversation separated by commas
+                                    (,). Example: "1,2,3".
                                 message_ids:
                                   type: array
                                   description: |

zerver/tests/test_events.py

@@ -656,6 +656,43 @@ class NormalActionsTest(BaseAction):
                 state_change_expected=True,
             )
 
+            events = self.verify_action(
+                lambda: do_update_message_flags(user_profile, "remove", "read", [message]),
+                state_change_expected=True,
+            )
+            check_update_message_flags_remove("events[0]", events[0])
+
+            personal_message = self.send_personal_message(
+                from_user=user_profile, to_user=self.example_user("cordelia"), content=content
+            )
+            self.verify_action(
+                lambda: do_update_message_flags(user_profile, "add", "read", [personal_message]),
+                state_change_expected=True,
+            )
+
+            events = self.verify_action(
+                lambda: do_update_message_flags(user_profile, "remove", "read", [personal_message]),
+                state_change_expected=True,
+            )
+            check_update_message_flags_remove("events[0]", events[0])
+
+            huddle_message = self.send_huddle_message(
+                from_user=self.example_user("cordelia"),
+                to_users=[user_profile, self.example_user("othello")],
+                content=content,
+            )
+
+            self.verify_action(
+                lambda: do_update_message_flags(user_profile, "add", "read", [huddle_message]),
+                state_change_expected=True,
+            )
+
+            events = self.verify_action(
+                lambda: do_update_message_flags(user_profile, "remove", "read", [huddle_message]),
+                state_change_expected=True,
+            )
+            check_update_message_flags_remove("events[0]", events[0])
+
     def test_send_message_to_existing_recipient(self) -> None:
         sender = self.example_user("cordelia")
         self.send_stream_message(

zerver/tests/test_message_flags.py

@@ -5,14 +5,19 @@ import orjson
 from django.db import connection
 from django.http import HttpResponse
 
-from zerver.lib.actions import do_change_stream_permission
+from zerver.lib.actions import do_change_stream_permission, do_update_message_flags
 from zerver.lib.fix_unreads import fix, fix_unsubscribed
 from zerver.lib.message import (
+    MessageDetailsDict,
     MessageDict,
+    RawUnreadMessagesResult,
+    RawUnreadPrivateMessageDict,
     UnreadMessagesResult,
+    add_message_to_unread_msgs,
     aggregate_unread_data,
     apply_unread_message_event,
     bulk_access_messages,
+    format_unread_message_details,
     get_raw_unread_data,
 )
 from zerver.lib.test_classes import ZulipTestCase
@@ -1395,3 +1400,595 @@ class PersonalMessagesFlagTest(ZulipTestCase):
 
         for msg in self.get_messages():
             self.assertNotIn("is_private", msg["flags"])
+
+
+class MarkUnreadTest(ZulipTestCase):
+    def mute_stream(self, stream_name: str, user: int) -> None:
+        realm = get_realm("zulip")
+        stream = get_stream(stream_name, realm)
+        recipient = stream.recipient
+        subscription = Subscription.objects.get(
+            user_profile=user,
+            recipient=recipient,
+        )
+        subscription.is_muted = True
+        subscription.save()
+
+    def test_missing_usermessage_record(self) -> None:
+        cordelia = self.example_user("cordelia")
+        othello = self.example_user("othello")
+
+        stream_name = "Some new stream"
+        self.subscribe(cordelia, stream_name)
+
+        message_id1 = self.send_stream_message(
+            sender=cordelia,
+            stream_name=stream_name,
+            topic_name="lunch",
+            content="whatever",
+        )
+
+        self.subscribe(othello, stream_name)
+
+        raw_unread_data = get_raw_unread_data(
+            user_profile=othello,
+        )
+
+        self.assertEqual(raw_unread_data["stream_dict"], {})
+
+        message_id2 = self.send_stream_message(
+            sender=cordelia,
+            stream_name=stream_name,
+            topic_name="lunch",
+            content="whatever",
+        )
+
+        raw_unread_data = get_raw_unread_data(
+            user_profile=othello,
+        )
+
+        self.assertEqual(raw_unread_data["stream_dict"].keys(), {message_id2})
+
+        do_update_message_flags(othello, "remove", "read", [message_id1])
+
+        raw_unread_data = get_raw_unread_data(
+            user_profile=othello,
+        )
+
+        self.assertEqual(raw_unread_data["stream_dict"].keys(), {message_id1, message_id2})
+
+    def test_format_unread_message_details(self) -> None:
+        user = self.example_user("cordelia")
+        message_id = 999
+
+        # send message to self
+        pm_dict = {
+            message_id: RawUnreadPrivateMessageDict(other_user_id=user.id),
+        }
+
+        raw_unread_data = RawUnreadMessagesResult(
+            pm_dict=pm_dict,
+            stream_dict={},
+            huddle_dict={},
+            mentions=set(),
+            muted_stream_ids=[],
+            unmuted_stream_msgs=set(),
+            old_unreads_missing=False,
+        )
+
+        message_details = format_unread_message_details(user.id, raw_unread_data)
+        self.assertEqual(
+            message_details,
+            {
+                str(message_id): dict(type="private", user_ids=[]),
+            },
+        )
+
+    def test_add_message_to_unread_msgs(self) -> None:
+        user = self.example_user("cordelia")
+        message_id = 999
+
+        raw_unread_data = RawUnreadMessagesResult(
+            pm_dict={},
+            stream_dict={},
+            huddle_dict={},
+            mentions=set(),
+            muted_stream_ids=[],
+            unmuted_stream_msgs=set(),
+            old_unreads_missing=False,
+        )
+
+        # message to self
+        message_details = MessageDetailsDict(type="private", user_ids=[])
+        add_message_to_unread_msgs(user.id, raw_unread_data, message_id, message_details)
+        self.assertEqual(
+            raw_unread_data["pm_dict"],
+            {message_id: RawUnreadPrivateMessageDict(other_user_id=user.id)},
+        )
+
+    def test_stream_messages_unread(self) -> None:
+        sender = self.example_user("cordelia")
+        receiver = self.example_user("hamlet")
+        stream_name = "Denmark"
+        stream = self.subscribe(receiver, stream_name)
+        self.subscribe(sender, stream_name)
+        topic_name = "test"
+        message_ids = [
+            self.send_stream_message(
+                sender=sender,
+                stream_name=stream_name,
+                topic_name=topic_name,
+            )
+            for i in range(4)
+        ]
+        self.login("hamlet")
+        result = self.client_post(
+            "/json/messages/flags",
+            {"messages": orjson.dumps(message_ids).decode(), "op": "add", "flag": "read"},
+        )
+        self.assert_json_success(result)
+        for message_id in message_ids:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertTrue(um.flags.read)
+        messages_to_unread = message_ids[2:]
+        messages_still_read = message_ids[:2]
+
+        params = {
+            "messages": orjson.dumps(messages_to_unread).decode(),
+            "op": "remove",
+            "flag": "read",
+        }
+
+        events: List[Mapping[str, Any]] = []
+
+        # Use the tornado_redirected_to_list context manager to capture
+        # events.
+        with self.tornado_redirected_to_list(events, expected_num_events=1):
+            result = self.api_post(receiver, "/api/v1/messages/flags", params)
+
+        self.assert_json_success(result)
+        event = events[0]["event"]
+        self.assertEqual(event["messages"], messages_to_unread)
+        unread_message_ids = set(str(message_id) for message_id in messages_to_unread)
+        self.assertSetEqual(set(event["message_details"].keys()), unread_message_ids)
+        for message_id in event["message_details"]:
+            self.assertEqual(
+                event["message_details"][message_id],
+                dict(
+                    type="stream",
+                    topic="test",
+                    unmuted_stream_msg=True,
+                    stream_id=stream.id,
+                ),
+            )
+
+        for message_id in messages_to_unread:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertFalse(um.flags.read)
+        for message_id in messages_still_read:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertTrue(um.flags.read)
+
+    def test_stream_messages_unread_muted(self) -> None:
+        sender = self.example_user("cordelia")
+        receiver = self.example_user("hamlet")
+        stream_name = "Denmark"
+        stream = self.subscribe(receiver, stream_name)
+        self.subscribe(sender, stream_name)
+        topic_name = "test"
+        message_ids = [
+            self.send_stream_message(
+                sender=sender,
+                stream_name=stream_name,
+                topic_name=topic_name,
+            )
+            for i in range(4)
+        ]
+        self.mute_stream(stream_name, receiver)
+        self.login("hamlet")
+        result = self.client_post(
+            "/json/messages/flags",
+            {"messages": orjson.dumps(message_ids).decode(), "op": "add", "flag": "read"},
+        )
+        self.assert_json_success(result)
+        for message_id in message_ids:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertTrue(um.flags.read)
+        messages_to_unread = message_ids[2:]
+        messages_still_read = message_ids[:2]
+
+        params = {
+            "messages": orjson.dumps(messages_to_unread).decode(),
+            "op": "remove",
+            "flag": "read",
+        }
+
+        events: List[Mapping[str, Any]] = []
+
+        # Use the tornado_redirected_to_list context manager to capture
+        # events.
+        with self.tornado_redirected_to_list(events, expected_num_events=1):
+            result = self.api_post(receiver, "/api/v1/messages/flags", params)
+
+        self.assert_json_success(result)
+        event = events[0]["event"]
+        self.assertEqual(event["messages"], messages_to_unread)
+        unread_message_ids = set(str(message_id) for message_id in messages_to_unread)
+        self.assertSetEqual(set(event["message_details"].keys()), unread_message_ids)
+        for message_id in event["message_details"]:
+            self.assertEqual(
+                event["message_details"][message_id],
+                dict(
+                    type="stream",
+                    topic="test",
+                    unmuted_stream_msg=False,
+                    stream_id=stream.id,
+                ),
+            )
+
+        for message_id in messages_to_unread:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertFalse(um.flags.read)
+        for message_id in messages_still_read:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertTrue(um.flags.read)
+
+    def test_stream_messages_unread_mention(self) -> None:
+        sender = self.example_user("cordelia")
+        receiver = self.example_user("hamlet")
+        stream_name = "Denmark"
+        stream = self.subscribe(receiver, stream_name)
+        self.subscribe(sender, stream_name)
+        topic_name = "test"
+        message_ids = [
+            self.send_stream_message(
+                sender=sender,
+                stream_name=stream_name,
+                topic_name=topic_name,
+                content="@**King Hamlet**",
+            )
+            for i in range(4)
+        ]
+        self.login("hamlet")
+        result = self.client_post(
+            "/json/messages/flags",
+            {"messages": orjson.dumps(message_ids).decode(), "op": "add", "flag": "read"},
+        )
+        self.assert_json_success(result)
+        for message_id in message_ids:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertTrue(um.flags.read)
+        messages_to_unread = message_ids[2:]
+        messages_still_read = message_ids[:2]
+
+        params = {
+            "messages": orjson.dumps(messages_to_unread).decode(),
+            "op": "remove",
+            "flag": "read",
+        }
+
+        events: List[Mapping[str, Any]] = []
+
+        # Use the tornado_redirected_to_list context manager to capture
+        # events.
+        with self.tornado_redirected_to_list(events, expected_num_events=1):
+            result = self.api_post(receiver, "/api/v1/messages/flags", params)
+
+        self.assert_json_success(result)
+        event = events[0]["event"]
+        self.assertEqual(event["messages"], messages_to_unread)
+        unread_message_ids = set(str(message_id) for message_id in messages_to_unread)
+        self.assertSetEqual(set(event["message_details"].keys()), unread_message_ids)
+        for message_id in event["message_details"]:
+            self.assertEqual(
+                event["message_details"][message_id],
+                dict(
+                    type="stream",
+                    mentioned=True,
+                    topic="test",
+                    unmuted_stream_msg=True,
+                    stream_id=stream.id,
+                ),
+            )
+
+        for message_id in messages_to_unread:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertFalse(um.flags.read)
+        for message_id in messages_still_read:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertTrue(um.flags.read)
+
+    def test_pm_messages_unread(self) -> None:
+        sender = self.example_user("cordelia")
+        receiver = self.example_user("hamlet")
+        message_ids = [
+            self.send_personal_message(sender, receiver, content="Hello") for i in range(4)
+        ]
+        self.login("hamlet")
+        for message_id in message_ids:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertFalse(um.flags.read)
+        result = self.client_post(
+            "/json/messages/flags",
+            {"messages": orjson.dumps(message_ids).decode(), "op": "add", "flag": "read"},
+        )
+        self.assert_json_success(result)
+        for message_id in message_ids:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertTrue(um.flags.read)
+        messages_to_unread = message_ids[2:]
+        messages_still_read = message_ids[:2]
+
+        params = {
+            "messages": orjson.dumps(messages_to_unread).decode(),
+            "op": "remove",
+            "flag": "read",
+        }
+
+        events: List[Mapping[str, Any]] = []
+
+        # Use the tornado_redirected_to_list context manager to capture
+        # events.
+        with self.tornado_redirected_to_list(events, expected_num_events=1):
+            result = self.api_post(receiver, "/api/v1/messages/flags", params)
+
+        self.assert_json_success(result)
+        event = events[0]["event"]
+        self.assertEqual(event["messages"], messages_to_unread)
+        unread_message_ids = set(str(message_id) for message_id in messages_to_unread)
+        self.assertSetEqual(set(event["message_details"].keys()), unread_message_ids)
+        for message_id in event["message_details"]:
+            self.assertEqual(
+                event["message_details"][message_id],
+                dict(
+                    type="private",
+                    user_ids=[sender.id],
+                ),
+            )
+
+        for message_id in messages_to_unread:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertFalse(um.flags.read)
+        for message_id in messages_still_read:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertTrue(um.flags.read)
+
+    def test_pm_messages_unread_mention(self) -> None:
+        sender = self.example_user("cordelia")
+        receiver = self.example_user("hamlet")
+        stream_name = "Denmark"
+        self.subscribe(receiver, stream_name)
+        message_ids = [
+            self.send_personal_message(sender, receiver, content="@**King Hamlet**")
+            for i in range(4)
+        ]
+        self.login("hamlet")
+        for message_id in message_ids:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertFalse(um.flags.read)
+        result = self.client_post(
+            "/json/messages/flags",
+            {"messages": orjson.dumps(message_ids).decode(), "op": "add", "flag": "read"},
+        )
+        self.assert_json_success(result)
+        for message_id in message_ids:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertTrue(um.flags.read)
+        messages_to_unread = message_ids[2:]
+        messages_still_read = message_ids[:2]
+
+        params = {
+            "messages": orjson.dumps(messages_to_unread).decode(),
+            "op": "remove",
+            "flag": "read",
+        }
+
+        events: List[Mapping[str, Any]] = []
+
+        # Use the tornado_redirected_to_list context manager to capture
+        # events.
+        with self.tornado_redirected_to_list(events, expected_num_events=1):
+            result = self.api_post(receiver, "/api/v1/messages/flags", params)
+
+        self.assert_json_success(result)
+        event = events[0]["event"]
+        self.assertEqual(event["messages"], messages_to_unread)
+        unread_message_ids = set(str(message_id) for message_id in messages_to_unread)
+        self.assertSetEqual(set(event["message_details"].keys()), unread_message_ids)
+        for message_id in event["message_details"]:
+            self.assertEqual(
+                event["message_details"][message_id],
+                dict(
+                    type="private",
+                    user_ids=[sender.id],
+                    mentioned=True,
+                ),
+            )
+
+        for message_id in messages_to_unread:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertFalse(um.flags.read)
+        for message_id in messages_still_read:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertTrue(um.flags.read)
+
+    def test_huddle_messages_unread(self) -> None:
+        sender = self.example_user("cordelia")
+        receiver = self.example_user("hamlet")
+        user1 = self.example_user("othello")
+        message_ids = [
+            # self.send_huddle_message(sender, receiver, content="Hello") for i in range(4)
+            self.send_huddle_message(sender, [receiver, user1])
+            for i in range(4)
+        ]
+        self.login("hamlet")
+        for message_id in message_ids:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertFalse(um.flags.read)
+        result = self.client_post(
+            "/json/messages/flags",
+            {"messages": orjson.dumps(message_ids).decode(), "op": "add", "flag": "read"},
+        )
+        self.assert_json_success(result)
+        for message_id in message_ids:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertTrue(um.flags.read)
+        messages_to_unread = message_ids[2:]
+        messages_still_read = message_ids[:2]
+
+        params = {
+            "messages": orjson.dumps(messages_to_unread).decode(),
+            "op": "remove",
+            "flag": "read",
+        }
+
+        events: List[Mapping[str, Any]] = []
+
+        # Use the tornado_redirected_to_list context manager to capture
+        # events.
+        with self.tornado_redirected_to_list(events, expected_num_events=1):
+            result = self.api_post(receiver, "/api/v1/messages/flags", params)
+
+        self.assert_json_success(result)
+        event = events[0]["event"]
+        self.assertEqual(event["messages"], messages_to_unread)
+        unread_message_ids = set(str(message_id) for message_id in messages_to_unread)
+        self.assertSetEqual(set(event["message_details"].keys()), unread_message_ids)
+        for message_id in event["message_details"]:
+            self.assertNotIn("mentioned", event["message_details"][message_id]),
+
+        for message_id in messages_to_unread:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertFalse(um.flags.read)
+        for message_id in messages_still_read:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertTrue(um.flags.read)
+
+    def test_huddle_messages_unread_mention(self) -> None:
+        sender = self.example_user("cordelia")
+        receiver = self.example_user("hamlet")
+        user1 = self.example_user("othello")
+        message_ids = [
+            # self.send_huddle_message(sender, receiver, content="Hello") for i in range(4)
+            self.send_huddle_message(
+                from_user=sender, to_users=[receiver, user1], content="@**King Hamlet**"
+            )
+            for i in range(4)
+        ]
+        self.login("hamlet")
+        for message_id in message_ids:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertFalse(um.flags.read)
+        result = self.client_post(
+            "/json/messages/flags",
+            {"messages": orjson.dumps(message_ids).decode(), "op": "add", "flag": "read"},
+        )
+        self.assert_json_success(result)
+        for message_id in message_ids:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertTrue(um.flags.read)
+        messages_to_unread = message_ids[2:]
+        messages_still_read = message_ids[:2]
+
+        params = {
+            "messages": orjson.dumps(messages_to_unread).decode(),
+            "op": "remove",
+            "flag": "read",
+        }
+
+        events: List[Mapping[str, Any]] = []
+
+        # Use the tornado_redirected_to_list context manager to capture
+        # events.
+        with self.tornado_redirected_to_list(events, expected_num_events=1):
+            result = self.api_post(receiver, "/api/v1/messages/flags", params)
+
+        self.assert_json_success(result)
+        event = events[0]["event"]
+        self.assertEqual(event["messages"], messages_to_unread)
+        unread_message_ids = set(str(message_id) for message_id in messages_to_unread)
+        self.assertSetEqual(set(event["message_details"].keys()), unread_message_ids)
+        for message_id in event["message_details"]:
+            self.assertEqual(event["message_details"][message_id]["mentioned"], True),
+
+        for message_id in messages_to_unread:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertFalse(um.flags.read)
+        for message_id in messages_still_read:
+            um = UserMessage.objects.get(
+                user_profile_id=receiver.id,
+                message_id=message_id,
+            )
+            self.assertTrue(um.flags.read)