from datetime import timedelta
from typing import Any
from unittest import mock

import orjson
import time_machine
from django.test import override_settings
from django.utils.timezone import now as timezone_now

from zerver.actions.message_delete import do_delete_messages
from zerver.actions.message_edit import (
    build_message_edit_request,
    check_update_message,
    do_update_message,
    maybe_send_resolve_topic_notifications,
)
from zerver.actions.reactions import do_add_reaction
from zerver.actions.realm_settings import (
    do_change_realm_permission_group_setting,
    do_set_realm_property,
)
from zerver.actions.streams import do_change_stream_group_based_setting, do_set_stream_property
from zerver.actions.user_groups import check_add_user_group
from zerver.actions.user_settings import do_change_user_setting
from zerver.actions.user_topics import do_set_user_topic_visibility_policy
from zerver.lib.message import truncate_topic
from zerver.lib.test_classes import ZulipTestCase, get_topic_messages
from zerver.lib.timestamp import datetime_to_timestamp
from zerver.lib.topic import RESOLVED_TOPIC_PREFIX
from zerver.lib.types import StreamMessageEditRequest, UserGroupMembersData
from zerver.lib.user_topics import (
    get_users_with_user_topic_visibility_policy,
    set_topic_visibility_policy,
    topic_has_visibility_policy,
)
from zerver.lib.utils import assert_is_not_none
from zerver.models import Message, UserMessage, UserProfile, UserTopic
from zerver.models.constants import MAX_TOPIC_NAME_LENGTH
from zerver.models.groups import NamedUserGroup, SystemGroups
from zerver.models.realms import RealmTopicsPolicyEnum
from zerver.models.streams import Stream, StreamTopicsPolicyEnum, get_stream
from zerver.models.users import ResolvedTopicNoticeAutoReadPolicyEnum


