from collections import defaultdict
from collections.abc import Collection, Iterable, Mapping
from typing import Any, TypeAlias

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 override as override_language

from zerver.actions.default_streams import (
    do_remove_default_stream,
    do_remove_streams_from_default_stream_group,
)
from zerver.actions.message_send import maybe_send_channel_events_notice
from zerver.lib.cache import (
    cache_delete_many,
    cache_set,
    display_recipient_cache_key,
    to_dict_cache_key_id,
)
from zerver.lib.exceptions import JsonableError
from zerver.lib.mention import silent_mention_syntax_for_user, silent_mention_syntax_for_user_group
from zerver.lib.message import get_last_message_id
from zerver.lib.queue import queue_event_on_commit
from zerver.lib.stream_color import pick_colors
from zerver.lib.stream_subscription import (
    SubInfo,
    SubscriberPeerInfo,
    bulk_update_subscriber_counts,
    get_active_subscriptions_for_stream_id,
    get_bulk_stream_subscriber_info,
    get_used_colors_for_user_ids,
    get_user_ids_for_streams,
    get_users_for_streams,
)
from zerver.lib.stream_traffic import get_streams_traffic
from zerver.lib.streams import (
    can_access_stream_metadata_user_ids,
    check_basic_stream_access,
    get_anonymous_group_membership_dict_for_streams,
    get_stream_permission_policy_key,
    get_stream_post_policy_value_based_on_group_setting,
    get_user_ids_with_metadata_access_via_permission_groups,
    get_users_dict_with_metadata_access_to_streams_via_permission_groups,
    render_stream_description,
    send_stream_creation_event,
    send_stream_deletion_event,
    stream_to_dict,
)
from zerver.lib.subscription_info import bulk_get_subscriber_peer_info, get_subscribers_query
from zerver.lib.topic import get_topic_display_name
from zerver.lib.types import APISubscriptionDict, UserGroupMembersData
from zerver.lib.user_groups import (
    convert_to_user_group_members_dict,
    get_group_setting_value_for_api,
    get_group_setting_value_for_audit_log_data,
    update_or_create_user_group_for_setting,
)
from zerver.lib.users import (
    all_users_accessible_by_everyone_in_realm,
    get_subscribers_of_target_user_subscriptions,
    get_users_involved_in_dms_with_target_users,
)
from zerver.models import (
    ArchivedAttachment,
    Attachment,
    ChannelEmailAddress,
    ChannelFolder,
    DefaultStream,
    DefaultStreamGroup,
    Message,
    Realm,
    RealmAuditLog,
    Recipient,
    Stream,
    Subscription,
    UserProfile,
)
from zerver.models.groups import NamedUserGroup, UserGroup
from zerver.models.realm_audit_logs import AuditLogEventType
from zerver.models.realms import get_realm_by_id
from zerver.models.streams import StreamTopicsPolicyEnum
from zerver.models.users import active_non_guest_user_ids, active_user_ids, get_system_bot
from zerver.tornado.django_api import send_event_on_commit


def maybe_set_moderation_or_announcement_channels_none(stream: Stream) -> None:
    realm = get_realm_by_id(realm_id=stream.realm_id)
    realm_moderation_or_announcement_channels = (
        "moderation_request_channel_id",
        "new_stream_announcements_stream_id",
        "signup_announcements_stream_id",
        "zulip_update_announcements_stream_id",
    )
    update_realm_moderation_or_announcement_channels = []

    for field in realm_moderation_or_announcement_channels:
        if getattr(realm, field) == stream.id:
            setattr(realm, field, None)
            update_realm_moderation_or_announcement_channels.append(field)

    if update_realm_moderation_or_announcement_channels:
        realm.save(update_fields=update_realm_moderation_or_announcement_channels)

        event_data: dict[str, int] = {}
        for field in update_realm_moderation_or_announcement_channels:
            event_data[field] = -1

        event = dict(
            type="realm",
            op="update_dict",
            property="default",
            data=event_data,
        )
        send_event_on_commit(realm, event, active_user_ids(realm.id))


