import itertools
from collections import defaultdict
from collections.abc import Iterable
from collections.abc import Set as AbstractSet
from dataclasses import dataclass
from datetime import timedelta
from typing import Any

from django.conf import settings
from django.db import transaction
from django.db.models import Q, QuerySet
from django.utils.timezone import now as timezone_now
from django.utils.translation import gettext as _
from django.utils.translation import gettext_lazy
from django.utils.translation import override as override_language
from django_stubs_ext import StrPromise

from zerver.actions.message_delete import DeleteMessagesEvent, do_delete_messages
from zerver.actions.message_flags import do_update_mobile_push_notification
from zerver.actions.message_send import (
    filter_presence_idle_user_ids,
    get_recipient_info,
    internal_send_stream_message,
    render_incoming_message,
)
from zerver.actions.uploads import AttachmentChangeResult, check_attachment_reference_change
from zerver.actions.user_topics import bulk_do_set_user_topic_visibility_policy
from zerver.lib import utils
from zerver.lib.exceptions import (
    JsonableError,
    MessageMoveError,
    MessagesNotAllowedInEmptyTopicError,
    PreviousMessageContentMismatchedError,
    StreamWildcardMentionNotAllowedError,
    TopicsNotAllowedError,
    TopicWildcardMentionNotAllowedError,
)
from zerver.lib.markdown import MessageRenderingResult, topic_links
from zerver.lib.markdown import version as markdown_version
from zerver.lib.mention import MentionBackend, MentionData, silent_mention_syntax_for_user
from zerver.lib.message import (
    access_message,
    bulk_access_stream_messages_query,
    check_user_group_mention_allowed,
    event_recipient_ids_for_action_on_messages,
    normalize_body,
    stream_wildcard_mention_allowed,
    topic_wildcard_mention_allowed,
    truncate_topic,
)
from zerver.lib.message_cache import update_message_cache
from zerver.lib.queue import queue_event_on_commit
from zerver.lib.stream_subscription import get_active_subscriptions_for_stream_id
from zerver.lib.stream_topic import StreamTopicTarget
from zerver.lib.streams import (
    access_stream_by_id,
    access_stream_by_id_for_message,
    can_access_stream_history,
    can_edit_topic,
    can_move_messages_out_of_channel,
    can_resolve_topics,
    check_for_can_create_topic_group_violation,
    check_stream_access_based_on_can_send_message_group,
    get_stream_topics_policy,
    notify_stream_is_recently_active_update,
)
from zerver.lib.string_validation import check_stream_topic
from zerver.lib.thumbnail import manifest_and_get_user_upload_previews, rewrite_thumbnailed_images
from zerver.lib.timestamp import datetime_to_timestamp
from zerver.lib.topic import (
    ORIG_TOPIC,
    RESOLVED_TOPIC_PREFIX,
    TOPIC_LINKS,
    TOPIC_NAME,
    get_topic_display_name,
    maybe_rename_general_chat_to_empty_topic,
    messages_for_topic,
    participants_for_topic,
    save_message_for_edit_use_case,
    update_edit_history,
    update_messages_for_topic_edit,
)
from zerver.lib.topic_link_util import get_stream_topic_link_syntax
from zerver.lib.types import DirectMessageEditRequest, EditHistoryEvent, StreamMessageEditRequest
from zerver.lib.url_encoding import stream_message_url
from zerver.lib.user_message import bulk_insert_all_ums
from zerver.lib.user_topics import get_users_with_user_topic_visibility_policy
from zerver.lib.widget import is_widget_message
from zerver.models import (
    ArchivedAttachment,
    ArchivedMessage,
    Attachment,
    Message,
    Recipient,
    Stream,
    Subscription,
    UserMessage,
    UserProfile,
    UserTopic,
)
from zerver.models.streams import (
    StreamTopicsPolicyEnum,
    get_stream_by_id_for_sending_message,
    get_stream_by_id_in_realm,
)
from zerver.models.users import ResolvedTopicNoticeAutoReadPolicyEnum, get_system_bot
from zerver.tornado.django_api import send_event_on_commit


@dataclass
class UpdateMessageResult:
    changed_message_count: int
    detached_uploads: list[dict[str, Any]]


