move_topic_to_stream: Allow moving to/between/from private streams.

Fixes #16284.

Most of the work for this was done when we implemented correct
behavior for guest users, since they treat public streams like private
streams anyway.

The general method involves moving the messages to the new stream with
special care of UserMessage.

We delete UserMessages for subs who are losing access to the message.
For private streams with protected history, we also create UserMessage
elements for users who are not present in the old stream, since that's
important for those users to access the moved messages.

templates/zerver/help/rename-a-topic.md

@@ -36,7 +36,7 @@ for the details on when topic editing is allowed.
 
 {!admin-only.md!}
 
-Organization administrators can move a topic from one public stream to
+Organization administrators can move a topic from one stream to
 another.
 
 {start_tabs}
@@ -54,6 +54,12 @@ another.
 
 1. Click **Move topic**.
 
+
+!!! warn ""
+    **Note**: When a topic is moved to a private stream with protected history,
+              messages in the topic will be visible to all the subscribers.
+
+
 {end_tabs}
 
 ## Move message(s) in a topic to another stream

zerver/lib/actions.py

@@ -4582,6 +4582,7 @@ def do_update_message(user_profile: UserProfile, message: Message,
         subs_to_new_stream = list(get_active_subscriptions_for_stream_id(
             new_stream.id).select_related("user_profile"))
 
+        old_stream_sub_ids = [user.user_profile_id for user in subscribers]
         new_stream_sub_ids = [user.user_profile_id for user in subs_to_new_stream]
 
         # Get users who aren't subscribed to the new_stream.
@@ -4591,17 +4592,24 @@ def do_update_message(user_profile: UserProfile, message: Message,
         ]
         # Users who can longer access the message without some action
         # from administrators.
-        #
-        # TODO: Extend this list to also contain users losing access
-        # due to the messages moving to a private stream they are not
-        # subscribed to.
         subs_losing_access = [
             sub for sub in subs_losing_usermessages
-            if sub.user_profile.is_guest
+            if sub.user_profile.is_guest or not new_stream.is_public()
         ]
         ums = ums.exclude(user_profile_id__in=[
             sub.user_profile_id for sub in subs_losing_usermessages])
 
+        subs_gaining_usermessages = []
+        if not new_stream.is_history_public_to_subscribers():
+            # For private streams, with history not public to subscribers,
+            # We find out users who are not present in the msgs' old stream
+            # and create new UserMessage for these users so that they can
+            # access this message.
+            subs_gaining_usermessages += [
+                user_id for user_id in new_stream_sub_ids
+                if user_id not in old_stream_sub_ids
+            ]
+
     if topic_name is not None:
         topic_name = truncate_topic(topic_name)
         message.set_topic_name(topic_name)
@@ -4627,6 +4635,25 @@ def do_update_message(user_profile: UserProfile, message: Message,
         if new_stream is not None:
             assert stream_being_edited is not None
 
+            if subs_gaining_usermessages:
+                ums_to_create = []
+                for msg in changed_messages:
+                    for user_profile_id in subs_gaining_usermessages:
+                        # The fact that the user didn't have a UserMessage originally means we can infer that the user
+                        # was not mentioned in the original message (even if mention syntax was present, it would not
+                        # take effect for a user who was not subscribed). If we were editing the message's content, we
+                        # would rerender the message and then use the new stream's data to determine whether this is
+                        # a mention of a subscriber; but as we are not doing so, we choose to preserve the "was this
+                        # mention syntax an actual mention" decision made during the original rendering for implementation
+                        # simplicity. As a result, the only flag to consider applying here is read.
+                        um = UserMessageLite(
+                            user_profile_id=user_profile_id,
+                            message_id=msg.id,
+                            flags=UserMessage.flags.read,
+                        )
+                        ums_to_create.append(um)
+                bulk_insert_ums(ums_to_create)
+
             message_ids = [msg.id for msg in changed_messages]
             # Delete UserMessage objects for users who will no
             # longer have access to these messages.  Note: This could be

zerver/tests/test_message_edit.py

@@ -963,7 +963,7 @@ class EditMessageTest(ZulipTestCase):
                 'propagate_mode': 'change_all',
                 'topic': 'new topic',
             })
-        self.assertEqual(len(queries), 52)
+        self.assertEqual(len(queries), 51)
 
         messages = get_topic_messages(user_profile, old_stream, "test")
         self.assertEqual(len(messages), 1)
@@ -1068,33 +1068,101 @@ class EditMessageTest(ZulipTestCase):
         messages = get_topic_messages(user_profile, new_stream, "test")
         self.assertEqual(len(messages), 3)
 
-    def test_move_message_to_stream_to_private_stream(self) -> None:
+    def parameterized_test_move_message_involving_private_stream(
-        user_profile = self.example_user("iago")
+        self,
+        from_invite_only: bool,
+        history_public_to_subscribers: bool,
+        create_user_message: bool,
+        to_invite_only: bool=True,
+    ) -> None:
+        admin_user = self.example_user("iago")
+        user_losing_access = self.example_user('cordelia')
+        user_gaining_access = self.example_user('hamlet')
+
         self.login("iago")