@transaction.atomic(savepoint=False)
def do_deactivate_stream(stream: Stream, *, acting_user: UserProfile | None) -> None:
    # If the stream is already deactivated, this is a no-op
    if stream.deactivated is True:
        raise JsonableError(_("Channel is already deactivated"))

    # Get the affected user ids *before* we deactivate everybody.
    affected_user_ids = can_access_stream_metadata_user_ids(stream)

    was_public = stream.is_public()
    was_web_public = stream.is_web_public
    stream.deactivated = True
    stream.save(update_fields=["deactivated"])

    ChannelEmailAddress.objects.filter(realm=stream.realm, channel=stream).update(deactivated=True)

    maybe_set_moderation_or_announcement_channels_none(stream)

    assert stream.recipient_id is not None
    if was_web_public:
        assert was_public
        # Unset the is_web_public and is_realm_public cache on attachments,
        # since the stream is now archived.
        Attachment.objects.filter(messages__recipient_id=stream.recipient_id).update(
            is_web_public=None, is_realm_public=None
        )
        ArchivedAttachment.objects.filter(messages__recipient_id=stream.recipient_id).update(
            is_web_public=None, is_realm_public=None
        )
    elif was_public:
        # Unset the is_realm_public cache on attachments, since the stream is now archived.
        Attachment.objects.filter(messages__recipient_id=stream.recipient_id).update(
            is_realm_public=None
        )
        ArchivedAttachment.objects.filter(messages__recipient_id=stream.recipient_id).update(
            is_realm_public=None
        )

    # If this is a default stream, remove it, properly sending a
    # notification to browser clients.
    if DefaultStream.objects.filter(realm_id=stream.realm_id, stream_id=stream.id).exists():
        do_remove_default_stream(stream)

    default_stream_groups_for_stream = DefaultStreamGroup.objects.filter(streams__id=stream.id)
    for group in default_stream_groups_for_stream:
        do_remove_streams_from_default_stream_group(stream.realm, group, [stream])

    event = dict(
        type="stream",
        op="update",
        stream_id=stream.id,
        name=stream.name,
        property="is_archived",
        value=True,
    )
    send_event_on_commit(stream.realm, event, affected_user_ids)

    send_stream_deletion_event(stream.realm, affected_user_ids, [stream], for_archiving=True)

    event_time = timezone_now()
    RealmAuditLog.objects.create(
        realm=stream.realm,
        acting_user=acting_user,
        modified_stream=stream,
        event_type=AuditLogEventType.CHANNEL_DEACTIVATED,
        event_time=event_time,
    )

    sender = get_system_bot(settings.NOTIFICATION_BOT, stream.realm_id)
    with override_language(stream.realm.default_language):
        maybe_send_channel_events_notice(
            sender,
            stream,
            content=_("Channel #**{channel_name}** has been archived.").format(
                channel_name=stream.name
            ),
            archived_channel_notice=True,
            limit_unread_user_ids=set(),
        )


def deactivated_streams_by_old_name(realm: Realm, stream_name: str) -> QuerySet[Stream]:
    fixed_length_prefix = ".......!DEACTIVATED:"
    truncated_name = stream_name[0 : Stream.MAX_NAME_LENGTH - len(fixed_length_prefix)]

    old_names: list[str] = [
        ("!" * bang_length + "DEACTIVATED:" + stream_name)[: Stream.MAX_NAME_LENGTH]
        for bang_length in range(1, 21)
    ]

    possible_streams = Stream.objects.filter(realm=realm, deactivated=True).filter(
        # We go looking for names as they are post-1b6f68bb59dc; 8
        # characters, followed by `!DEACTIVATED:`, followed by at
        # most MAX_NAME_LENGTH-(length of the prefix) of the name
        # they provided:
        Q(name=stream_name)
        | Q(name__regex=rf"^{fixed_length_prefix}{truncated_name}")
        # Finally, we go looking for the pre-1b6f68bb59dc version,
        # which is any number of `!` followed by `DEACTIVATED:`
        # and a prefix of the old stream name
        | Q(name__in=old_names),
    )

    return possible_streams