def subscriber_info(user_id: int) -> dict[str, Any]:
    return {"id": user_id, "flags": ["read"]}


def validate_message_edit_payload(
    message: Message,
    stream_id: int | None,
    topic_name: str | None,
    propagate_mode: str | None,
    content: str | None,
    prev_content_sha256: str | None,
) -> None:
    """
    Validates that a message edit request is well-formed. Does not handle permissions.
    """
    if topic_name is None and content is None and stream_id is None:
        raise JsonableError(_("Nothing to change"))

    if not message.is_channel_message:
        if stream_id is not None:
            raise JsonableError(_("Direct messages cannot be moved to channels."))
        if topic_name is not None:
            raise JsonableError(_("Direct messages cannot have topics."))

    if propagate_mode != "change_one" and topic_name is None and stream_id is None:
        raise JsonableError(_("Invalid propagate_mode without topic edit"))

    if topic_name in {
        RESOLVED_TOPIC_PREFIX.strip(),
        f"{RESOLVED_TOPIC_PREFIX}{Message.EMPTY_TOPIC_FALLBACK_NAME}",
    }:
        raise JsonableError(_("General chat cannot be marked as resolved"))

    if topic_name is not None:
        check_stream_topic(topic_name)

    if stream_id is not None and content is not None:
        raise JsonableError(_("Cannot change message content while changing channel"))

    # Right now, we prevent users from editing widgets.
    if content is not None and is_widget_message(message):
        raise JsonableError(_("Widgets cannot be edited."))

    # We don't restrict this check to requests to edit content, to
    # give the parameter a pure meaning. But clients sending this for
    # topic moves are likely doing so in error.
    if prev_content_sha256 is not None and prev_content_sha256 != utils.sha256_hash(
        message.content
    ):
        raise PreviousMessageContentMismatchedError


def validate_user_can_edit_message(
    user_profile: UserProfile, message: Message, edit_limit_buffer: int
) -> None:
    """
    Checks if the user has the permission to edit the message.
    """
    if not user_profile.realm.allow_message_editing:
        raise JsonableError(_("Your organization has turned off message editing"))

    # You cannot edit the content of message sent by someone else.
    if message.sender_id != user_profile.id:
        raise JsonableError(_("You don't have permission to edit this message"))

    if user_profile.realm.message_content_edit_limit_seconds is not None:
        deadline_seconds = user_profile.realm.message_content_edit_limit_seconds + edit_limit_buffer
        if (timezone_now() - message.date_sent) > timedelta(seconds=deadline_seconds):
            raise JsonableError(_("The time limit for editing this message has passed"))


def get_unread_user_ids_for_resolved_topic_notifications(
    stream: Stream,
    topic_name: str,
    users_following_topic: set[UserProfile] | None,
) -> set[int]:
    unread_user_ids: set[int] = set()

    all_stream_user_profiles = UserProfile.objects.filter(
        id__in=get_active_subscriptions_for_stream_id(
            stream.id, include_deactivated_users=False
        ).values_list("user_profile_id", flat=True),
        is_bot=False,
    )

    user_ids_following_topic: set[int] = set()
    if users_following_topic is not None:
        user_ids_following_topic = {user.id for user in users_following_topic}
    else:
        user_ids_following_topic = set(
            UserTopic.objects.filter(
                stream=stream,
                topic_name__iexact=topic_name,
                visibility_policy=UserTopic.VisibilityPolicy.FOLLOWED,
            ).values_list("user_profile_id", flat=True)
        )

    for user_profile in all_stream_user_profiles:
        if (
            user_profile.resolved_topic_notice_auto_read_policy
            == ResolvedTopicNoticeAutoReadPolicyEnum.never.value
        ):
            unread_user_ids.add(user_profile.id)
        elif (
            user_profile.resolved_topic_notice_auto_read_policy
            == ResolvedTopicNoticeAutoReadPolicyEnum.except_followed.value
            and user_profile.id in user_ids_following_topic
        ):
            unread_user_ids.add(user_profile.id)

    return unread_user_ids