-        stream = self.make_stream("test move stream")
+        old_stream = self.make_stream("test move stream", invite_only=from_invite_only)
-        new_stream = self.make_stream("new stream", None, True)
+        new_stream = self.make_stream("new stream", invite_only=to_invite_only, history_public_to_subscribers=history_public_to_subscribers)
-        self.subscribe(user_profile, stream.name)
+
-        self.subscribe(user_profile, new_stream.name)
+        self.subscribe(admin_user, old_stream.name)
-        msg_id = self.send_stream_message(user_profile, stream.name,
+        self.subscribe(user_losing_access, old_stream.name)
+
+        self.subscribe(admin_user, new_stream.name)
+        self.subscribe(user_gaining_access, new_stream.name)
+
+        msg_id = self.send_stream_message(admin_user, old_stream.name,
                                           topic_name="test", content="First")
-        self.send_stream_message(user_profile, stream.name,
+        self.send_stream_message(admin_user, old_stream.name,
                                  topic_name="test", content="Second")
 
+        self.assertEqual(UserMessage.objects.filter(
+            user_profile_id=user_losing_access.id,
+            message_id=msg_id,
+        ).count(), 1)
+        self.assertEqual(UserMessage.objects.filter(
+            user_profile_id=user_gaining_access.id,
+            message_id=msg_id,
+        ).count(), 0)
+
         result = self.client_patch("/json/messages/" + str(msg_id), {
             'message_id': msg_id,
             'stream_id': new_stream.id,
             'propagate_mode': 'change_all',
         })
+        self.assert_json_success(result)
 
-        self.assert_json_error(result, "Streams must be public")
+        messages = get_topic_messages(admin_user, old_stream, "test")
+        self.assertEqual(len(messages), 1)
+        self.assertEqual(messages[0].content, f"This topic was moved by @_**Iago|{admin_user.id}** to #**new stream>test**")
 
-        # We expect the messages to remain in the original stream/topic
+        messages = get_topic_messages(admin_user, new_stream, "test")
-        messages = get_topic_messages(user_profile, stream, "test")
+        self.assertEqual(len(messages), 3)
-        self.assertEqual(len(messages), 2)
 
-        messages = get_topic_messages(user_profile, new_stream, "test")
+        self.assertEqual(UserMessage.objects.filter(
-        self.assertEqual(len(messages), 0)
+            user_profile_id=user_losing_access.id,
+            message_id=msg_id,
+        ).count(), 0)
+        # When the history is shared, UserMessage is not created for the user but the user
+        # can see the message.
+        self.assertEqual(UserMessage.objects.filter(
+            user_profile_id=user_gaining_access.id,
+            message_id=msg_id,
+        ).count(), 1 if create_user_message else 0)
 
+    def test_move_message_from_public_to_private_stream_not_shared_history(self) -> None:
+        self.parameterized_test_move_message_involving_private_stream(
+            from_invite_only=False,
+            history_public_to_subscribers=False,
+            create_user_message=True,
+        )
+
+    def test_move_message_from_public_to_private_stream_shared_history(self) -> None:
+        self.parameterized_test_move_message_involving_private_stream(
+            from_invite_only=False,
+            history_public_to_subscribers=True,
+            create_user_message=False,
+        )
+
+    def test_move_message_from_private_to_private_stream_not_shared_history(self) -> None:
+        self.parameterized_test_move_message_involving_private_stream(
+            from_invite_only=True,
+            history_public_to_subscribers=False,
+            create_user_message=True,
+        )
+
+    def test_move_message_from_private_to_private_stream_shared_history(self) -> None:
+        self.parameterized_test_move_message_involving_private_stream(
+            from_invite_only=True,
+            history_public_to_subscribers=True,
+            create_user_message=False,
+        )
+
+    def test_move_message_from_private_to_public(self) -> None:
+        self.parameterized_test_move_message_involving_private_stream(
+            from_invite_only=True,
+            history_public_to_subscribers=True,
+            create_user_message=False,
+            to_invite_only=False,
+        )
 
 class DeleteMessageTest(ZulipTestCase):
     def test_delete_message_invalid_request_format(self) -> None:

zerver/views/message_edit.py

@@ -187,7 +187,6 @@ def update_message_backend(request: HttpRequest, user_profile: UserMessage,
         mention_user_ids = message.mentions_user_ids
 
     new_stream = None
-    old_stream = None
     number_changed = 0
 
     if stream_id is not None:
@@ -196,15 +195,8 @@ def update_message_backend(request: HttpRequest, user_profile: UserMessage,
         if content is not None:
             raise JsonableError(_("Cannot change message content while changing stream"))
 
-        old_stream = get_stream_by_id(message.recipient.type_id)
         new_stream = get_stream_by_id(stream_id)
 
-        if not (old_stream.is_public() and new_stream.is_public()):
-            # We'll likely decide to relax this condition in the
-            # future; it just requires more care with details like the
-            # breadcrumb messages.
-            raise JsonableError(_("Streams must be public"))
-
     number_changed = do_update_message(user_profile, message, new_stream,
                                        topic_name, propagate_mode,
                                        send_notification_to_old_thread,