@transaction.atomic(savepoint=False)
def do_unarchive_stream(stream: Stream, new_name: str, *, acting_user: UserProfile | None) -> None:
    realm = stream.realm
    stream_subscribers = get_active_subscriptions_for_stream_id(
        stream.id, include_deactivated_users=True
    ).select_related("user_profile")

    if not stream.deactivated:
        raise JsonableError(_("Channel is not currently deactivated"))
    if stream.name != new_name and Stream.objects.filter(realm=realm, name=new_name).exists():
        raise JsonableError(
            _("Channel named {channel_name} already exists").format(channel_name=new_name)
        )
    if stream.invite_only and not stream_subscribers:
        raise JsonableError(_("Channel is private and have no subscribers"))
    assert stream.recipient_id is not None

    stream.deactivated = False
    stream.name = new_name
    if stream.invite_only and stream.is_web_public:
        # Previously, because archiving a channel set invite_only=True
        # without mutating is_web_public, it was possible for archived
        # channels to have this invalid state. Fix that.
        stream.is_web_public = False

    stream.save(
        update_fields=[
            "name",
            "deactivated",
            "is_web_public",
        ]
    )

    ChannelEmailAddress.objects.filter(realm=realm, channel=stream).update(deactivated=False)

    # Update caches
    cache_set(display_recipient_cache_key(stream.recipient_id), new_name, pickled_tupled=False)
    messages = Message.objects.filter(
        # Uses index: zerver_message_realm_recipient_id
        realm_id=realm.id,
        recipient_id=stream.recipient_id,
    ).only("id")
    cache_delete_many(to_dict_cache_key_id(message.id) for message in messages)

    # Unset the is_web_public and is_realm_public cache on attachments,
    # since the stream is now private.
    Attachment.objects.filter(messages__recipient_id=stream.recipient_id).update(
        is_web_public=None, is_realm_public=None
    )
    ArchivedAttachment.objects.filter(messages__recipient_id=stream.recipient_id).update(
        is_web_public=None, is_realm_public=None
    )

    RealmAuditLog.objects.create(
        realm=realm,
        acting_user=acting_user,
        modified_stream=stream,
        event_type=AuditLogEventType.CHANNEL_REACTIVATED,
        event_time=timezone_now(),
    )

    recent_traffic = get_streams_traffic(realm, {stream.id})

    notify_user_ids = list(can_access_stream_metadata_user_ids(stream))

    event = dict(
        type="stream",
        op="update",
        stream_id=stream.id,
        name=stream.name,
        property="is_archived",
        value=False,
    )
    send_event_on_commit(stream.realm, event, notify_user_ids)

    anonymous_group_membership = get_anonymous_group_membership_dict_for_streams([stream])
    send_stream_creation_event(
        realm,
        stream,
        notify_user_ids,
        recent_traffic,
        anonymous_group_membership,
        for_unarchiving=True,
    )

    sender = get_system_bot(settings.NOTIFICATION_BOT, stream.realm_id)
    with override_language(stream.realm.default_language):
        maybe_send_channel_events_notice(
            sender,
            stream,
            _("Channel #**{channel_name}** has been unarchived.").format(channel_name=new_name),
        )


def bulk_delete_cache_keys(message_ids_to_clear: list[int]) -> None:
    while len(message_ids_to_clear) > 0:
        batch = message_ids_to_clear[0:5000]

        keys_to_delete = [to_dict_cache_key_id(message_id) for message_id in batch]
        cache_delete_many(keys_to_delete)

        message_ids_to_clear = message_ids_to_clear[5000:]


def merge_streams(
    realm: Realm, stream_to_keep: Stream, stream_to_destroy: Stream
) -> tuple[int, int, int]:
    recipient_to_destroy = stream_to_destroy.recipient
    recipient_to_keep = stream_to_keep.recipient
    assert recipient_to_keep is not None
    assert recipient_to_destroy is not None
    if recipient_to_destroy.id == recipient_to_keep.id:
        return (0, 0, 0)

    # The high-level approach here is to move all the messages to
    # the surviving stream, deactivate all the subscriptions on
    # the stream to be removed and deactivate the stream, and add
    # new subscriptions to the stream to keep for any users who
    # were only on the now-deactivated stream.
    #
    # The order of operations is carefully chosen so that calling this
    # function again is likely to be an effective way to recover if
    # this process is interrupted by an error.

    # Move the Subscription objects.  This algorithm doesn't
    # preserve any stream settings/colors/etc. from the stream
    # being destroyed, but it's convenient.
    existing_subs = Subscription.objects.filter(recipient=recipient_to_keep)
    users_already_subscribed = {sub.user_profile_id: sub.active for sub in existing_subs}

    subs_to_deactivate = Subscription.objects.filter(recipient=recipient_to_destroy, active=True)
    users_to_activate = [
        sub.user_profile
        for sub in subs_to_deactivate
        if not users_already_subscribed.get(sub.user_profile_id, False)
    ]

    if len(users_to_activate) > 0:
        bulk_add_subscriptions(realm, [stream_to_keep], users_to_activate, acting_user=None)

    # Move the messages, and delete the old copies from caches. We do
    # this before removing the subscription objects, to avoid messages
    # "disappearing" if an error interrupts this function.
    message_ids_to_clear = list(
        Message.objects.filter(
            # Uses index: zerver_message_realm_recipient_id
            realm_id=realm.id,
            recipient=recipient_to_destroy,
        ).values_list("id", flat=True)
    )
    count = Message.objects.filter(
        # Uses index: zerver_message_realm_recipient_id (prefix)
        realm_id=realm.id,
        recipient=recipient_to_destroy,
    ).update(recipient=recipient_to_keep)
    bulk_delete_cache_keys(message_ids_to_clear)

    # Remove subscriptions to the old stream.
    if len(subs_to_deactivate) > 0:
        bulk_remove_subscriptions(
            realm,
            [sub.user_profile for sub in subs_to_deactivate],
            [stream_to_destroy],
            acting_user=None,
        )

    do_deactivate_stream(stream_to_destroy, acting_user=None)

    return (len(users_to_activate), count, len(subs_to_deactivate))