def maybe_send_resolve_topic_notifications(
    *,
    user_profile: UserProfile,
    message_edit_request: StreamMessageEditRequest,
    users_following_topic: set[UserProfile] | None,
) -> tuple[int | None, bool]:
    """Returns resolved_topic_message_id if resolve topic notifications were in fact sent."""
    # Note that topics will have already been stripped in check_update_message.
    topic_resolved = message_edit_request.topic_resolved
    topic_unresolved = message_edit_request.topic_unresolved
    if not topic_resolved and not topic_unresolved:
        # If there's some other weird topic that does not toggle the
        # state of "topic starts with RESOLVED_TOPIC_PREFIX", we do
        # nothing. Any other logic could result in cases where we send
        # these notifications in a non-alternating fashion.
        #
        # Note that it is still possible for an individual topic to
        # have multiple "This topic was marked as resolved"
        # notifications in a row: one can send new messages to the
        # pre-resolve topic and then resolve the topic created that
        # way to get multiple in the resolved topic. And then an
        # administrator can delete the messages in between. We consider this
        # to be a fundamental risk of irresponsible message deletion,
        # not a bug with the "resolve topics" feature.
        return None, False

    stream = message_edit_request.orig_stream
    # Sometimes a user might accidentally resolve a topic, and then
    # have to undo the action. We don't want to spam "resolved",
    # "unresolved" messages one after another in such a situation.
    # For that reason, we apply a short grace period during which
    # such an undo action will just delete the previous notification
    # message instead.
    if maybe_delete_previous_resolve_topic_notification(
        user_profile, stream, message_edit_request.target_topic_name
    ):
        return None, True

    # Compute the users for whom this message should be marked as unread
    # based on the `resolved_topic_notice_auto_read_policy` user setting.
    topic_name = message_edit_request.target_topic_name
    unread_user_ids = get_unread_user_ids_for_resolved_topic_notifications(
        stream=stream,
        topic_name=topic_name,
        users_following_topic=users_following_topic,
    )

    sender = get_system_bot(settings.NOTIFICATION_BOT, user_profile.realm_id)
    user_mention = silent_mention_syntax_for_user(user_profile)
    with override_language(stream.realm.default_language):
        if topic_resolved:
            notification_string = _("{user} has marked this topic as resolved.")
        elif topic_unresolved:
            notification_string = _("{user} has marked this topic as unresolved.")

        resolved_topic_message_id = internal_send_stream_message(
            sender,
            stream,
            topic_name,
            notification_string.format(
                user=user_mention,
            ),
            message_type=Message.MessageType.RESOLVE_TOPIC_NOTIFICATION,
            limit_unread_user_ids=unread_user_ids,
            mark_as_read_for_acting_user=True,
            acting_user=user_profile,
        )

    return resolved_topic_message_id, False


def maybe_delete_previous_resolve_topic_notification(
    user_profile: UserProfile, stream: Stream, topic: str
) -> bool:
    assert stream.recipient_id is not None
    last_message = messages_for_topic(stream.realm_id, stream.recipient_id, topic).last()

    if last_message is None:
        return False

    if last_message.type != Message.MessageType.RESOLVE_TOPIC_NOTIFICATION:
        return False

    current_time = timezone_now()
    time_difference = (current_time - last_message.date_sent).total_seconds()

    if time_difference > settings.RESOLVE_TOPIC_UNDO_GRACE_PERIOD_SECONDS:
        return False

    do_delete_messages(stream.realm, [last_message], acting_user=user_profile)
    return True


