user_topics: Update 'do_update_message' to handle 'merge userTopic states'.

This commit updates the logic for migrating user_topic rows
during the move-messages operation when the target topic
already has messages.

Previously, the target_topic's visibility_policy was simply
set to the original_topic's visibility_policy,
and the original_topic's visibility_policy was set to INHERIT.

This commit updates the move-messages code path to determine
the new visibility_policy depending on the visibility policies
of the original and target topics.
The target_topic's visibility_policy is then updated.

The number of db queries has increased by two:
One query corresponds to determining if 'target_topic_has_messages'.
Another query corresponds to 'get_users_with_user_topic_visibility_policy'
to determine 'target_topic_user_profile_to_visibility_policy'.

zerver/actions/message_edit.py

@@ -1,6 +1,7 @@
 import datetime
+import itertools
 from collections import defaultdict
-from typing import Any, Dict, Iterable, List, Optional, Set
+from typing import Any, Dict, Iterable, List, Optional, Set, Tuple
 
 from django.conf import settings
 from django.db import transaction
@@ -324,6 +325,27 @@ def do_update_embedded_data(
     send_event(user_profile.realm, event, list(map(user_info, ums)))
 
 
+def get_visibility_policy_after_merge(
+    orig_topic_visibility_policy: int, target_topic_visibility_policy: int
+) -> int:
+    # This function determines the final visibility_policy after the merge
+    # operation, based on the visibility policies of the original and target
+    # topics.
+    #
+    # The algorithm to decide is based on:
+    # Whichever of the two policies is most visible is what we keep.
+    # The general motivation is to err on the side of showing messages
+    # rather than hiding them.
+    if orig_topic_visibility_policy == target_topic_visibility_policy:
+        return orig_topic_visibility_policy
+    elif (
+        orig_topic_visibility_policy == UserTopic.VisibilityPolicy.UNMUTED
+        or target_topic_visibility_policy == UserTopic.VisibilityPolicy.UNMUTED
+    ):
+        return UserTopic.VisibilityPolicy.UNMUTED
+    return UserTopic.VisibilityPolicy.INHERIT
+
+
 # We use transaction.atomic to support select_for_update in the attachment codepath.
 @transaction.atomic(savepoint=False)
 def do_update_message(
@@ -547,6 +569,23 @@ def do_update_message(
 
     update_edit_history(target_message, timestamp, edit_history_event)
 
+    # 'target_topic_has_messages', 'target_stream', and 'target_topic'
+    # will be used while migrating user_topic records later in this function.
+    #
+    # We need to calculate 'target_topic_has_messages' here,
+    # as we are moving the messages in the next step.
+    if topic_name is not None or new_stream is not None:
+        assert stream_being_edited is not None
+        assert orig_topic_name is not None
+
+        target_stream: Stream = new_stream if new_stream is not None else stream_being_edited
+        target_topic: str = topic_name if topic_name is not None else orig_topic_name
+
+        assert target_stream.recipient_id is not None
+        target_topic_has_messages = messages_for_topic(
+            target_stream.recipient_id, target_topic
+        ).exists()
+
     delete_event_notify_user_ids: List[int] = []
     if propagate_mode in ["change_later", "change_all"]:
         assert topic_name is not None or new_stream is not None
@@ -757,16 +796,14 @@ def do_update_message(
     #   the acting user has moved the entire topic (as visible to them).
     #
     # This rule corresponds to checking moved_all_visible_messages.
-    #
-    # We may want more complex behavior in cases where one appears to
-    # be merging topics (E.g. there are existing messages in the
-    # target topic).
     if moved_all_visible_messages:
         assert stream_being_edited is not None
-        assert topic_name is not None or new_stream is not None
+        assert target_stream is not None
+        assert target_topic is not None
 
-        user_profiles_for_visibility_policy: Dict[int, List[UserProfile]] = defaultdict(list)
         stream_inaccessible_to_user_profiles: List[UserProfile] = []
+        orig_topic_user_profile_to_visibility_policy: Dict[UserProfile, int] = {}
+        target_topic_user_profile_to_visibility_policy: Dict[UserProfile, int] = {}
         for user_topic in get_users_with_user_topic_visibility_policy(
             stream_being_edited.id, orig_topic_name
         ):
@@ -776,9 +813,47 @@ def do_update_message(
             ):
                 stream_inaccessible_to_user_profiles.append(user_topic.user_profile)
             else:
-                user_profiles_for_visibility_policy[user_topic.visibility_policy].append(
+                orig_topic_user_profile_to_visibility_policy[
                     user_topic.user_profile
-                )
+                ] = user_topic.visibility_policy
+
+        for user_topic in get_users_with_user_topic_visibility_policy(
+            target_stream.id, target_topic
+        ):
+            target_topic_user_profile_to_visibility_policy[
+                user_topic.user_profile
+            ] = user_topic.visibility_policy
+
+        # User profiles having any of the visibility policies set for either the original or target topic.
+        user_profiles_having_visibility_policy: Set[UserProfile] = set(
+            itertools.chain(
+                orig_topic_user_profile_to_visibility_policy.keys(),
+                target_topic_user_profile_to_visibility_policy.keys(),
+            )
+        )
+
+        user_profiles_for_visibility_policy_pair: Dict[
+            Tuple[int, int], List[UserProfile]
+        ] = defaultdict(list)
+        for user_profile in user_profiles_having_visibility_policy:
+            if user_profile not in target_topic_user_profile_to_visibility_policy:
+                target_topic_user_profile_to_visibility_policy[
+                    user_profile
+                ] = UserTopic.VisibilityPolicy.INHERIT
+            elif user_profile not in orig_topic_user_profile_to_visibility_policy:
+                orig_topic_user_profile_to_visibility_policy[
+                    user_profile
+                ] = UserTopic.VisibilityPolicy.INHERIT
+
+            orig_topic_visibility_policy = orig_topic_user_profile_to_visibility_policy[
+                user_profile
+            ]
+            target_topic_visibility_policy = target_topic_user_profile_to_visibility_policy[
+                user_profile
+            ]
+            user_profiles_for_visibility_policy_pair[
+                (orig_topic_visibility_policy, target_topic_visibility_policy)
+            ].append(user_profile)
 
         # If the messages are being moved to a stream the user
         # cannot access, then we treat this as the
@@ -797,29 +872,66 @@ def do_update_message(
         # can access. We move the user topic records for such
         # users, but removing the old topic visibility_policy
         # and then creating a new one.
-        for visibility_policy, user_profiles in user_profiles_for_visibility_policy.items():
+        #
-            # Using the dict, we get the user_profiles lists for the original_topic
+        # Algorithm used for the 'merge userTopic states' case:
-            # that corresponds to each visibility_policy.
+        # Using the 'user_profiles_for_visibility_policy_pair' dictionary,
-            # Perform the remove and create operation for each of
+        # we have 'orig_topic_visibility_policy', 'target_topic_visibility_policy',
-            # these user_profiles lists.
+        # and a list of 'user_profiles' having the mentioned visibility policies.
-            bulk_do_set_user_topic_visibility_policy(
+        #
-                user_profiles,
+        # For every 'orig_topic_visibility_policy and target_topic_visibility_policy' pair,
-                stream_being_edited,
+        # we determine the final visibility_policy that should be after the merge.
-                orig_topic_name,
+        # Update the visibility_policy for the concerned set of user_profiles.
-                visibility_policy=UserTopic.VisibilityPolicy.INHERIT,
+        for (
-                # do_set_user_topic_visibility_policy with visibility_policy
+            visibility_policy_pair,
-                # set to 'new_visibility_policy' will send an updated muted topic
+            user_profiles,
-                # event, which contains the full set of muted
+        ) in user_profiles_for_visibility_policy_pair.items():
-                # topics, just after this.
+            orig_topic_visibility_policy, target_topic_visibility_policy = visibility_policy_pair
-                skip_muted_topics_event=True,
-            )
 
-            bulk_do_set_user_topic_visibility_policy(
+            if orig_topic_visibility_policy != UserTopic.VisibilityPolicy.INHERIT:
-                user_profiles,
+                bulk_do_set_user_topic_visibility_policy(
-                new_stream if new_stream is not None else stream_being_edited,
+                    user_profiles,
-                topic_name if topic_name is not None else orig_topic_name,
+                    stream_being_edited,
-                visibility_policy=visibility_policy,
+                    orig_topic_name,
-            )
+                    visibility_policy=UserTopic.VisibilityPolicy.INHERIT,
+                    # bulk_do_set_user_topic_visibility_policy with visibility_policy
+                    # set to 'new_visibility_policy' will send an updated muted topic
+                    # event, which contains the full set of muted
+                    # topics, just after this.
+                    skip_muted_topics_event=True,
+                )
+
+            new_visibility_policy = orig_topic_visibility_policy
+
+            if target_topic_has_messages:
+                # Here, we handle the complex case when target_topic already has
+                # some messages. We determine the resultant visibility_policy
+                # based on the visibility_policy of the orig_topic + target_topic.
+                # Finally, bulk_update the user_topic rows with the new visibility_policy.
+                new_visibility_policy = get_visibility_policy_after_merge(
+                    orig_topic_visibility_policy, target_topic_visibility_policy
+                )
+                if new_visibility_policy == target_topic_visibility_policy:
+                    continue
+                bulk_do_set_user_topic_visibility_policy(
+                    user_profiles,
+                    target_stream,
+                    target_topic,
+                    visibility_policy=new_visibility_policy,
+                )
+            else:
+                # This corresponds to the case when messages are moved
+                # to a stream-topic pair that didn't exist.
+                if new_visibility_policy == UserTopic.VisibilityPolicy.INHERIT:
+                    # This avoids unnecessary db operations and INFO logs.
+                    # As INHERIT is the default visibility_policy in the new
+                    # stream-topic pair for users.
+                    continue
+                bulk_do_set_user_topic_visibility_policy(
+                    user_profiles,
+                    target_stream,
+                    target_topic,
+                    visibility_policy=new_visibility_policy,
+                )
 
     send_event(user_profile.realm, event, users_to_be_notified)
 

zerver/tests/test_message_edit.py

@@ -16,6 +16,7 @@ from zerver.actions.message_edit import (
 from zerver.actions.reactions import do_add_reaction
 from zerver.actions.realm_settings import do_change_realm_plan_type, do_set_realm_property
 from zerver.actions.streams import do_change_stream_post_policy, do_deactivate_stream
+from zerver.actions.user_topics import do_set_user_topic_visibility_policy
 from zerver.actions.users import do_change_user_role
 from zerver.lib.message import MessageDict, has_message_access, messages_for_ids, truncate_topic
 from zerver.lib.test_classes import ZulipTestCase, get_topic_messages
@@ -1359,7 +1360,7 @@ class EditMessageTest(EditMessageTestCase):
         # state + 1/user with a UserTopic row for the events data)
         # beyond what is typical were there not UserTopic records to
         # update. Ideally, we'd eliminate the per-user component.
-        with self.assert_database_query_count(19):
+        with self.assert_database_query_count(21):
             check_update_message(
                 user_profile=hamlet,
                 message_id=message_id,
@@ -1464,7 +1465,7 @@ class EditMessageTest(EditMessageTestCase):
         set_topic_visibility_policy(desdemona, muted_topics, UserTopic.VisibilityPolicy.MUTED)
         set_topic_visibility_policy(cordelia, muted_topics, UserTopic.VisibilityPolicy.MUTED)
 
-        with self.assert_database_query_count(28):
+        with self.assert_database_query_count(30):
             check_update_message(
                 user_profile=desdemona,
                 message_id=message_id,
@@ -1495,7 +1496,7 @@ class EditMessageTest(EditMessageTestCase):
         set_topic_visibility_policy(desdemona, muted_topics, UserTopic.VisibilityPolicy.MUTED)
         set_topic_visibility_policy(cordelia, muted_topics, UserTopic.VisibilityPolicy.MUTED)
 
-        with self.assert_database_query_count(33):
+        with self.assert_database_query_count(35):
             check_update_message(
                 user_profile=desdemona,
                 message_id=message_id,
@@ -1528,7 +1529,7 @@ class EditMessageTest(EditMessageTestCase):
         set_topic_visibility_policy(desdemona, muted_topics, UserTopic.VisibilityPolicy.MUTED)
         set_topic_visibility_policy(cordelia, muted_topics, UserTopic.VisibilityPolicy.MUTED)
 
-        with self.assert_database_query_count(28):
+        with self.assert_database_query_count(30):
             check_update_message(
                 user_profile=desdemona,
                 message_id=message_id,
@@ -1551,7 +1552,7 @@ class EditMessageTest(EditMessageTestCase):
         second_message_id = self.send_stream_message(
             hamlet, stream_name, topic_name="changed topic name", content="Second message"
         )
-        with self.assert_database_query_count(24):
+        with self.assert_database_query_count(26):
             check_update_message(
                 user_profile=desdemona,
                 message_id=second_message_id,
@@ -1650,7 +1651,7 @@ class EditMessageTest(EditMessageTestCase):
             users_to_be_notified_via_muted_topics_event.append(user_topic.user_profile_id)
 
         change_all_topic_name = "Topic 1 edited"
-        with self.assert_database_query_count(24):
+        with self.assert_database_query_count(26):
             check_update_message(
                 user_profile=hamlet,
                 message_id=message_id,
@@ -1721,6 +1722,174 @@ class EditMessageTest(EditMessageTestCase):
             aaron, change_all_topic_name, UserTopic.VisibilityPolicy.MUTED, expected=False
         )
 
+    def test_merge_user_topic_states_on_move_messages(self) -> None:
+        stream_name = "Stream 123"
+        stream = self.make_stream(stream_name)
+
+        hamlet = self.example_user("hamlet")
+        cordelia = self.example_user("cordelia")
+        aaron = self.example_user("aaron")
+
+        def assert_has_visibility_policy(
+            user_profile: UserProfile,
+            topic_name: str,
+            visibility_policy: int,
+        ) -> None:
+            self.assertTrue(
+                topic_has_visibility_policy(user_profile, stream.id, topic_name, visibility_policy)
+            )
+
+        self.subscribe(hamlet, stream_name)
+        self.login_user(hamlet)
+        self.subscribe(cordelia, stream_name)
+        self.login_user(cordelia)
+        self.subscribe(aaron, stream_name)
+        self.login_user(aaron)
+
+        # Test the following cases:
+        #
+        #  orig_topic | target_topic | final behaviour
+        #   INHERIT       INHERIT       INHERIT
+        #   INHERIT        MUTED        INHERIT
+        #   INHERIT       UNMUTED       UNMUTED
+        orig_topic = "Topic1"
+        target_topic = "Topic1 edited"
+        orig_message_id = self.send_stream_message(
+            hamlet, stream_name, topic_name=orig_topic, content="Hello World"
+        )
+        self.send_stream_message(
+            hamlet, stream_name, topic_name=target_topic, content="Hello World 2"
+        )
+
+        # By default:
+        # visibility_policy of 'hamlet', 'cordelia', 'aaron' for 'orig_topic': INHERIT
+        # visibility_policy of 'hamlet' for 'target_topic': INHERIT
+        #
+        # So we don't need to manually set visibility_policy to INHERIT whenever required,
+        # here and later in this test.
+        do_set_user_topic_visibility_policy(
+            cordelia, stream, target_topic, visibility_policy=UserTopic.VisibilityPolicy.MUTED
+        )
+        do_set_user_topic_visibility_policy(
+            aaron, stream, target_topic, visibility_policy=UserTopic.VisibilityPolicy.UNMUTED
+        )
+
+        check_update_message(
+            user_profile=hamlet,
+            message_id=orig_message_id,
+            stream_id=None,
+            topic_name=target_topic,
+            propagate_mode="change_all",
+            send_notification_to_old_thread=False,
+            send_notification_to_new_thread=False,
+            content=None,
+        )
+
+        assert_has_visibility_policy(hamlet, orig_topic, UserTopic.VisibilityPolicy.INHERIT)
+        assert_has_visibility_policy(cordelia, orig_topic, UserTopic.VisibilityPolicy.INHERIT)
+        assert_has_visibility_policy(aaron, orig_topic, UserTopic.VisibilityPolicy.INHERIT)
+        assert_has_visibility_policy(hamlet, target_topic, UserTopic.VisibilityPolicy.INHERIT)
+        assert_has_visibility_policy(cordelia, target_topic, UserTopic.VisibilityPolicy.INHERIT)
+        assert_has_visibility_policy(aaron, target_topic, UserTopic.VisibilityPolicy.UNMUTED)
+
+        # Test the following cases:
+        #
+        #  orig_topic | target_topic | final behaviour
+        #     MUTED       INHERIT        INHERIT
+        #     MUTED        MUTED          MUTED
+        #     MUTED       UNMUTED        UNMUTED
+        orig_topic = "Topic2"
+        target_topic = "Topic2 edited"
+        orig_message_id = self.send_stream_message(
+            hamlet, stream_name, topic_name=orig_topic, content="Hello World"
+        )
+        self.send_stream_message(
+            hamlet, stream_name, topic_name=target_topic, content="Hello World 2"
+        )
+
+        do_set_user_topic_visibility_policy(
+            hamlet, stream, orig_topic, visibility_policy=UserTopic.VisibilityPolicy.MUTED
+        )
+        do_set_user_topic_visibility_policy(
+            cordelia, stream, orig_topic, visibility_policy=UserTopic.VisibilityPolicy.MUTED
+        )
+        do_set_user_topic_visibility_policy(
+            aaron, stream, orig_topic, visibility_policy=UserTopic.VisibilityPolicy.MUTED
+        )
+        do_set_user_topic_visibility_policy(
+            cordelia, stream, target_topic, visibility_policy=UserTopic.VisibilityPolicy.MUTED
+        )
+        do_set_user_topic_visibility_policy(
+            aaron, stream, target_topic, visibility_policy=UserTopic.VisibilityPolicy.UNMUTED
+        )
+
+        check_update_message(
+            user_profile=hamlet,
+            message_id=orig_message_id,
+            stream_id=None,
+            topic_name=target_topic,
+            propagate_mode="change_all",
+            send_notification_to_old_thread=False,
+            send_notification_to_new_thread=False,
+            content=None,
+        )
+
+        assert_has_visibility_policy(hamlet, orig_topic, UserTopic.VisibilityPolicy.INHERIT)
+        assert_has_visibility_policy(cordelia, orig_topic, UserTopic.VisibilityPolicy.INHERIT)
+        assert_has_visibility_policy(aaron, orig_topic, UserTopic.VisibilityPolicy.INHERIT)
+        assert_has_visibility_policy(hamlet, target_topic, UserTopic.VisibilityPolicy.INHERIT)
+        assert_has_visibility_policy(cordelia, target_topic, UserTopic.VisibilityPolicy.MUTED)
+        assert_has_visibility_policy(aaron, target_topic, UserTopic.VisibilityPolicy.UNMUTED)
+
+        # Test the following cases:
+        #
+        #  orig_topic | target_topic | final behaviour
+        #    UNMUTED       INHERIT        UNMUTED
+        #    UNMUTED        MUTED         UNMUTED
+        #    UNMUTED       UNMUTED        UNMUTED
+        orig_topic = "Topic3"
+        target_topic = "Topic3 edited"
+        orig_message_id = self.send_stream_message(
+            hamlet, stream_name, topic_name=orig_topic, content="Hello World"
+        )
+        self.send_stream_message(
+            hamlet, stream_name, topic_name=target_topic, content="Hello World 2"
+        )
+
+        do_set_user_topic_visibility_policy(
+            hamlet, stream, orig_topic, visibility_policy=UserTopic.VisibilityPolicy.UNMUTED
+        )
+        do_set_user_topic_visibility_policy(
+            cordelia, stream, orig_topic, visibility_policy=UserTopic.VisibilityPolicy.UNMUTED
+        )
+        do_set_user_topic_visibility_policy(
+            aaron, stream, orig_topic, visibility_policy=UserTopic.VisibilityPolicy.UNMUTED
+        )
+        do_set_user_topic_visibility_policy(
+            cordelia, stream, target_topic, visibility_policy=UserTopic.VisibilityPolicy.MUTED
+        )
+        do_set_user_topic_visibility_policy(
+            aaron, stream, target_topic, visibility_policy=UserTopic.VisibilityPolicy.UNMUTED
+        )
+
+        check_update_message(
+            user_profile=hamlet,
+            message_id=orig_message_id,
+            stream_id=None,
+            topic_name=target_topic,
+            propagate_mode="change_all",
+            send_notification_to_old_thread=False,
+            send_notification_to_new_thread=False,
+            content=None,
+        )
+
+        assert_has_visibility_policy(hamlet, orig_topic, UserTopic.VisibilityPolicy.INHERIT)
+        assert_has_visibility_policy(cordelia, orig_topic, UserTopic.VisibilityPolicy.INHERIT)
+        assert_has_visibility_policy(aaron, orig_topic, UserTopic.VisibilityPolicy.INHERIT)
+        assert_has_visibility_policy(hamlet, target_topic, UserTopic.VisibilityPolicy.UNMUTED)
+        assert_has_visibility_policy(cordelia, target_topic, UserTopic.VisibilityPolicy.UNMUTED)
+        assert_has_visibility_policy(aaron, target_topic, UserTopic.VisibilityPolicy.UNMUTED)
+
     @mock.patch("zerver.actions.message_edit.send_event")
     def test_wildcard_mention(self, mock_send_event: mock.MagicMock) -> None:
         stream_name = "Macbeth"
@@ -2981,7 +3150,7 @@ class EditMessageTest(EditMessageTestCase):
             "iago", "test move stream", "new stream", "test"
         )
 
-        with self.assert_database_query_count(55), cache_tries_captured() as cache_tries:
+        with self.assert_database_query_count(57), cache_tries_captured() as cache_tries:
             result = self.client_patch(
                 f"/json/messages/{msg_id}",
                 {