def get_subscriber_ids(
    stream: Stream, requesting_user: UserProfile | None = None
) -> QuerySet[Subscription, int]:
    subscriptions_query = get_subscribers_query(stream, requesting_user)
    return subscriptions_query.values_list("user_profile_id", flat=True)


def send_subscription_add_events(
    realm: Realm,
    sub_info_list: list[SubInfo],
    subscriber_dict: dict[int, set[int]],
) -> None:
    info_by_user: dict[int, list[SubInfo]] = defaultdict(list)
    for sub_info in sub_info_list:
        info_by_user[sub_info.user.id].append(sub_info)

    stream_ids = {sub_info.stream.id for sub_info in sub_info_list}
    recent_traffic = get_streams_traffic(realm=realm, stream_ids=stream_ids)

    # We generally only have a few streams, so we compute subscriber
    # data in its own loop.
    stream_subscribers_dict: dict[int, list[int]] = {}
    for sub_info in sub_info_list:
        stream = sub_info.stream
        if stream.id not in stream_subscribers_dict:
            subscribers = list(subscriber_dict[stream.id])
            stream_subscribers_dict[stream.id] = subscribers

    streams = [sub_info.stream for sub_info in sub_info_list]
    anonymous_group_membership = get_anonymous_group_membership_dict_for_streams(streams)

    for user_id, sub_infos in info_by_user.items():
        sub_dicts: list[APISubscriptionDict] = []
        for sub_info in sub_infos:
            stream = sub_info.stream
            stream_subscribers = stream_subscribers_dict[stream.id]
            subscription = sub_info.sub
            stream_dict = stream_to_dict(stream, recent_traffic, anonymous_group_membership)
            # This is verbose as we cannot unpack existing TypedDict
            # to initialize another TypedDict while making mypy happy.
            # https://github.com/python/mypy/issues/5382
            sub_dict = APISubscriptionDict(
                # Fields from Subscription.API_FIELDS
                audible_notifications=subscription.audible_notifications,
                color=subscription.color,
                desktop_notifications=subscription.desktop_notifications,
                email_notifications=subscription.email_notifications,
                is_muted=subscription.is_muted,
                pin_to_top=subscription.pin_to_top,
                push_notifications=subscription.push_notifications,
                wildcard_mentions_notify=subscription.wildcard_mentions_notify,
                # Computed fields not present in Subscription.API_FIELDS
                in_home_view=not subscription.is_muted,
                stream_weekly_traffic=stream_dict["stream_weekly_traffic"],
                subscribers=stream_subscribers,
                # Fields from Stream.API_FIELDS
                is_archived=stream_dict["is_archived"],
                can_add_subscribers_group=stream_dict["can_add_subscribers_group"],
                can_administer_channel_group=stream_dict["can_administer_channel_group"],
                can_create_topic_group=stream_dict["can_create_topic_group"],
                can_delete_any_message_group=stream_dict["can_delete_any_message_group"],
                can_delete_own_message_group=stream_dict["can_delete_own_message_group"],
                can_move_messages_out_of_channel_group=stream_dict[
                    "can_move_messages_out_of_channel_group"
                ],
                can_move_messages_within_channel_group=stream_dict[
                    "can_move_messages_within_channel_group"
                ],
                can_send_message_group=stream_dict["can_send_message_group"],
                can_remove_subscribers_group=stream_dict["can_remove_subscribers_group"],
                can_resolve_topics_group=stream_dict["can_resolve_topics_group"],
                can_subscribe_group=stream_dict["can_subscribe_group"],
                creator_id=stream_dict["creator_id"],
                date_created=stream_dict["date_created"],
                description=stream_dict["description"],
                first_message_id=stream_dict["first_message_id"],
                folder_id=stream_dict["folder_id"],
                is_recently_active=stream_dict["is_recently_active"],
                history_public_to_subscribers=stream_dict["history_public_to_subscribers"],
                invite_only=stream_dict["invite_only"],
                is_web_public=stream_dict["is_web_public"],
                message_retention_days=stream_dict["message_retention_days"],
                name=stream_dict["name"],
                rendered_description=stream_dict["rendered_description"],
                stream_id=stream_dict["stream_id"],
                stream_post_policy=stream_dict["stream_post_policy"],
                subscriber_count=stream_dict["subscriber_count"],
                topics_policy=stream_dict["topics_policy"],
                # Computed fields not present in Stream.API_FIELDS
                is_announcement_only=stream_dict["is_announcement_only"],
            )

            sub_dicts.append(sub_dict)

        # Send a notification to the user who subscribed.
        event = dict(type="subscription", op="add", subscriptions=sub_dicts)
        send_event_on_commit(realm, event, [user_id])