def send_message_moved_breadcrumbs(
    target_message: Message,
    user_profile: UserProfile,
    message_edit_request: StreamMessageEditRequest,
    old_thread_notification_string: StrPromise | None,
    new_thread_notification_string: StrPromise | None,
    changed_messages_count: int,
) -> None:
    # Since moving content between streams is highly disruptive,
    # it's worth adding a couple tombstone messages showing what
    # happened.
    old_stream = message_edit_request.orig_stream
    sender = get_system_bot(settings.NOTIFICATION_BOT, old_stream.realm_id)

    user_mention = silent_mention_syntax_for_user(user_profile)
    old_topic_name = message_edit_request.orig_topic_name
    new_stream = message_edit_request.target_stream
    new_topic_name = message_edit_request.target_topic_name
    old_topic_link = get_stream_topic_link_syntax(old_stream.id, old_stream.name, old_topic_name)
    new_topic_link = get_stream_topic_link_syntax(new_stream.id, new_stream.name, new_topic_name)
    message = {
        "id": target_message.id,
        "stream_id": new_stream.id,
        "display_recipient": new_stream.name,
        "topic": new_topic_name,
    }
    moved_message_link = stream_message_url(target_message.realm, message)

    if new_thread_notification_string is not None:
        with override_language(new_stream.realm.default_language):
            internal_send_stream_message(
                sender,
                new_stream,
                new_topic_name,
                new_thread_notification_string.format(
                    message_link=moved_message_link,
                    old_location=old_topic_link,
                    user=user_mention,
                    changed_messages_count=changed_messages_count,
                ),
                mark_as_read_for_acting_user=True,
                acting_user=user_profile,
            )

    if old_thread_notification_string is not None:
        with override_language(old_stream.realm.default_language):
            # Send a notification to the old stream that the topic was moved.
            internal_send_stream_message(
                sender,
                old_stream,
                old_topic_name,
                old_thread_notification_string.format(
                    user=user_mention,
                    new_location=new_topic_link,
                    changed_messages_count=changed_messages_count,
                ),
                mark_as_read_for_acting_user=True,
                acting_user=user_profile,
            )


def get_mentions_for_message_updates(message: Message) -> set[int]:
    # We exclude UserMessage.flags.historical rows since those
    # users did not receive the message originally, and thus
    # probably are not relevant for reprocessed alert_words,
    # mentions and similar rendering features.  This may be a
    # decision we change in the future.
    mentioned_user_ids = (
        UserMessage.objects.filter(
            message=message.id,
            flags=~UserMessage.flags.historical,
        )
        .filter(
            Q(
                flags__andnz=UserMessage.flags.mentioned
                | UserMessage.flags.stream_wildcard_mentioned
                | UserMessage.flags.topic_wildcard_mentioned
                | UserMessage.flags.group_mentioned
            )
        )
        .values_list("user_profile_id", flat=True)
    )

    user_ids_having_message_access = event_recipient_ids_for_action_on_messages([message])

    return set(mentioned_user_ids) & user_ids_having_message_access


def update_user_message_flags(
    rendering_result: MessageRenderingResult,
    ums: Iterable[UserMessage],
    topic_participant_user_ids: AbstractSet[int] = set(),
) -> None:
    mentioned_ids = rendering_result.mentions_user_ids
    ids_with_alert_words = rendering_result.user_ids_with_alert_words
    stream_wildcard_mentioned = rendering_result.mentions_stream_wildcard
    changed_ums: set[UserMessage] = set()

    def update_flag(um: UserMessage, should_set: bool, flag: int) -> None:
        if should_set:
            if not (um.flags & flag):
                um.flags |= flag
                changed_ums.add(um)
        else:
            if um.flags & flag:
                um.flags &= ~flag
                changed_ums.add(um)

    for um in ums:
        has_alert_word = um.user_profile_id in ids_with_alert_words
        update_flag(um, has_alert_word, UserMessage.flags.has_alert_word)

        mentioned = um.user_profile_id in mentioned_ids
        update_flag(um, mentioned, UserMessage.flags.mentioned)

        update_flag(um, stream_wildcard_mentioned, UserMessage.flags.stream_wildcard_mentioned)

        topic_wildcard_mentioned = um.user_profile_id in topic_participant_user_ids
        update_flag(um, topic_wildcard_mentioned, UserMessage.flags.topic_wildcard_mentioned)

    for um in changed_ums:
        um.save(update_fields=["flags"])


