message_edit: Add backend for moving a topic to another stream.

This commit reuses the existing infrastructure for moving a topic
within a stream to add support for moving topics from one stream to
another.

Split from the original full-feature commit so that we can merge just
the backend, which is finished, at this time.

This is a large part of #6427.

The feature is incomplete, in that we don't have real-time update of
the frontend to handle the event, documentation, etc., but this commit
is a good mergable checkpoint that we can do further work on top of.
We also still ideally would have a test_events test for the backend,
but I'm willing to leave that for follow-up work.

This appears to have switched to tabbott as the author during commit
squashing sometime ago, but this commit is certainly:

Co-Authored-By: Wbert Adrián Castro Vera <wbertc@gmail.com>

templates/zerver/api/update-message.md

@@ -38,7 +38,7 @@ zulip(config).then((client) => {
 
 {tab|curl}
 
-{generate_code_example(curl)|/messages/{message_id}:patch|example}
+{generate_code_example(curl, exclude=["stream_id"])|/messages/{message_id}:patch|example}
 
 {end_tabs}
 

zerver/lib/actions.py

@@ -4367,6 +4367,30 @@ MessageUpdateUserInfoResult = TypedDict('MessageUpdateUserInfoResult', {
     'mention_user_ids': Set[int],
 })
 
+def notify_topic_moved_streams(user_profile: UserProfile,
+                               old_stream: Stream, old_topic: str,
+                               new_stream: Stream, new_topic: Optional[str]) -> None:
+    # Since moving content between streams is highly disruptive,
+    # it's worth adding a couple tombstone messages showing what
+    # happened.
+    sender = get_system_bot(settings.NOTIFICATION_BOT)
+
+    if new_topic is None:
+        new_topic = old_topic
+
+    user_mention = "@_**%s|%s**" % (user_profile.full_name, user_profile.id)
+    old_topic_link = "#**%s>%s**" % (old_stream.name, old_topic)
+    new_topic_link = "#**%s>%s**" % (new_stream.name, new_topic)
+
+    internal_send_stream_message(
+        new_stream.realm, sender, new_stream, new_topic,
+        _("This topic was moved here from %s by %s") % (old_topic_link, user_mention))
+
+    # Send a notification to the old stream that the topic was moved.
+    internal_send_stream_message(
+        old_stream.realm, sender, old_stream, old_topic,
+        _("This topic was moved by %s to %s") % (user_mention, new_topic_link))
+
 def get_user_info_for_message_updates(message_id: int) -> MessageUpdateUserInfoResult:
 
     # We exclude UserMessage.flags.historical rows since those
@@ -4475,7 +4499,8 @@ def do_update_embedded_data(user_profile: UserProfile,
 
 # We use transaction.atomic to support select_for_update in the attachment codepath.
 @transaction.atomic
-def do_update_message(user_profile: UserProfile, message: Message, topic_name: Optional[str],
+def do_update_message(user_profile: UserProfile, message: Message,
+                      new_stream: Optional[Stream], topic_name: Optional[str],
                       propagate_mode: str, content: Optional[str],
                       rendered_content: Optional[str], prior_mention_user_ids: Set[int],
                       mention_user_ids: Set[int], mention_data: Optional[bugdown.MentionData]=None) -> int:
@@ -4576,13 +4601,26 @@ def do_update_message(user_profile: UserProfile, message: Message, topic_name: O
         else:
             event['wildcard_mention_user_ids'] = []
 
-    if topic_name is not None:
+    if topic_name is not None or new_stream is not None:
         orig_topic_name = message.topic_name()
-        topic_name = truncate_topic(topic_name)
         event["propagate_mode"] = propagate_mode
-        message.set_topic_name(topic_name)
         event["stream_id"] = message.recipient.type_id
 
+    if new_stream is not None:
+        assert content is None
+        assert message.is_stream_message()
+        assert stream_being_edited is not None
+
+        edit_history_event['prev_stream'] = stream_being_edited.id
+        message.recipient_id = new_stream.recipient_id
+
+        event["new_stream_id"] = new_stream.id
+        event["propagate_mode"] = propagate_mode
+
+    if topic_name is not None:
+        topic_name = truncate_topic(topic_name)
+        message.set_topic_name(topic_name)
+
         # These fields have legacy field names.
         event[ORIG_TOPIC] = orig_topic_name
         event[TOPIC_NAME] = topic_name
@@ -4590,12 +4628,13 @@ def do_update_message(user_profile: UserProfile, message: Message, topic_name: O
         edit_history_event[LEGACY_PREV_TOPIC] = orig_topic_name
 
     if propagate_mode in ["change_later", "change_all"]:
-        assert topic_name is not None
+        assert topic_name is not None or new_stream is not None
         messages_list = update_messages_for_topic_edit(
             message=message,
             propagate_mode=propagate_mode,
             orig_topic_name=orig_topic_name,
             topic_name=topic_name,
+            new_stream=new_stream
         )
         changed_messages += messages_list
 
@@ -4650,6 +4689,13 @@ def do_update_message(user_profile: UserProfile, message: Message, topic_name: O
             users_to_be_notified += list(map(subscriber_info, subscribers_ids))
 
     send_event(user_profile.realm, event, users_to_be_notified)
+
+    if (len(changed_messages) > 0 and new_stream is not None and
+       stream_being_edited is not None):
+        # Notify users that the topic was moved.
+        notify_topic_moved_streams(user_profile, stream_being_edited, orig_topic_name,
+                                   new_stream, topic_name)
+
     return len(changed_messages)
 
 def do_delete_messages(realm: Realm, messages: Iterable[Message]) -> None:
@@ -5909,3 +5955,11 @@ def do_delete_realm_export(user_profile: UserProfile, export: RealmAuditLog) ->
     export.extra_data = ujson.dumps(export_data)
     export.save(update_fields=['extra_data'])
     notify_realm_export(user_profile)
+
+def get_topic_messages(user_profile: UserProfile, stream: Stream,
+                       topic_name: str) -> List[Message]:
+    query = UserMessage.objects.filter(
+        user_profile=user_profile,
+        message__recipient=stream.recipient
+    ).order_by("id")
+    return [um.message for um in filter_by_topic_name_via_message(query, topic_name)]

zerver/lib/topic.py

@@ -14,6 +14,7 @@ from zerver.lib.request import REQ
 from zerver.models import (
     Message,
     Recipient,
+    Stream,
     UserMessage,
     UserProfile,
 )
@@ -119,11 +120,15 @@ def user_message_exists_for_topic(user_profile: UserProfile,
 def update_messages_for_topic_edit(message: Message,
                                    propagate_mode: str,
                                    orig_topic_name: str,
-                                   topic_name: str) -> List[Message]:
+                                   topic_name: Optional[str],
+                                   new_stream: Optional[Stream]) -> List[Message]:
     propagate_query = Q(recipient = message.recipient, subject = orig_topic_name)
+    if propagate_mode == 'change_all':
         # We only change messages up to 7 days in the past, to avoid hammering our
         # DB by changing an unbounded amount of messages
-    if propagate_mode == 'change_all':
+        #
+        # TODO: Look at removing this restriction and/or add a "change_last_week"
+        # option; this behavior feels buggy.
         before_bound = timezone_now() - datetime.timedelta(days=7)
 
         propagate_query = (propagate_query & ~Q(id = message.id) &
@@ -133,15 +138,26 @@ def update_messages_for_topic_edit(message: Message,
 
     messages = Message.objects.filter(propagate_query).select_related()
 
+    update_fields = {}
+
     # Evaluate the query before running the update
     messages_list = list(messages)
-    messages.update(subject=topic_name)
 
+    # The cached ORM objects are not changed by the upcoming
+    # messages.update(), and the remote cache update (done by the
+    # caller) requires the new value, so we manually update the
+    # objects in addition to sending a bulk query to the database.
+    if new_stream is not None:
+        update_fields["recipient"] = new_stream.recipient
+        for m in messages_list:
+            m.recipient = new_stream.recipient
+    if topic_name is not None:
+        update_fields["subject"] = topic_name
         for m in messages_list:
-        # The cached ORM object is not changed by messages.update()
-        # and the remote cache update requires the new value
             m.set_topic_name(topic_name)
 
+    messages.update(**update_fields)
+
     return messages_list
 
 def generate_topic_history_from_db_rows(rows: List[Tuple[str, int]]) -> List[Dict[str, Any]]:

zerver/openapi/zulip.yaml

@@ -675,6 +675,12 @@ paths:
         schema:
           type: string
         example: Hello
+      - name: stream_id
+        in: query
+        description: The stream ID to move the message(s) to, if moving to another stream.
+        schema:
+          type: integer
+        example: 2
       responses:
         '200':
           description: Success.

zerver/tests/test_events.py

@@ -746,7 +746,7 @@ class EventsRegisterTest(ZulipTestCase):
         )
 
         events = self.do_test(
-            lambda: do_update_message(self.user_profile, message, topic,
+            lambda: do_update_message(self.user_profile, message, None, topic,
                                       propagate_mode, content, rendered_content,
                                       prior_mention_user_ids,
                                       mentioned_user_ids, mention_data),

zerver/tests/test_messages.py

@@ -30,6 +30,7 @@ from zerver.lib.actions import (
     get_active_presence_idle_user_ids,
     get_client,
     get_last_message_id,
+    get_topic_messages,
     get_user_info_for_message_updates,
     internal_prep_private_message,
     internal_prep_stream_message_by_name,
@@ -120,7 +121,7 @@ import mock
 from operator import itemgetter
 import time
 import ujson
-from typing import Any, Dict, List, Set, Union
+from typing import Any, Dict, List, Set, Union, Tuple
 
 from collections import namedtuple
 
@@ -3227,6 +3228,7 @@ class EditMessageTest(ZulipTestCase):
             do_update_message(
                 user_profile=user_profile,
                 message=message,
+                new_stream=None,
                 topic_name=topic_name,
                 propagate_mode="change_later",
                 content=None,
@@ -3390,6 +3392,146 @@ class EditMessageTest(ZulipTestCase):
         self.assert_json_error(result, 'Invalid propagate_mode without topic edit')
         self.check_topic(id1, topic_name="topic1")
 
+    def prepare_move_topics(self, user_email: str, old_stream: str, new_stream: str, topic: str) -> Tuple[UserProfile, Stream, Stream, int, int]:
+        user_profile = self.example_user(user_email)
+        self.login(user_email)
+        stream = self.make_stream(old_stream)
+        new_stream = self.make_stream(new_stream)
+        self.subscribe(user_profile, stream.name)
+        self.subscribe(user_profile, new_stream.name)
+        msg_id = self.send_stream_message(user_profile, stream.name,
+                                          topic_name=topic, content="First")
+        msg_id_lt = self.send_stream_message(user_profile, stream.name,
+                                             topic_name=topic, content="Second")
+
+        self.send_stream_message(user_profile, stream.name,
+                                 topic_name=topic, content="third")
+
+        return (user_profile, stream, new_stream, msg_id, msg_id_lt)
+
+    def test_move_message_to_stream(self) -> None:
+        (user_profile, old_stream, new_stream, msg_id, msg_id_lt) = self.prepare_move_topics(
+            "iago", "test move stream", "new stream", "test")
+
+        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)
+
+        messages = get_topic_messages(user_profile, old_stream, "test")
+        self.assertEqual(len(messages), 1)
+        self.assertEqual(messages[0].content, "This topic was moved by @_**Iago|%s** to #**new stream>test**" % (user_profile.id,))
+
+        messages = get_topic_messages(user_profile, new_stream, "test")
+        self.assertEqual(len(messages), 4)
+        self.assertEqual(messages[3].content, "This topic was moved here from #**test move stream>test** by @_**Iago|%s**" % (user_profile.id,))
+
+    def test_move_message_to_stream_change_later(self) -> None:
+        (user_profile, old_stream, new_stream, msg_id, msg_id_later) = self.prepare_move_topics(
+            "iago", "test move stream", "new stream", "test")
+
+        result = self.client_patch("/json/messages/" + str(msg_id_later), {
+            'message_id': msg_id_later,
+            'stream_id': new_stream.id,
+            'propagate_mode': 'change_later'
+        })
+        self.assert_json_success(result)
+
+        messages = get_topic_messages(user_profile, old_stream, "test")
+        self.assertEqual(len(messages), 2)
+        self.assertEqual(messages[0].id, msg_id)
+        self.assertEqual(messages[1].content, "This topic was moved by @_**Iago|%s** to #**new stream>test**" % (user_profile.id,))
+
+        messages = get_topic_messages(user_profile, new_stream, "test")
+        self.assertEqual(len(messages), 3)
+        self.assertEqual(messages[0].id, msg_id_later)
+        self.assertEqual(messages[2].content, "This topic was moved here from #**test move stream>test** by @_**Iago|%d**" % (user_profile.id,))
+
+    def test_move_message_to_stream_no_allowed(self) -> None:
+        (user_profile, old_stream, new_stream, msg_id, msg_id_later) = self.prepare_move_topics(
+            "aaron", "test move stream", "new stream", "test")
+
+        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_error(result, "You don't have permission to move this message")
+
+        messages = get_topic_messages(user_profile, old_stream, "test")
+        self.assertEqual(len(messages), 3)
+
+        messages = get_topic_messages(user_profile, new_stream, "test")
+        self.assertEqual(len(messages), 0)
+
+    def test_move_message_to_stream_with_content(self) -> None:
+        (user_profile, old_stream, new_stream, msg_id, msg_id_later) = self.prepare_move_topics(
+            "iago", "test move stream", "new stream", "test")
+
+        result = self.client_patch("/json/messages/" + str(msg_id), {
+            'message_id': msg_id,
+            'stream_id': new_stream.id,
+            'propagate_mode': 'change_all',
+            'content': 'Not allowed'
+        })
+        self.assert_json_error(result, "Cannot change message content while changing stream")
+
+        messages = get_topic_messages(user_profile, old_stream, "test")
+        self.assertEqual(len(messages), 3)
+
+        messages = get_topic_messages(user_profile, new_stream, "test")
+        self.assertEqual(len(messages), 0)
+
+    def test_move_message_to_stream_and_topic(self) -> None:
+        (user_profile, old_stream, new_stream, msg_id, msg_id_later) = self.prepare_move_topics(
+            "iago", "test move stream", "new stream", "test")
+
+        result = self.client_patch("/json/messages/" + str(msg_id), {
+            'message_id': msg_id,
+            'stream_id': new_stream.id,
+            'propagate_mode': 'change_all',
+            'topic': 'new topic'
+        })
+
+        messages = get_topic_messages(user_profile, old_stream, "test")
+        self.assertEqual(len(messages), 1)
+        self.assertEqual(messages[0].content, "This topic was moved by @_**Iago|%s** to #**new stream>new topic**" % (user_profile.id,))
+
+        messages = get_topic_messages(user_profile, new_stream, "new topic")
+        self.assertEqual(len(messages), 4)
+        self.assertEqual(messages[3].content, "This topic was moved here from #**test move stream>test** by @_**Iago|%s**" % (user_profile.id,))
+        self.assert_json_success(result)
+
+    def test_move_message_to_stream_to_private_stream(self) -> None:
+        user_profile = self.example_user("iago")
+        self.login("iago")
+        stream = self.make_stream("test move stream")
+        new_stream = self.make_stream("new stream", None, True)
+        self.subscribe(user_profile, stream.name)
+        self.subscribe(user_profile, new_stream.name)
+        msg_id = self.send_stream_message(user_profile, stream.name,
+                                          topic_name="test", content="First")
+        self.send_stream_message(user_profile, stream.name,
+                                 topic_name="test", content="Second")
+
+        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_error(result, "Streams must be public")
+
+        # We expect the messages to remain in the original stream/topic
+        messages = get_topic_messages(user_profile, stream, "test")
+        self.assertEqual(len(messages), 2)
+
+        messages = get_topic_messages(user_profile, new_stream, "test")
+        self.assertEqual(len(messages), 0)
+
 class MirroredMessageUsersTest(ZulipTestCase):
     def test_invalid_sender(self) -> None:
         user = self.example_user('hamlet')
@@ -3974,7 +4116,8 @@ class MessageHasKeywordsTest(ZulipTestCase):
         realm_id = hamlet.realm.id
         rendered_content = render_markdown(msg, content)
         mention_data = bugdown.MentionData(realm_id, content)
-        do_update_message(hamlet, msg, None, "change_one", content, rendered_content, set(), set(), mention_data=mention_data)
+        do_update_message(hamlet, msg, None, None, "change_one", content,
+                          rendered_content, set(), set(), mention_data=mention_data)
 
     def test_finds_link_after_edit(self) -> None:
         hamlet = self.example_user('hamlet')

zerver/views/messages.py

@@ -32,7 +32,7 @@ from zerver.lib.response import json_success, json_error
 from zerver.lib.sqlalchemy_utils import get_sqlalchemy_connection
 from zerver.lib.streams import access_stream_by_id, get_public_streams_queryset, \
     can_access_stream_history_by_name, can_access_stream_history_by_id, \
-    get_stream_by_narrow_operand_access_unchecked
+    get_stream_by_narrow_operand_access_unchecked, get_stream_by_id
 from zerver.lib.timestamp import datetime_to_timestamp, convert_to_UTC
 from zerver.lib.timezone import get_timezone
 from zerver.lib.topic import (
@@ -1502,6 +1502,7 @@ PROPAGATE_MODE_VALUES = ["change_later", "change_one", "change_all"]
 @has_request_variables
 def update_message_backend(request: HttpRequest, user_profile: UserMessage,
                            message_id: int=REQ(converter=to_non_negative_int, path_only=True),
+                           stream_id: Optional[int]=REQ(converter=to_non_negative_int, default=None),
                            topic_name: Optional[str]=REQ_topic(),
                            propagate_mode: Optional[str]=REQ(
                                default="change_one",
@@ -1510,7 +1511,7 @@ def update_message_backend(request: HttpRequest, user_profile: UserMessage,
     if not user_profile.realm.allow_message_editing:
         return json_error(_("Your organization has turned off message editing"))
 
-    if propagate_mode != "change_one" and topic_name is None:
+    if propagate_mode != "change_one" and topic_name is None and stream_id is None:
         return json_error(_("Invalid propagate_mode without topic edit"))
 
     message, ignored_user_message = access_message(user_profile, message_id)
@@ -1552,7 +1553,7 @@ def update_message_backend(request: HttpRequest, user_profile: UserMessage,
         if (timezone_now() - message.date_sent) > datetime.timedelta(seconds=deadline_seconds):
             raise JsonableError(_("The time limit for editing this message has passed"))
 
-    if topic_name is None and content is None:
+    if topic_name is None and content is None and stream_id is None:
         return json_error(_("Nothing to change"))
     if topic_name is not None:
         topic_name = topic_name.strip()
@@ -1589,8 +1590,28 @@ def update_message_backend(request: HttpRequest, user_profile: UserMessage,
 
         mention_user_ids = message.mentions_user_ids
 
-    number_changed = do_update_message(user_profile, message, topic_name,
-                                       propagate_mode, content, rendered_content,
+    new_stream = None
+    old_stream = None
+    number_changed = 0
+
+    if stream_id is not None:
+        if not user_profile.is_realm_admin:
+            raise JsonableError(_("You don't have permission to move this message"))
+        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,
+                                       content, rendered_content,
                                        prior_mention_user_ids,
                                        mention_user_ids, mention_data)