# This function contains all the database changes as part of
# subscribing users to streams; the transaction ensures that the
# RealmAuditLog entries are created atomically with the Subscription
# object creation (and updates).
@transaction.atomic(savepoint=False)
def bulk_add_subs_to_db_with_logging(
    realm: Realm,
    acting_user: UserProfile | None,
    subs_to_add: list[SubInfo],
    subs_to_activate: list[SubInfo],
) -> None:
    Subscription.objects.bulk_create(info.sub for info in subs_to_add)
    sub_ids = [info.sub.id for info in subs_to_activate]
    Subscription.objects.filter(id__in=sub_ids).update(active=True)

    # Log subscription activities in RealmAuditLog
    event_time = timezone_now()
    event_last_message_id = get_last_message_id()

    all_subscription_logs = [
        RealmAuditLog(
            realm=realm,
            acting_user=acting_user,
            modified_user=sub_info.user,
            modified_stream=sub_info.stream,
            event_last_message_id=event_last_message_id,
            event_type=event_type,
            event_time=event_time,
        )
        for event_type, subs in [
            (AuditLogEventType.SUBSCRIPTION_CREATED, subs_to_add),
            (AuditLogEventType.SUBSCRIPTION_ACTIVATED, subs_to_activate),
        ]
        for sub_info in subs
    ]
    # Now since we have all log objects generated we can do a bulk insert
    RealmAuditLog.objects.bulk_create(all_subscription_logs)


def send_stream_creation_events_for_previously_inaccessible_streams(
    realm: Realm,
    stream_dict: dict[int, Stream],
    altered_user_dict: dict[int, set[int]],
    altered_guests: set[int],
    users_with_metadata_access_via_permission_groups: dict[int, set[int]] | None = None,
) -> None:
    stream_ids = set(altered_user_dict.keys())
    recent_traffic = get_streams_traffic(realm, stream_ids)

    streams = [stream_dict[stream_id] for stream_id in stream_ids]
    anonymous_group_membership: dict[int, UserGroupMembersData] | None = None

    for stream_id, stream_users_ids in altered_user_dict.items():
        stream = stream_dict[stream_id]

        notify_user_ids = []
        if not stream.is_public():
            assert users_with_metadata_access_via_permission_groups is not None
            # Users newly added to invite-only streams
            # need a `create` notification.  The former, because
            # they need the stream to exist before
            # they get the "subscribe" notification, and the latter so
            # they can manage the new stream.
            # Realm admins already have all created private streams.
            realm_admin_ids = {user.id for user in realm.get_admin_users_and_bots()}
            notify_user_ids = list(
                stream_users_ids
                - realm_admin_ids
                - users_with_metadata_access_via_permission_groups[stream.id]
            )
        elif not stream.is_web_public:
            # Guese users need a `create` notification for
            # public streams as well because they need the stream
            # to exist before they get the "subscribe" notification.
            notify_user_ids = list(stream_users_ids & altered_guests)

        if notify_user_ids:
            if anonymous_group_membership is None:
                anonymous_group_membership = get_anonymous_group_membership_dict_for_streams(
                    streams
                )

            send_stream_creation_event(
                realm, stream, notify_user_ids, recent_traffic, anonymous_group_membership
            )