def do_update_embedded_data(
    user_profile: UserProfile,
    message: Message,
    rendered_content: str | MessageRenderingResult,
    mention_data: MentionData | None = None,
) -> None:
    ums = UserMessage.objects.filter(message=message.id)
    update_fields = ["rendered_content"]
    if isinstance(rendered_content, MessageRenderingResult):
        assert mention_data is not None
        for group_id in rendered_content.mentions_user_group_ids:
            members = mention_data.get_group_members(group_id)
            rendered_content.mentions_user_ids.update(members)
        update_user_message_flags(rendered_content, ums)
        message.rendered_content = rendered_content.rendered_content
        message.rendered_content_version = markdown_version
        update_fields.append("rendered_content_version")
    else:
        message.rendered_content = rendered_content
    message.save(update_fields=update_fields)

    update_message_cache([message])
    event: dict[str, Any] = {
        "type": "update_message",
        "user_id": None,
        "edit_timestamp": datetime_to_timestamp(timezone_now()),
        "message_id": message.id,
        "message_ids": [message.id],
        "content": message.content,
        "rendered_content": message.rendered_content,
        "rendering_only": True,
    }

    users_to_notify = event_recipient_ids_for_action_on_messages([message])
    filtered_ums = [um for um in ums if um.user_profile_id in users_to_notify]

    def user_info(um: UserMessage) -> dict[str, Any]:
        return {
            "id": um.user_profile_id,
            "flags": um.flags_list(),
        }

    send_event_on_commit(user_profile.realm, event, list(map(user_info, filtered_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.
    visibility_policies = {orig_topic_visibility_policy, target_topic_visibility_policy}

    if UserTopic.VisibilityPolicy.FOLLOWED in visibility_policies:
        return UserTopic.VisibilityPolicy.FOLLOWED

    if UserTopic.VisibilityPolicy.UNMUTED in visibility_policies:
        return UserTopic.VisibilityPolicy.UNMUTED

    if UserTopic.VisibilityPolicy.INHERIT in visibility_policies:
        return UserTopic.VisibilityPolicy.INHERIT

    return UserTopic.VisibilityPolicy.MUTED


def update_message_content(
    user_profile: UserProfile,
    target_message: Message,
    content: str,
    rendering_result: MessageRenderingResult,
    prior_mention_user_ids: set[int],
    mention_data: MentionData,
    event: dict[str, Any],
    edit_history_event: EditHistoryEvent,
    stream_topic: StreamTopicTarget | None,
) -> None:
    realm = user_profile.realm

    ums = UserMessage.objects.filter(message=target_message.id)

    # add data from group mentions to mentions_user_ids.
    for group_id in rendering_result.mentions_user_group_ids:
        members = mention_data.get_group_members(group_id)
        rendering_result.mentions_user_ids.update(members)

    # One could imagine checking realm.message_edit_history_visibility_policy here and
    # modifying the events based on that setting, but doing so
    # doesn't really make sense.  We need to send the edit event
    # to clients regardless, and a client already had access to
    # the original/pre-edit content of the message anyway.  That
    # setting must be enforced on the client side, and making a
    # change here simply complicates the logic for clients parsing
    # edit history events.
    edit_history_event["prev_content"] = target_message.content
    edit_history_event["prev_rendered_content"] = target_message.rendered_content
    edit_history_event["prev_rendered_content_version"] = target_message.rendered_content_version

    event["orig_content"] = target_message.content
    event["orig_rendered_content"] = target_message.rendered_content
    event["content"] = content
    event["rendered_content"] = rendering_result.rendered_content
    event["is_me_message"] = Message.is_status_message(content, rendering_result.rendered_content)

    target_message.content = content
    target_message.rendered_content = rendering_result.rendered_content
    target_message.rendered_content_version = markdown_version

    info = get_recipient_info(
        realm_id=realm.id,
        recipient=target_message.recipient,
        sender_id=target_message.sender_id,
        stream_topic=stream_topic,
        possible_topic_wildcard_mention=mention_data.message_has_topic_wildcards(),
        possible_stream_wildcard_mention=mention_data.message_has_stream_wildcards(),
    )

    event["online_push_user_ids"] = list(info.online_push_user_ids)
    event["dm_mention_push_disabled_user_ids"] = list(info.dm_mention_push_disabled_user_ids)
    event["dm_mention_email_disabled_user_ids"] = list(info.dm_mention_email_disabled_user_ids)
    event["stream_push_user_ids"] = list(info.stream_push_user_ids)
    event["stream_email_user_ids"] = list(info.stream_email_user_ids)
    event["followed_topic_push_user_ids"] = list(info.followed_topic_push_user_ids)
    event["followed_topic_email_user_ids"] = list(info.followed_topic_email_user_ids)
    event["muted_sender_user_ids"] = list(info.muted_sender_user_ids)
    event["prior_mention_user_ids"] = list(prior_mention_user_ids)
    event["presence_idle_user_ids"] = filter_presence_idle_user_ids(info.active_user_ids)
    event["all_bot_user_ids"] = list(info.all_bot_user_ids)
    event["push_device_registered_user_ids"] = list(info.push_device_registered_user_ids)
    if rendering_result.mentions_stream_wildcard:
        event["stream_wildcard_mention_user_ids"] = list(info.stream_wildcard_mention_user_ids)
        event["stream_wildcard_mention_in_followed_topic_user_ids"] = list(
            info.stream_wildcard_mention_in_followed_topic_user_ids
        )
    else:
        event["stream_wildcard_mention_user_ids"] = []
        event["stream_wildcard_mention_in_followed_topic_user_ids"] = []

    if rendering_result.mentions_topic_wildcard:
        event["topic_wildcard_mention_user_ids"] = list(info.topic_wildcard_mention_user_ids)
        event["topic_wildcard_mention_in_followed_topic_user_ids"] = list(
            info.topic_wildcard_mention_in_followed_topic_user_ids
        )
        topic_participant_user_ids = info.topic_participant_user_ids
    else:
        event["topic_wildcard_mention_user_ids"] = []
        event["topic_wildcard_mention_in_followed_topic_user_ids"] = []
        topic_participant_user_ids = set()

    update_user_message_flags(rendering_result, ums, topic_participant_user_ids)

    do_update_mobile_push_notification(
        target_message,
        prior_mention_user_ids,
        rendering_result.mentions_user_ids,
        info.stream_push_user_ids,
    )


def apply_automatic_unmute_follow_topics_policy(
    user_profile: UserProfile,
    target_stream: Stream,
    target_topic_name: str,
) -> None:
    if (
        user_profile.automatically_follow_topics_policy
        == UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_INITIATION
    ):
        bulk_do_set_user_topic_visibility_policy(
            [user_profile],
            target_stream,
            target_topic_name,
            visibility_policy=UserTopic.VisibilityPolicy.FOLLOWED,
        )
    elif (
        user_profile.automatically_unmute_topics_in_muted_streams_policy
        == UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_INITIATION
    ):
        subscription = Subscription.objects.filter(
            recipient=target_stream.recipient,
            user_profile=user_profile,
            active=True,
            is_user_active=True,
        ).first()

        if subscription is not None and subscription.is_muted:
            bulk_do_set_user_topic_visibility_policy(
                [user_profile],
                target_stream,
                target_topic_name,
                visibility_policy=UserTopic.VisibilityPolicy.UNMUTED,
            )


# This must be called already in a transaction, with a write lock on
# the target_message.
@transaction.atomic(savepoint=False)
def do_update_message(
    user_profile: UserProfile,
    target_message: Message,
    message_edit_request: StreamMessageEditRequest | DirectMessageEditRequest,
    send_notification_to_old_thread: bool,
    send_notification_to_new_thread: bool,
    rendering_result: MessageRenderingResult | None,
    prior_mention_user_ids: set[int],
    mention_data: MentionData | None = None,
) -> UpdateMessageResult:
    """
    The main function for message editing.  A message edit event can
    modify:
    * the message's content (in which case the caller will have set
      both content and rendered_content in message_edit_request object),
    * the topic, in which case the caller will have set target_topic_name
      field with the new topic name in message_edit_request object
    * or both message's content and the topic
    * or stream and/or topic, in which case the caller will have set
      target_stream and/or target_topic_name to their new values in
      message_edit_request object.

    With topic edits, propagate_mode field in message_edit_request
    determines whether other message also have their topics edited.
    """
    timestamp = timezone_now()
    target_message.last_edit_time = timestamp

    event: dict[str, Any] = {
        "type": "update_message",
        "user_id": user_profile.id,
        "edit_timestamp": datetime_to_timestamp(timestamp),
        "message_id": target_message.id,
        "rendering_only": False,
    }

    edit_history_event: EditHistoryEvent = {
        "user_id": user_profile.id,
        "timestamp": event["edit_timestamp"],
    }

    realm = user_profile.realm
    attachment_reference_change = AttachmentChangeResult(False, [])

    ums = UserMessage.objects.filter(message=target_message.id)

    def user_info(um: UserMessage) -> dict[str, Any]:
        return {
            "id": um.user_profile_id,
            "flags": um.flags_list(),
        }

    if message_edit_request.is_content_edited:
        assert rendering_result is not None

        # mention_data is required if there's a content edit.
        assert mention_data is not None

        if isinstance(message_edit_request, StreamMessageEditRequest):
            # We do not allow changing content and stream together,
            # so we use ID of orig_stream.
            stream_topic: StreamTopicTarget | None = StreamTopicTarget(
                stream_id=message_edit_request.orig_stream.id,
                topic_name=message_edit_request.target_topic_name,
            )
        else:
            stream_topic = None

        update_message_content(
            user_profile,
            target_message,
            message_edit_request.content,
            rendering_result,
            prior_mention_user_ids,
            mention_data,
            event,
            edit_history_event,
            stream_topic,
        )

        # target_message.has_image and target_message.has_link will have been
        # already updated by Markdown rendering in the caller.
        attachment_reference_change = check_attachment_reference_change(
            target_message, rendering_result
        )
        target_message.has_attachment = attachment_reference_change.did_attachment_change

        if isinstance(message_edit_request, DirectMessageEditRequest):
            update_edit_history(target_message, timestamp, edit_history_event)

            # This does message.save(update_fields=[...])
            save_message_for_edit_use_case(message=target_message)

            event["message_ids"] = sorted(update_message_cache([target_message]))
            users_to_be_notified = list(map(user_info, ums))
            send_event_on_commit(user_profile.realm, event, users_to_be_notified)

            changed_messages_count = 1
            return UpdateMessageResult(
                changed_messages_count, attachment_reference_change.detached_attachments
            )

    assert isinstance(message_edit_request, StreamMessageEditRequest)

    stream_being_edited = message_edit_request.orig_stream
    orig_topic_name = message_edit_request.orig_topic_name

    event["stream_name"] = stream_being_edited.name
    event["stream_id"] = stream_being_edited.id

    if message_edit_request.is_message_moved:
        event["propagate_mode"] = message_edit_request.propagate_mode

    users_losing_access = UserProfile.objects.none()
    user_ids_gaining_usermessages: list[int] = []
    if message_edit_request.is_stream_edited:
        new_stream = message_edit_request.target_stream

        edit_history_event["prev_stream"] = stream_being_edited.id
        edit_history_event["stream"] = new_stream.id
        event[ORIG_TOPIC] = orig_topic_name
        assert new_stream.recipient_id is not None
        target_message.recipient_id = new_stream.recipient_id

        event["new_stream_id"] = new_stream.id
        event["propagate_mode"] = message_edit_request.propagate_mode

        # When messages are moved from one stream to another, some
        # users may lose access to those messages, including guest
        # users and users not subscribed to the new stream (if it is a
        # private stream).  For those users, their experience is as
        # though the messages were deleted, and we should send a
        # delete_message event to them instead.

        # We select _all_ current subscriptions, not just active ones,
        # for the current stream, since there may be users who were
        # previously subscribed when the message was sent, but are no
        # longer, who should also lose their UserMessage rows.
        old_stream_all_users = UserProfile.objects.filter(
            id__in=Subscription.objects.filter(
                recipient__type=Recipient.STREAM,
                recipient__type_id=stream_being_edited.id,
            ).values_list("user_profile_id")
        ).only("id")

        new_stream_current_users = UserProfile.objects.filter(
            id__in=get_active_subscriptions_for_stream_id(
                new_stream.id, include_deactivated_users=True
            ).values_list("user_profile_id")
        ).only("id")

        users_losing_usermessages = old_stream_all_users.difference(new_stream_current_users)
        if new_stream.is_public():
            # Only guest users are losing access, if it's moving to a public stream
            users_losing_access = old_stream_all_users.filter(
                role=UserProfile.ROLE_GUEST
            ).difference(new_stream_current_users)
        else:
            # If it's moving to a private stream, all non-subscribed users are losing access
            users_losing_access = users_losing_usermessages

        unmodified_user_messages = ums.exclude(user_profile__in=users_losing_usermessages)

        if not new_stream.is_history_public_to_subscribers():
            # We need to guarantee that every currently-subscribed
            # user of the new stream has a UserMessage row, since