class MessageMoveTopicTest(ZulipTestCase):
    def check_topic(self, msg_id: int, topic_name: str) -> None:
        msg = Message.objects.get(id=msg_id)
        self.assertEqual(msg.topic_name(), topic_name)

    def assert_move_message(
        self,
        user: str,
        orig_stream: Stream,
        orig_topic_name: str = "test",
        stream_id: int | None = None,
        topic_name: str | None = None,
        expected_error: str | None = None,
    ) -> None:
        user_profile = self.example_user(user)
        self.subscribe(user_profile, orig_stream.name)
        message_id = self.send_stream_message(
            user_profile, orig_stream.name, topic_name=orig_topic_name
        )

        params_dict: dict[str, str | int] = {}
        if stream_id is not None:
            params_dict["stream_id"] = stream_id
        if topic_name is not None:
            params_dict["topic"] = topic_name

        result = self.api_patch(
            user_profile,
            "/api/v1/messages/" + str(message_id),
            params_dict,
        )
        if expected_error is not None:
            self.assert_json_error(result, expected_error)
        else:
            self.assert_json_success(result)

    def assert_has_visibility_policy(
        self,
        user_profile: UserProfile,
        topic_name: str,
        stream: Stream,
        visibility_policy: int,
        *,
        expected: bool = True,
    ) -> None:
        if expected:
            self.assertTrue(
                topic_has_visibility_policy(user_profile, stream.id, topic_name, visibility_policy)
            )
        else:
            self.assertFalse(
                topic_has_visibility_policy(user_profile, stream.id, topic_name, visibility_policy)
            )

    def test_private_message_edit_topic(self) -> None:
        hamlet = self.example_user("hamlet")
        self.login("hamlet")
        cordelia = self.example_user("cordelia")
        msg_id = self.send_personal_message(hamlet, cordelia)

        result = self.client_patch(
            "/json/messages/" + str(msg_id),
            {
                "topic": "Should not exist",
            },
        )

        self.assert_json_error(result, "Direct messages cannot have topics.")

    def test_propagate_invalid(self) -> None:
        self.login("hamlet")
        id1 = self.send_stream_message(self.example_user("hamlet"), "Denmark", topic_name="topic1")

        result = self.client_patch(
            "/json/messages/" + str(id1),
            {
                "topic": "edited",
                "propagate_mode": "invalid",
            },
        )
        self.assert_json_error(result, "Invalid propagate_mode")
        self.check_topic(id1, topic_name="topic1")

        result = self.client_patch(
            "/json/messages/" + str(id1),
            {
                "content": "edited",
                "propagate_mode": "change_all",
            },
        )
        self.assert_json_error(result, "Invalid propagate_mode without topic edit")
        self.check_topic(id1, topic_name="topic1")

    def test_edit_message_empty_topic_with_extra_space(self) -> None:
        self.login("hamlet")
        msg_id = self.send_stream_message(
            self.example_user("hamlet"), "Denmark", topic_name="editing", content="before edit"
        )
        result = self.client_patch(
            "/json/messages/" + str(msg_id),
            {
                "topic": " ",
            },
        )
        self.assert_json_success(result)
        self.check_topic(msg_id, "")

    def test_topic_required_in_mandatory_topic_realm(self) -> None:
        admin_user = self.example_user("iago")
        hamlet = self.example_user("hamlet")
        realm = admin_user.realm
        self.login_user(admin_user)

        stream = self.make_stream("new_stream")
        stream_mandatory_topics = self.make_stream(
            "topics_required", topics_policy=StreamTopicsPolicyEnum.disable_empty_topic.value
        )
        self.subscribe(admin_user, stream.name)
        self.subscribe(hamlet, stream.name)

        original_topic_name = "topic 1"
        message_id = self.send_stream_message(
            hamlet,
            stream.name,
            topic_name=original_topic_name,
        )

        # Verify with topics_policy=disable_empty_topic:
        # * A topic can't be moved to an empty topic
        # * A topic can be moved to a non-empty topic
        do_set_realm_property(
            realm,
            "topics_policy",
            RealmTopicsPolicyEnum.disable_empty_topic,
            acting_user=admin_user,
        )

        for topic_name in ["(no topic)", ""]:
            result = self.client_patch(
                f"/json/messages/{message_id}",
                {
                    "topic": topic_name,
                },
            )
            self.assert_json_error(
                result, "Sending messages to the general chat is not allowed in this channel."
            )
            self.check_topic(message_id, topic_name=original_topic_name)

        new_topic_name = "new valid topic"
        result = self.client_patch(
            f"/json/messages/{message_id}",
            {
                "topic": new_topic_name,
            },
        )
        self.assert_json_success(result)
        self.check_topic(message_id, new_topic_name)

        # Verify with topics_policy=allow_empty_topic:
        # * A topic can be moved to an empty topic
        # * A topic can be moved to a non-empty topic
        do_set_realm_property(
            realm, "topics_policy", RealmTopicsPolicyEnum.allow_empty_topic, acting_user=admin_user
        )

        for topic_name in ["(no topic)", "", "non-empty topic"]:
            result = self.client_patch(
                f"/json/messages/{message_id}",
                {
                    "topic": topic_name,
                },
            )
            self.assert_json_success(result)
            self.check_topic(message_id, topic_name)

        # Test that message cannot be moved to empty topic in stream with
        # `topics_policy=disable_empty_topic`.
        for topic_name in ["(no topic)", ""]:
            result = self.client_patch(
                f"/json/messages/{message_id}",
                {"topic": topic_name, "stream_id": stream_mandatory_topics.id},
            )
            self.assert_json_error(
                result, "Sending messages to the general chat is not allowed in this channel."
            )
            self.check_topic(message_id, topic_name="non-empty topic")

        # Test that message cannot be moved to empty topic in stream with
        # `topics_policy=disable_empty_topic` when `topic_name` is `None`.
        for topic_name in ["(no topic)", ""]:
            message_id = self.send_stream_message(
                hamlet,
                stream.name,
                topic_name=topic_name,
            )
            result = self.client_patch(
                f"/json/messages/{message_id}",
                {"stream_id": stream_mandatory_topics.id},
            )
            self.assert_json_error(
                result, "Sending messages to the general chat is not allowed in this channel."
            )
            self.check_topic(message_id, topic_name="")

    def test_edit_message_invalid_topic(self) -> None:
        self.login("hamlet")
        msg_id = self.send_stream_message(
            self.example_user("hamlet"), "Denmark", topic_name="editing", content="before edit"
        )
        result = self.client_patch(
            "/json/messages/" + str(msg_id),
            {
                "topic": "editing\nfun",
            },
        )
        self.assert_json_error(result, "Invalid character in topic, at position 8!")

        result = self.client_patch(
            "/json/messages/" + str(msg_id),
            {
                "topic": f"{Message.DM_TOPIC}",
            },
        )
        self.assert_json_error(result, "Invalid character in topic, at position 1!")

    @mock.patch("zerver.actions.message_edit.send_event_on_commit")
    def test_edit_topic_public_history_stream(self, mock_send_event: mock.MagicMock) -> None:
        stream_name = "Macbeth"
        hamlet = self.example_user("hamlet")
        cordelia = self.example_user("cordelia")
        self.make_stream(stream_name, history_public_to_subscribers=True)
        self.subscribe(hamlet, stream_name)
        self.login_user(hamlet)
        message_id = self.send_stream_message(hamlet, stream_name, "Where am I?")

        self.login_user(cordelia)
        self.subscribe(cordelia, stream_name)
        message = Message.objects.get(id=message_id)

        def do_update_message_topic_success(
            user_profile: UserProfile,
            message: Message,
            topic_name: str,
            users_to_be_notified: list[dict[str, Any]],
        ) -> None:
            message_edit_request = build_message_edit_request(
                message=message,
                user_profile=user_profile,
                propagate_mode="change_later",
                stream_id=None,
                topic_name=topic_name,
                content=None,
            )
            do_update_message(
                user_profile=user_profile,
                target_message=message,
                message_edit_request=message_edit_request,
                send_notification_to_old_thread=False,
                send_notification_to_new_thread=False,
                rendering_result=None,
                prior_mention_user_ids=set(),
                mention_data=None,
            )

            mock_send_event.assert_called_with(mock.ANY, mock.ANY, users_to_be_notified)

        # Returns the users that need to be notified when a message topic is changed
        def notify(user_id: int) -> dict[str, Any]:
            um = UserMessage.objects.get(message=message_id)
            if um.user_profile_id == user_id:
                return {
                    "id": user_id,
                    "flags": um.flags_list(),
                }

            else:
                return {
                    "id": user_id,
                    "flags": ["read"],
                }

        users_to_be_notified = list(map(notify, [hamlet.id, cordelia.id]))
        # Edit topic of a message sent before Cordelia subscribed the stream
        do_update_message_topic_success(
            cordelia, message, "Othello eats apple", users_to_be_notified
        )

        # If Cordelia is long-term idle, she doesn't get a notification.
        cordelia.long_term_idle = True
        cordelia.save()
        users_to_be_notified = list(map(notify, [hamlet.id]))
        do_update_message_topic_success(
            cordelia, message, "Another topic idle", users_to_be_notified
        )
        cordelia.long_term_idle = False
        cordelia.save()

        # Even if Hamlet unsubscribes the stream, he should be notified when the topic is changed
        # because he has a UserMessage row.
        self.unsubscribe(hamlet, stream_name)
        users_to_be_notified = list(map(notify, [hamlet.id, cordelia.id]))
        do_update_message_topic_success(cordelia, message, "Another topic", users_to_be_notified)

        # Hamlet subscribes to the stream again and Cordelia unsubscribes, then Hamlet changes
        # the message topic. Cordelia won't receive any updates when a message on that stream is
        # changed because she is not a subscriber and doesn't have a UserMessage row.
        self.subscribe(hamlet, stream_name)
        self.unsubscribe(cordelia, stream_name)
        self.login_user(hamlet)
        users_to_be_notified = list(map(notify, [hamlet.id]))
        do_update_message_topic_success(hamlet, message, "Change again", users_to_be_notified)

    @mock.patch("zerver.actions.user_topics.send_event_on_commit")
    def test_edit_muted_topic(self, mock_send_event_on_commit: mock.MagicMock) -> 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")
        self.subscribe(hamlet, stream_name)
        self.login_user(hamlet)
        message_id = self.send_stream_message(
            hamlet, stream_name, topic_name="Topic1", content="Hello World"
        )

        self.subscribe(cordelia, stream_name)
        self.login_user(cordelia)
        self.subscribe(aaron, stream_name)
        self.login_user(aaron)

        def assert_is_topic_muted(
            user_profile: UserProfile,
            stream_id: int,
            topic_name: str,
            *,
            muted: bool,
        ) -> None:
            if muted:
                self.assertTrue(
                    topic_has_visibility_policy(
                        user_profile, stream_id, topic_name, UserTopic.VisibilityPolicy.MUTED
                    )
                )
            else:
                self.assertFalse(
                    topic_has_visibility_policy(
                        user_profile, stream_id, topic_name, UserTopic.VisibilityPolicy.MUTED
                    )
                )

        already_muted_topic_name = "Already muted topic"
        muted_topics = [
            [stream_name, "Topic1"],
            [stream_name, "Topic2"],
            [stream_name, already_muted_topic_name],
        ]
        set_topic_visibility_policy(hamlet, muted_topics, UserTopic.VisibilityPolicy.MUTED)
        set_topic_visibility_policy(cordelia, muted_topics, UserTopic.VisibilityPolicy.MUTED)

        # users that need to be notified by send_event in the case of change-topic-name operation.
        users_to_be_notified_via_muted_topics_event: list[int] = []
        users_to_be_notified_via_user_topic_event: list[int] = []
        for user_topic in get_users_with_user_topic_visibility_policy(stream.id, "Topic1"):
            # We are appending the same data twice because 'user_topic' event notifies
            # the user during delete and create operation.
            users_to_be_notified_via_user_topic_event.append(user_topic.user_profile_id)
            users_to_be_notified_via_user_topic_event.append(user_topic.user_profile_id)
            # 'muted_topics' event notifies the user of muted topics during create
            # operation only.
            users_to_be_notified_via_muted_topics_event.append(user_topic.user_profile_id)

        change_all_topic_name = "Topic 1 edited"
        # Verify how many total database queries are required. We
        # expect 6 queries (4/visibility_policy to update the muted
        # 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(27):
            check_update_message(
                user_profile=hamlet,
                message_id=message_id,
                stream_id=None,
                topic_name=change_all_topic_name,
                propagate_mode="change_all",
                send_notification_to_old_thread=False,
                send_notification_to_new_thread=False,
                content=None,
            )

        # Extract the send_event call where event type is 'user_topic' or 'muted_topics.
        # Here we assert that the expected users are notified properly.
        users_notified_via_muted_topics_event: list[int] = []
        users_notified_via_user_topic_event: list[int] = []
        for call_args in mock_send_event_on_commit.call_args_list:
            (_arg_realm, arg_event, arg_notified_users) = call_args[0]
            if arg_event["type"] == "user_topic":
                users_notified_via_user_topic_event.append(*arg_notified_users)
            elif arg_event["type"] == "muted_topics":
                users_notified_via_muted_topics_event.append(*arg_notified_users)
        self.assertEqual(
            sorted(users_notified_via_muted_topics_event),
            sorted(users_to_be_notified_via_muted_topics_event),
        )
        self.assertEqual(
            sorted(users_notified_via_user_topic_event),
            sorted(users_to_be_notified_via_user_topic_event),
        )

        assert_is_topic_muted(hamlet, stream.id, "Topic1", muted=False)
        assert_is_topic_muted(cordelia, stream.id, "Topic1", muted=False)
        assert_is_topic_muted(aaron, stream.id, "Topic1", muted=False)
        assert_is_topic_muted(hamlet, stream.id, "Topic2", muted=True)
        assert_is_topic_muted(cordelia, stream.id, "Topic2", muted=True)
        assert_is_topic_muted(aaron, stream.id, "Topic2", muted=False)
        assert_is_topic_muted(hamlet, stream.id, change_all_topic_name, muted=True)
        assert_is_topic_muted(cordelia, stream.id, change_all_topic_name, muted=True)
        assert_is_topic_muted(aaron, stream.id, change_all_topic_name, muted=False)

        change_later_topic_name = "Topic 1 edited again"
        check_update_message(
            user_profile=hamlet,
            message_id=message_id,
            stream_id=None,
            topic_name=change_later_topic_name,
            propagate_mode="change_later",
            send_notification_to_old_thread=False,
            send_notification_to_new_thread=False,
            content=None,
        )
        assert_is_topic_muted(hamlet, stream.id, change_all_topic_name, muted=False)
        assert_is_topic_muted(hamlet, stream.id, change_later_topic_name, muted=True)

        # Make sure we safely handle the case of the new topic being already muted.
        check_update_message(
            user_profile=hamlet,
            message_id=message_id,
            stream_id=None,
            topic_name=already_muted_topic_name,
            propagate_mode="change_all",
            send_notification_to_old_thread=False,
            send_notification_to_new_thread=False,
            content=None,
        )
        assert_is_topic_muted(hamlet, stream.id, change_later_topic_name, muted=False)
        assert_is_topic_muted(hamlet, stream.id, already_muted_topic_name, muted=True)

        change_one_topic_name = "Topic 1 edited change_one"
        check_update_message(
            user_profile=hamlet,
            message_id=message_id,
            stream_id=None,
            topic_name=change_one_topic_name,
            propagate_mode="change_one",
            send_notification_to_old_thread=False,
            send_notification_to_new_thread=False,
            content=None,
        )
        assert_is_topic_muted(hamlet, stream.id, change_one_topic_name, muted=True)
        assert_is_topic_muted(hamlet, stream.id, change_later_topic_name, muted=False)

        # Move topic between two public streams.
        desdemona = self.example_user("desdemona")
        message_id = self.send_stream_message(
            hamlet, stream_name, topic_name="New topic", content="Hello World"
        )
        new_public_stream = self.make_stream("New public stream")
        self.subscribe(desdemona, new_public_stream.name)
        self.login_user(desdemona)
        muted_topics = [
            [stream_name, "New topic"],
        ]
        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(29):
            check_update_message(
                user_profile=desdemona,
                message_id=message_id,
                stream_id=new_public_stream.id,
                propagate_mode="change_all",
                send_notification_to_old_thread=False,
                send_notification_to_new_thread=False,
                content=None,
            )

        assert_is_topic_muted(desdemona, stream.id, "New topic", muted=False)
        assert_is_topic_muted(cordelia, stream.id, "New topic", muted=False)
        assert_is_topic_muted(aaron, stream.id, "New topic", muted=False)
        assert_is_topic_muted(desdemona, new_public_stream.id, "New topic", muted=True)
        assert_is_topic_muted(cordelia, new_public_stream.id, "New topic", muted=True)
        assert_is_topic_muted(aaron, new_public_stream.id, "New topic", muted=False)

        # Move topic to a private stream.
        message_id = self.send_stream_message(
            hamlet, stream_name, topic_name="New topic", content="Hello World"
        )
        new_private_stream = self.make_stream("New private stream", invite_only=True)
        self.subscribe(desdemona, new_private_stream.name)
        self.login_user(desdemona)
        muted_topics = [
            [stream_name, "New topic"],
        ]
        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(35):
            check_update_message(
                user_profile=desdemona,
                message_id=message_id,
                stream_id=new_private_stream.id,
                propagate_mode="change_all",
                send_notification_to_old_thread=False,
                send_notification_to_new_thread=False,
                content=None,
            )

        # Cordelia is not subscribed to the private stream, so
        # Cordelia should have had the topic unmuted, while Desdemona
        # should have had her muted topic record moved.
        assert_is_topic_muted(desdemona, stream.id, "New topic", muted=False)
        assert_is_topic_muted(cordelia, stream.id, "New topic", muted=False)
        assert_is_topic_muted(aaron, stream.id, "New topic", muted=False)
        assert_is_topic_muted(desdemona, new_private_stream.id, "New topic", muted=True)
        assert_is_topic_muted(cordelia, new_private_stream.id, "New topic", muted=False)
        assert_is_topic_muted(aaron, new_private_stream.id, "New topic", muted=False)

        # Move topic between two public streams with change in topic name.
        desdemona = self.example_user("desdemona")
        message_id = self.send_stream_message(
            hamlet, stream_name, topic_name="New topic 2", content="Hello World"
        )
        self.login_user(desdemona)
        muted_topics = [
            [stream_name, "New topic 2"],
        ]
        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(31):
            check_update_message(
                user_profile=desdemona,
                message_id=message_id,
                stream_id=new_public_stream.id,
                topic_name="changed topic name",
                propagate_mode="change_all",
                send_notification_to_old_thread=False,
                send_notification_to_new_thread=False,
                content=None,
            )

        assert_is_topic_muted(desdemona, stream.id, "New topic 2", muted=False)
        assert_is_topic_muted(cordelia, stream.id, "New topic 2", muted=False)
        assert_is_topic_muted(aaron, stream.id, "New topic 2", muted=False)
        assert_is_topic_muted(desdemona, new_public_stream.id, "changed topic name", muted=True)
        assert_is_topic_muted(cordelia, new_public_stream.id, "changed topic name", muted=True)
        assert_is_topic_muted(aaron, new_public_stream.id, "changed topic name", muted=False)

        # Moving only half the messages doesn't move UserTopic records.
        second_message_id = self.send_stream_message(
            hamlet, stream_name, topic_name="changed topic name", content="Second message"
        )
        with self.assert_database_query_count(25):
            check_update_message(
                user_profile=desdemona,
                message_id=second_message_id,
                stream_id=new_public_stream.id,
                topic_name="final topic name",
                propagate_mode="change_later",
                send_notification_to_old_thread=False,
                send_notification_to_new_thread=False,
                content=None,
            )

        assert_is_topic_muted(desdemona, new_public_stream.id, "changed topic name", muted=True)
        assert_is_topic_muted(cordelia, new_public_stream.id, "changed topic name", muted=True)
        assert_is_topic_muted(aaron, new_public_stream.id, "changed topic name", muted=False)
        assert_is_topic_muted(desdemona, new_public_stream.id, "final topic name", muted=False)
        assert_is_topic_muted(cordelia, new_public_stream.id, "final topic name", muted=False)
        assert_is_topic_muted(aaron, new_public_stream.id, "final topic name", muted=False)

    @mock.patch("zerver.actions.user_topics.send_event_on_commit")
    def test_edit_unmuted_topic(self, mock_send_event_on_commit: mock.MagicMock) -> 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")
        othello = self.example_user("othello")

        self.subscribe(hamlet, stream_name)
        self.login_user(hamlet)
        message_id = self.send_stream_message(
            hamlet, stream_name, topic_name="Topic1", content="Hello World"
        )

        self.subscribe(cordelia, stream_name)
        self.login_user(cordelia)
        self.subscribe(aaron, stream_name)
        self.login_user(aaron)
        self.subscribe(othello, stream_name)
        self.login_user(othello)

        # Initially, hamlet and othello set visibility_policy as UNMUTED for 'Topic1' and 'Topic2',
        # cordelia sets visibility_policy as MUTED for 'Topic1' and 'Topic2', while
        # aaron doesn't have a visibility_policy set for 'Topic1' or 'Topic2'.
        #
        # After moving messages from 'Topic1' to 'Topic 1 edited', the expected behaviour is:
        # hamlet and othello have UNMUTED 'Topic 1 edited' and no visibility_policy set for 'Topic1'
        # cordelia has MUTED 'Topic 1 edited' and no visibility_policy set for 'Topic1'
        #
        # There is no change in visibility_policy configurations for 'Topic2', i.e.
        # hamlet and othello have UNMUTED 'Topic2' + cordelia has MUTED 'Topic2'
        # aaron still doesn't have visibility_policy set for any topic.
        #
        # Note: We have used two users with UNMUTED 'Topic1' to verify that the query count
        # doesn't increase (in order to update UserTopic records) with an increase in users.
        # (We are using bulk database operations.)
        # 1 query/user is added in order to send muted_topics event.(which will be deprecated)
        topics = [
            [stream_name, "Topic1"],
            [stream_name, "Topic2"],
        ]
        set_topic_visibility_policy(hamlet, topics, UserTopic.VisibilityPolicy.UNMUTED)
        set_topic_visibility_policy(cordelia, topics, UserTopic.VisibilityPolicy.MUTED)
        set_topic_visibility_policy(othello, topics, UserTopic.VisibilityPolicy.UNMUTED)

        # users that need to be notified by send_event in the case of change-topic-name operation.
        users_to_be_notified_via_muted_topics_event: list[int] = []
        users_to_be_notified_via_user_topic_event: list[int] = []
        for user_topic in get_users_with_user_topic_visibility_policy(stream.id, "Topic1"):
            # We are appending the same data twice because 'user_topic' event notifies
            # the user during delete and create operation.
            users_to_be_notified_via_user_topic_event.append(user_topic.user_profile_id)
            users_to_be_notified_via_user_topic_event.append(user_topic.user_profile_id)
            # 'muted_topics' event notifies the user of muted topics during create
            # operation only.
            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(32):
            check_update_message(
                user_profile=hamlet,
                message_id=message_id,
                stream_id=None,
                topic_name=change_all_topic_name,
                propagate_mode="change_all",
                send_notification_to_old_thread=False,
                send_notification_to_new_thread=False,
                content=None,
            )

        # Extract the send_event call where event type is 'user_topic' or 'muted_topics.
        # Here we assert that the expected users are notified properly.
        users_notified_via_muted_topics_event: list[int] = []
        users_notified_via_user_topic_event: list[int] = []
        for call_args in mock_send_event_on_commit.call_args_list:
            (_arg_realm, arg_event, arg_notified_users) = call_args[0]
            if arg_event["type"] == "user_topic":
                users_notified_via_user_topic_event.append(*arg_notified_users)
            elif arg_event["type"] == "muted_topics":
                users_notified_via_muted_topics_event.append(*arg_notified_users)
        self.assertEqual(
            sorted(users_notified_via_muted_topics_event),
            sorted(users_to_be_notified_via_muted_topics_event),
        )
        self.assertEqual(
            sorted(users_notified_via_user_topic_event),
            sorted(users_to_be_notified_via_user_topic_event),
        )

        # No visibility_policy set for 'Topic1'
        self.assert_has_visibility_policy(
            hamlet, "Topic1", stream, UserTopic.VisibilityPolicy.UNMUTED, expected=False
        )
        self.assert_has_visibility_policy(
            cordelia, "Topic1", stream, UserTopic.VisibilityPolicy.MUTED, expected=False
        )
        self.assert_has_visibility_policy(
            othello, "Topic1", stream, UserTopic.VisibilityPolicy.UNMUTED, expected=False
        )
        self.assert_has_visibility_policy(
            aaron, "Topic1", stream, UserTopic.VisibilityPolicy.UNMUTED, expected=False
        )
        # No change in visibility_policy configurations for 'Topic2'
        self.assert_has_visibility_policy(
            hamlet, "Topic2", stream, UserTopic.VisibilityPolicy.UNMUTED, expected=True
        )
        self.assert_has_visibility_policy(
            cordelia, "Topic2", stream, UserTopic.VisibilityPolicy.MUTED, expected=True
        )
        self.assert_has_visibility_policy(
            othello, "Topic2", stream, UserTopic.VisibilityPolicy.UNMUTED, expected=True
        )
        self.assert_has_visibility_policy(
            aaron, "Topic2", stream, UserTopic.VisibilityPolicy.UNMUTED, expected=False
        )
        # UserTopic records moved to 'Topic 1 edited' after move-topic operation.
        self.assert_has_visibility_policy(