def send_peer_subscriber_events(
    op: str,
    realm: Realm,
    stream_dict: dict[int, Stream],
    altered_user_dict: dict[int, set[int]],
    subscriber_peer_info: SubscriberPeerInfo,
) -> None:
    # Send peer_add/peer_remove events to other users who are tracking the
    # subscribers lists of streams in their browser; everyone for
    # public streams and only existing subscribers for private streams.

    assert op in ["peer_add", "peer_remove"]

    private_stream_ids = [
        stream_id for stream_id in altered_user_dict if stream_dict[stream_id].invite_only
    ]
    public_stream_ids = [
        stream_id for stream_id in altered_user_dict if not stream_dict[stream_id].invite_only
    ]
    web_public_stream_ids = [
        stream_id for stream_id in public_stream_ids if stream_dict[stream_id].is_web_public
    ]

    private_peer_dict = subscriber_peer_info.private_peer_dict
    for stream_id in private_stream_ids:
        altered_user_ids = altered_user_dict[stream_id]
        peer_user_ids = private_peer_dict[stream_id] - altered_user_ids

        if peer_user_ids and altered_user_ids:
            event = dict(
                type="subscription",
                op=op,
                stream_ids=[stream_id],
                user_ids=sorted(altered_user_ids),
            )
            send_event_on_commit(realm, event, peer_user_ids)

    if public_stream_ids:
        subscriber_dict = subscriber_peer_info.subscribed_ids
        user_streams: dict[str, set[int]] = defaultdict(set)
        non_guest_user_ids = set(active_non_guest_user_ids(realm.id))

        for stream_id in public_stream_ids:
            altered_user_ids = altered_user_dict[stream_id]
            if altered_user_ids:
                altered_user_id_string = ",".join(map(str, sorted(altered_user_ids)))
                # We will store stream_ids related to each set of
                # unique user ids.
                user_streams[altered_user_id_string].add(stream_id)

        # For each set of unique user ids, we will send one event for
        # web public streams, one event for public streams without a
        # guest user and for each stream apart from that, we will send
        # one event each.
        for altered_user_id_string, stream_ids in user_streams.items():
            altered_user_ids = set(map(int, altered_user_id_string.split(",")))
            # Both of the lists below will have the same list of peer
            # user ids each, so we will send a single event for each of
            # these lists.
            web_public_user_stream_ids = []
            public_user_stream_ids_without_guest_users = []

            for stream_id in stream_ids:
                if stream_id in web_public_stream_ids:
                    web_public_user_stream_ids.append(stream_id)
                else:
                    if (non_guest_user_ids | subscriber_dict[stream_id]) == non_guest_user_ids:
                        public_user_stream_ids_without_guest_users.append(stream_id)
                    else:
                        # This channel likely has a unique set of
                        # peer_user_ids to notify, so we send the event
                        # directly for just this channel.
                        peer_user_ids = (
                            non_guest_user_ids | subscriber_dict[stream_id]
                        ) - altered_user_ids
                        event = dict(
                            type="subscription",
                            op=op,
                            stream_ids=[stream_id],
                            user_ids=list(altered_user_ids),
                        )
                        send_event_on_commit(realm, event, peer_user_ids)

            if len(web_public_user_stream_ids) > 0:
                web_public_peer_ids = set(active_user_ids(realm.id))
                web_public_streams_event = dict(
                    type="subscription",
                    op=op,
                    stream_ids=sorted(web_public_user_stream_ids),
                    user_ids=list(altered_user_ids),
                )
                send_event_on_commit(
                    realm, web_public_streams_event, web_public_peer_ids - altered_user_ids
                )
            if len(public_user_stream_ids_without_guest_users) > 0:
                public_streams_event = dict(
                    type="subscription",
                    op=op,
                    stream_ids=sorted(public_user_stream_ids_without_guest_users),
                    user_ids=list(altered_user_ids),
                )
                send_event_on_commit(
                    realm, public_streams_event, non_guest_user_ids - altered_user_ids
                )


def send_user_creation_events_on_adding_subscriptions(
    realm: Realm,
    altered_user_dict: dict[int, set[int]],
    altered_streams_dict: dict[UserProfile, set[int]],
    subscribers_of_altered_user_subscriptions: dict[int, set[int]],
) -> None:
    altered_users = list(altered_streams_dict.keys())
    non_guest_user_ids = active_non_guest_user_ids(realm.id)

    users_involved_in_dms = get_users_involved_in_dms_with_target_users(altered_users, realm)

    altered_stream_ids = altered_user_dict.keys()
    subscribers_dict = get_users_for_streams(set(altered_stream_ids))

    subscribers_user_id_map: dict[int, UserProfile] = {}
    subscriber_ids_dict: dict[int, set[int]] = defaultdict(set)
    for stream_id, subscribers in subscribers_dict.items():
        for user in subscribers:
            subscriber_ids_dict[stream_id].add(user.id)
            subscribers_user_id_map[user.id] = user

    from zerver.actions.create_user import notify_created_user

    for user in altered_users:
        streams_for_user = altered_streams_dict[user]
        subscribers_in_altered_streams: set[int] = set()
        for stream_id in streams_for_user:
            subscribers_in_altered_streams |= subscriber_ids_dict[stream_id]

        users_already_with_access_to_altered_user = (
            set(non_guest_user_ids)
            | subscribers_of_altered_user_subscriptions[user.id]
            | users_involved_in_dms[user.id]
            | {user.id}
        )

        users_to_receive_creation_event = (
            subscribers_in_altered_streams - users_already_with_access_to_altered_user
        )
        if users_to_receive_creation_event:
            notify_created_user(user, list(users_to_receive_creation_event))

        if user.is_guest:
            # If the altered user is a guest, then the user may receive
            # user creation events for subscribers of the new stream.
            users_already_accessible_to_altered_user = (
                subscribers_of_altered_user_subscriptions[user.id]
                | users_involved_in_dms[user.id]
                | {user.id}
            )

            new_accessible_user_ids = (
                subscribers_in_altered_streams - users_already_accessible_to_altered_user
            )
            for accessible_user_id in new_accessible_user_ids:
                accessible_user = subscribers_user_id_map[accessible_user_id]
                notify_created_user(accessible_user, [user.id])


SubT: TypeAlias = tuple[list[SubInfo], list[SubInfo]]


@transaction.atomic(savepoint=False)
def bulk_add_subscriptions(
    realm: Realm,
    streams: Collection[Stream],
    users: Iterable[UserProfile],
    color_map: Mapping[str, str] = {},
    from_user_creation: bool = False,
    *,
    acting_user: UserProfile | None,
) -> SubT:
    users = list(users)
    user_ids = [user.id for user in users]

    # Sanity check out callers
    for stream in streams:
        assert stream.realm_id == realm.id

    for user in users:
        assert user.realm_id == realm.id

    recipient_ids = [stream.recipient_id for stream in streams]
    recipient_id_to_stream = {stream.recipient_id: stream for stream in streams}

    recipient_color_map = {}
    recipient_ids_set: set[int] = set()
    for stream in streams:
        assert stream.recipient_id is not None
        recipient_ids_set.add(stream.recipient_id)
        color: str | None = color_map.get(stream.name, None)
        if color is not None:
            recipient_color_map[stream.recipient_id] = color

    used_colors_for_user_ids: dict[int, set[str]] = get_used_colors_for_user_ids(user_ids)

    existing_subs = Subscription.objects.filter(
        user_profile_id__in=user_ids,
        recipient__type=Recipient.STREAM,
        recipient_id__in=recipient_ids,
    )

    subs_by_user: dict[int, list[Subscription]] = defaultdict(list)
    for sub in existing_subs:
        subs_by_user[sub.user_profile_id].append(sub)

    already_subscribed: list[SubInfo] = []
    subs_to_activate: list[SubInfo] = []
    subs_to_add: list[SubInfo] = []
    for user_profile in users:
        my_subs = subs_by_user[user_profile.id]

        # Make a fresh set of all new recipient ids, and then we will
        # remove any for which our user already has a subscription
